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.

    Example:

    q, err := query.New("account.name='John'")
    if err != nil {
        return err
    }
    ctx, cancel := context.WithTimeout(context.Background(), 1 * time.Second)
    defer cancel()
    subscription, err := pubsub.Subscribe(ctx, "johns-transactions", q)
    if err != nil {
        return err
    }
    
    for {
        select {
        case msg <- subscription.Out():
            // handle msg.Data() and msg.Tags()
        case <-subscription.Cancelled():
            return subscription.Err()
        }
    }
    

    Index

    Constants

    This section is empty.

    Variables

    View Source
    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")
    )
    View Source
    var (
    	// ErrUnsubscribed is returned by Err when a client unsubscribes.
    	ErrUnsubscribed = errors.New("client unsubscribed")
    
    	// ErrOutOfCapacity is returned by Err when a client is not pulling messages
    	// fast enough. Note the client's subscription will be terminated.
    	ErrOutOfCapacity = errors.New("client is not pulling messages fast enough")
    )

    Functions

    This section is empty.

    Types

    type Message

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

      Message glues data and tags together.

      func NewMessage

      func NewMessage(data interface{}, tags map[string]string) Message

      func (Message) Data

      func (msg Message) Data() interface{}

        Data returns an original data published.

        func (Message) Tags

        func (msg Message) Tags() map[string]string

          Tags returns tags, which matched the client's query.

          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 map[string]string) 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) NumClientSubscriptions

                      func (s *Server) NumClientSubscriptions(clientID string) int

                        NumClientSubscriptions returns the number of subscriptions the client has.

                        func (*Server) NumClients

                        func (s *Server) NumClients() int

                          NumClients returns the number of clients.

                          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 map[string]string) 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, outCapacity ...int) (*Subscription, error)

                                      Subscribe creates a subscription for the given client.

                                      An error will be returned to the caller if the context is canceled or if subscription already exist for pair clientID and query.

                                      outCapacity can be used to set a capacity for Subscription#Out channel (1 by default). Panics if outCapacity is less than or equal to zero. If you want an unbuffered channel, use SubscribeUnbuffered.

                                      func (*Server) SubscribeUnbuffered

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

                                        SubscribeUnbuffered does the same as Subscribe, except it returns a subscription with unbuffered channel. Use with caution as it can freeze the server.

                                        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 Subscription

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

                                              A Subscription represents a client subscription for a particular query and consists of three things: 1) channel onto which messages and tags are published 2) channel which is closed if a client is too slow or choose to unsubscribe 3) err indicating the reason for (2)

                                              func NewSubscription

                                              func NewSubscription(outCapacity int) *Subscription

                                                NewSubscription returns a new subscription with the given outCapacity.

                                                func (*Subscription) Cancelled

                                                func (s *Subscription) Cancelled() <-chan struct{}

                                                  Cancelled returns a channel that's closed when the subscription is terminated and supposed to be used in a select statement.

                                                  func (*Subscription) Err

                                                  func (s *Subscription) Err() error

                                                    Err returns nil if the channel returned by Cancelled is not yet closed. If the channel is closed, Err returns a non-nil error explaining why:

                                                    - ErrUnsubscribed if the subscriber choose to unsubscribe,
                                                    - ErrOutOfCapacity if the subscriber is not pulling messages fast enough
                                                    and the channel returned by Out became full,
                                                    

                                                    After Err returns a non-nil error, successive calls to Err return the same error.

                                                    func (*Subscription) Out

                                                    func (s *Subscription) Out() <-chan Message

                                                      Out returns a channel onto which messages and tags are published. Unsubscribe/UnsubscribeAll does not close the channel to avoid clients from receiving a nil message.

                                                      Directories

                                                      Path Synopsis
                                                      Package query provides a parser for a custom query format: abci.invoice.number=22 AND abci.invoice.owner=Ivan See query.peg for the grammar, which is a https://en.wikipedia.org/wiki/Parsing_expression_grammar.
                                                      Package query provides a parser for a custom query format: abci.invoice.number=22 AND abci.invoice.owner=Ivan See query.peg for the grammar, which is a https://en.wikipedia.org/wiki/Parsing_expression_grammar.