redispb

Documentation

Overview

Package redis implements pubsub communication over redis pubsub system for use in the actor framework.

Index

• Variables
• type Config
• type PubSubFactoryGenerator
  • func PubSubFactory(publishers PublisherHandler, subscribers SubscriberHandler) PubSubFactoryGenerator
• type Publisher
  • func NewPublisher(ctx context.Context, cfg *Config, topic string, sink *redis.Client, ...) *Publisher
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(msg actorkit.Envelope) error
• type PublisherHandler
• type PublisherSubscriberFactory
  • func NewPublisherSubscriberFactory(ctx context.Context, config Config) (*PublisherSubscriberFactory, error)
  • func (pf *PublisherSubscriberFactory) Close() error
  • func (pf *PublisherSubscriberFactory) Publisher(topic string) (*Publisher, error)
  • func (pf *PublisherSubscriberFactory) Subscribe(topic string, id string, receiver pubsubs.Receiver) (*Subscription, error)
  • func (pf *PublisherSubscriberFactory) Wait()
• type SubscriberHandler
• type Subscription
  • func (s *Subscription) Error() error
  • func (s *Subscription) Group() string
  • func (s *Subscription) ID() string
  • func (s *Subscription) Stop() error
  • func (s *Subscription) Topic() string

Variables

var ErrBusyPublisher = errors.New("publisher busy, try again")

ErrBusyPublisher is returned when publisher fails to send a giving message.

Types

type Config

type Config struct {
	ProjectID   string
	Log         actorkit.Logs
	Host        *redis.Options
	Marshaler   pubsubs.Marshaler
	Unmarshaler pubsubs.Unmarshaler

	// MessageDeliveryTimeout is the timeout to wait before response
	// from the underline message broker before timeout.
	MessageDeliveryTimeout time.Duration
}

Config provides a config struct for instantiating a Publisher type.

type PubSubFactoryGenerator

type PubSubFactoryGenerator func(factory *PublisherSubscriberFactory) pubsubs.PubSubFactory

PubSubFactoryGenerator returns a function which taken a PublisherSubscriberFactory returning a factory for generating publishers and subscribers.

func PubSubFactory

func PubSubFactory(publishers PublisherHandler, subscribers SubscriberHandler) PubSubFactoryGenerator

PubSubFactory provides a partial function for the generation of a pubsubs.PubSubFactory using the PubSubFactorGenerator function.

type Publisher

type Publisher struct {
	// contains filtered or unexported fields
}

Publisher implements the topic publishing provider for the google redis layer.

func NewPublisher

func NewPublisher(ctx context.Context, cfg *Config, topic string, sink *redis.Client, marshaler pubsubs.Marshaler) *Publisher

NewPublisher returns a new instance of a Publisher.

func (*Publisher) Close

func (p *Publisher) Close() error

Close closes giving subscriber.

func (*Publisher) Publish

func (p *Publisher) Publish(msg actorkit.Envelope) error

Publish attempts to publish giving message into provided topic publisher returning an error for failed attempt.

type PublisherHandler

type PublisherHandler func(*PublisherSubscriberFactory, string) (pubsubs.Publisher, error)

PublisherHandler defines a function type which takes a giving PublisherFactory and a given topic, returning a new publisher with all related underline specific details added and instantiated.

type PublisherSubscriberFactory

type PublisherSubscriberFactory struct {
	// contains filtered or unexported fields
}

PublisherSubscriberFactory implements a Google redis Publisher factory which handles creation of publishers for topic publishing and management.

func NewPublisherSubscriberFactory

func NewPublisherSubscriberFactory(ctx context.Context, config Config) (*PublisherSubscriberFactory, error)

NewPublisherSubscriberFactory returns a new instance of publisher factory.

func (*PublisherSubscriberFactory) Close

func (pf *PublisherSubscriberFactory) Close() error

Close closes giving publisher factory and all previous created publishers.

func (*PublisherSubscriberFactory) Publisher

func (pf *PublisherSubscriberFactory) Publisher(topic string) (*Publisher, error)

Publisher returns giving publisher for giving topic, if provided config allows the creation of publisher if not present then a new publisher is created for topic and returned, else an error is returned if not found or due to some other issues.

func (*PublisherSubscriberFactory) Subscribe

func (pf *PublisherSubscriberFactory) Subscribe(topic string, id string, receiver pubsubs.Receiver) (*Subscription, error)

Subscribe returns a new subscription for a giving topic which will be used for processing messages for giving topic from the NATS streaming provider. If the topic already has a subscriber then a subscriber with a ever increasing _id is added and returned, the subscriber receives the giving topic_id as durable name for it's subscription.

func (*PublisherSubscriberFactory) Wait

func (pf *PublisherSubscriberFactory) Wait()

Wait blocks till all generated publishers close and have being reclaimed.

type SubscriberHandler

type SubscriberHandler func(*PublisherSubscriberFactory, string, string, pubsubs.Receiver) (pubsubs.Subscription, error)

SubscriberHandler defines a function type which takes a giving SubscriptionFactory and a given topic, returning a new subscription with all related underline specific details added and instantiated.

type Subscription

type Subscription struct {
	// contains filtered or unexported fields
}

Subscription implements a subscriber of a giving topic which is being subscribe to for. It implements the pubsubs.Subscription interface.

func (*Subscription) Error

func (s *Subscription) Error() error

Error returns the associated received error.

func (*Subscription) Group

func (s *Subscription) Group() string

Group returns the group or queue group name of giving subscription.

func (*Subscription) ID

func (s *Subscription) ID() string

ID returns the identification of giving subscription used for durability if supported.

func (*Subscription) Stop

func (s *Subscription) Stop() error

Stop ends giving subscription and it's operation in listening to given topic.

func (*Subscription) Topic

func (s *Subscription) Topic() string

Topic returns the topic name of giving subscription.