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.
Index ¶
- type Option
- type Query
- type Server
- func (s Server) BufferCapacity() int
- 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 map[string]interface{}) 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
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*Server)
Option sets a parameter for the server.
func BufferCapacity ¶
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 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 ¶
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 ¶
BufferCapacity returns capacity of the internal server's queue.
func (*Server) OnStop ¶
func (s *Server) OnStop()
OnStop implements Service.OnStop by shutting down the server.
func (*Server) Publish ¶
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]interface{}) 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. If the subscription already exists, the old channel will be closed. An error will be returned to the caller if the context is canceled.
func (*Server) Unsubscribe ¶
Unsubscribe removes the subscription on the given query. An error will be returned to the caller if the context is canceled.
Source Files
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. |