pubsub

Documentation

Overview

Package pubsub implements a pub-sub model with a single publisher (Server) and multiple subscribers (clients).

Though you can have multiple publishers by sharing a pointer to a server or by giving the same channel to each publisher and publishing messages from that channel (fan-in).

Clients subscribe for messages, which could be of any type, using a query. When some message is published, we match it with all queries. If there is a match, this message will be pushed to all clients, subscribed to that query. See query subpackage for our implementation.

Due to the blocking send implementation, a single subscriber can freeze an entire server by not reading messages before it unsubscribes. To avoid such scenario, subscribers must either:

a) make sure they continue to read from the out channel until Unsubscribe(All) is called

s.Subscribe(ctx, sub, qry, out)
go func() {
    for msg := range out {
        // handle msg
        // will exit automatically when out is closed by Unsubscribe(All)
    }
}()
s.UnsubscribeAll(ctx, sub)

b) drain the out channel before calling Unsubscribe(All)

    s.Subscribe(ctx, sub, qry, out)
    defer func() {
        // drain out to make sure we don't block
    LOOP:
		     for {
		     	   select {
		     	   case <-out:
		     	   default:
		     	   	   break LOOP
		     	   }
		     }
        s.UnsubscribeAll(ctx, sub)
    }()
    for msg := range out {
        // handle msg
        if err != nil {
           return err
        }
    }

Index

• Variables
• type Option
  • func BufferCapacity(cap int) Option
• type Query
• type Server
  • func NewServer(options ...Option) *Server
  • func (s *Server) BufferCapacity() int
  • func (s *Server) OnReset() error
  • func (s *Server) OnStart() error
  • func (s *Server) OnStop()
  • func (s *Server) Publish(ctx context.Context, msg interface{}) error
  • func (s *Server) PublishWithTags(ctx context.Context, msg interface{}, tags TagMap) error
  • func (s *Server) Subscribe(ctx context.Context, clientID string, query Query, out chan<- interface{}) error
  • func (s *Server) Unsubscribe(ctx context.Context, clientID string, query Query) error
  • func (s *Server) UnsubscribeAll(ctx context.Context, clientID string) error
• type TagMap
  • func NewTagMap(data map[string]string) TagMap

Variables

var (
	// ErrSubscriptionNotFound is returned when a client tries to unsubscribe
	// from not existing subscription.
	ErrSubscriptionNotFound = errors.New("subscription not found")

	// ErrAlreadySubscribed is returned when a client tries to subscribe twice or
	// more using the same query.
	ErrAlreadySubscribed = errors.New("already subscribed")
)

Types

type Option

type Option func(*Server)

Option sets a parameter for the server.

func BufferCapacity

func BufferCapacity(cap int) Option

BufferCapacity allows you to specify capacity for the internal server's queue. Since the server, given Y subscribers, could only process X messages, this option could be used to survive spikes (e.g. high amount of transactions during peak hours).

type Query

type Query interface {
	Matches(tags TagMap) bool
	String() string
}

Query defines an interface for a query to be used for subscribing.

type Server

type Server struct {
	cmn.BaseService
	// contains filtered or unexported fields
}

Server allows clients to subscribe/unsubscribe for messages, publishing messages with or without tags, and manages internal state.

func NewServer

func NewServer(options ...Option) *Server

NewServer returns a new server. See the commentary on the Option functions for a detailed description of how to configure buffering. If no options are provided, the resulting server's queue is unbuffered.

func (*Server) BufferCapacity

func (s *Server) BufferCapacity() int

BufferCapacity returns capacity of the internal server's queue.

func (*Server) OnReset

func (s *Server) OnReset() error

OnReset implements Service.OnReset

func (*Server) OnStart

func (s *Server) OnStart() error

OnStart implements Service.OnStart by starting the server.

func (*Server) OnStop

func (s *Server) OnStop()

OnStop implements Service.OnStop by shutting down the server.

func (*Server) Publish

func (s *Server) Publish(ctx context.Context, msg interface{}) error

Publish publishes the given message. An error will be returned to the caller if the context is canceled.

func (*Server) PublishWithTags

func (s *Server) PublishWithTags(ctx context.Context, msg interface{}, tags TagMap) error

PublishWithTags publishes the given message with the set of tags. The set is matched with clients queries. If there is a match, the message is sent to the client.

func (*Server) Subscribe

func (s *Server) Subscribe(ctx context.Context, clientID string, query Query, out chan<- interface{}) error

Subscribe creates a subscription for the given client. It accepts a channel on which messages matching the given query can be received. An error will be returned to the caller if the context is canceled or if subscription already exist for pair clientID and query.

func (*Server) Unsubscribe

func (s *Server) Unsubscribe(ctx context.Context, clientID string, query Query) error

Unsubscribe removes the subscription on the given query. An error will be returned to the caller if the context is canceled or if subscription does not exist.

func (*Server) UnsubscribeAll

func (s *Server) UnsubscribeAll(ctx context.Context, clientID string) error

UnsubscribeAll removes all client subscriptions. An error will be returned to the caller if the context is canceled or if subscription does not exist.

type TagMap

type TagMap interface {
	// Get returns the value for a key, or nil if no value is present.
	// The ok result indicates whether value was found in the tags.
	Get(key string) (value string, ok bool)
	// Len returns the number of tags.
	Len() int
}

TagMap is used to associate tags to a message. They can be queried by subscribers to choose messages they will received.

func NewTagMap

func NewTagMap(data map[string]string) TagMap

NewTagMap constructs a new immutable tag set from a map.