ps

README

ps

General purpose pub/sub for Go.

Create a Broker. Callers can Publish values to the broker, and clients can Subscribe with a channel that receives all published values which pass the provided allow func.

Publishing is best-effort; if a subscriber is slow or non-responsive, published values to that subscriber are dropped.

package pshttp provides an HTTP interface over a pub/sub broker.

Documentation

Overview

Package ps provides a general-purpose pub/sub broker.

Index

• Variables
• type Broker
  • func NewBroker[T any]() *Broker[T]
  • func (b *Broker[T]) ActiveSubscribers() []Stats
  • func (b *Broker[T]) Publish(v T) Stats
  • func (b *Broker[T]) Stats(c chan<- T) (Stats, error)
  • func (b *Broker[T]) Subscribe(c chan<- T, allow func(T) bool) error
  • func (b *Broker[T]) SubscribeAll(c chan<- T) error
  • func (b *Broker[T]) Unsubscribe(c chan<- T) (Stats, error)
• type Stats
  • func (s Stats) String() string
  • func (s Stats) Total() uint64

Variables

var (
	// ErrAlreadySubscribed signals that a given subscription already exists.
	ErrAlreadySubscribed = errors.New("already subscribed")

	// ErrNotSubscribed indicates that a given subscription doesn't exist.
	ErrNotSubscribed = errors.New("not subscribed")
)

Types

type Broker

type Broker[T any] struct {
	// contains filtered or unexported fields
}

Broker is a pub/sub coördination point for values of type T. See the Publish, Subscribe, and Unsubscribe methods for more information.

func NewBroker

func NewBroker[T any]() *Broker[T]

NewBroker returns a new broker for values of type T.

func (*Broker[T]) ActiveSubscribers

func (b *Broker[T]) ActiveSubscribers() []Stats

ActiveSubscribers returns statistics for every active subscriber.

func (*Broker[T]) Publish

func (b *Broker[T]) Publish(v T) Stats

Publish the given value to all active and matching subscribers. Each send is non-blocking, so values are dropped when subscribers aren't keeping up. Also, values are sent directly, so be mindful of copy costs and semantics. Returned stats reflect the outcome for all active subscribers at the time of the publish.

func (*Broker[T]) Stats

func (b *Broker[T]) Stats(c chan<- T) (Stats, error)

Stats returns current statistics for the subscription represented by c.

func (*Broker[T]) Subscribe

func (b *Broker[T]) Subscribe(c chan<- T, allow func(T) bool) error

Subscribe adds c to the broker, and forwards every published value that passes the allow func to c.

func (*Broker[T]) SubscribeAll

func (b *Broker[T]) SubscribeAll(c chan<- T) error

SubscribeAll subscribes to every published value.

func (*Broker[T]) Unsubscribe

func (b *Broker[T]) Unsubscribe(c chan<- T) (Stats, error)

Unsubscribe removes the given channel from the broker.

type Stats

type Stats struct {
	// Skips are values that were not sent due to filtering rules.
	Skips uint64 `json:"skips"`

	// Sends are values that were sent successfully.
	Sends uint64 `json:"sends"`

	// Drops are values that failed to send because the subscriber blocked.
	Drops uint64 `json:"drops"`
}

Stats represents the outcome of one or more published values.

func (Stats) String

func (s Stats) String() string

String representation of the stats.

func (Stats) Total

func (s Stats) Total() uint64

Total number of values represented by the stats.