Documentation

Overview

    This package was extracted from Tendermint

    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.

    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")
    )

    Functions

    This section is empty.

    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).

        func WithLogger

        func WithLogger(logger *logging.Logger) Option

        type Server

        type Server struct {
        	service.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 query.Tagged) 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, qry query.Query, outBuffer int) (<-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, qry 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.

                              Source Files