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.

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.Events()
    case <-subscription.Canceled():
        return subscription.Err()
    }
}

Package pubsub may be internalized (made private) in future releases. XXX Deprecated.

Index

• Variables
• type Message
  • func NewMessage(data any, events map[string][]string) Message
  • func (msg Message) Data() any
  • func (msg Message) Events() map[string][]string
• type Option
  • func BufferCapacity(cap int) Option
• type Query
• type Server
  • func NewServer(options ...Option) *Server
  • func (s *Server) BufferCapacity() int
  • func (s *Server) NumClientSubscriptions(clientID string) int
  • func (s *Server) NumClients() int
  • func (*Server) OnReset() error
  • func (s *Server) OnStart() error
  • func (s *Server) OnStop()
  • func (s *Server) Publish(ctx context.Context, msg any) error
  • func (s *Server) PublishWithEvents(ctx context.Context, msg any, events map[string][]string) error
  • func (s *Server) Subscribe(ctx context.Context, clientID string, query Query, outCapacity ...int) (*Subscription, error)
  • func (s *Server) SubscribeUnbuffered(ctx context.Context, clientID string, query Query) (*Subscription, error)
  • func (s *Server) Unsubscribe(ctx context.Context, clientID string, query Query) error
  • func (s *Server) UnsubscribeAll(ctx context.Context, clientID string) error
• type Subscription
  • func NewSubscription(outCapacity int) *Subscription
  • func (s *Subscription) Canceled() <-chan struct{}
  • func (s *Subscription) Err() error
  • func (s *Subscription) Out() <-chan Message

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

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("internal subscription event buffer is out of capacity")
)

Types

type Message

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

Message glues data and events together.

func NewMessage

func NewMessage(data any, events map[string][]string) Message

func (Message) Data

func (msg Message) Data() any

Data returns an original data published.

func (Message) Events

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

Events returns events, 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(events map[string][]string) (bool, error)
	String() string
}

Query defines an interface for a query to be used for subscribing. A query matches against a map of events. Each key in this map is a composite of the even type and an attribute key (e.g. "{eventType}.{eventAttrKey}") and the values are the event values that are contained under that relationship. This allows event types to repeat themselves with the same set of keys and different values.

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 events, 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 (*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 any) error

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

func (*Server) PublishWithEvents

func (s *Server) PublishWithEvents(ctx context.Context, msg any, events map[string][]string) error

PublishWithEvents publishes the given message with the set of events. 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 events 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) Canceled

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

Canceled 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 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 events are published. Unsubscribe/UnsubscribeAll does not close the channel to avoid clients from receiving a nil message.