broker

Documentation

Index

• Constants
• Variables
• type Broker
  • func New(events store.EventStore) *Broker
  • func (b *Broker) Close(id rlid.RLID) error
  • func (b *Broker) Lock()
  • func (b *Broker) NumPublishers() int
  • func (b *Broker) NumSubscribers() int
  • func (b *Broker) Publish(publisherID rlid.RLID, event *api.EventWrapper) error
  • func (b *Broker) RLock()
  • func (b *Broker) RUnlock()
  • func (b *Broker) Register() (rlid.RLID, <-chan PublishResult, error)
  • func (b *Broker) Run(errc chan<- error)
  • func (b *Broker) Shutdown() error
  • func (b *Broker) Subscribe(topics ...ulid.ULID) (rlid.RLID, <-chan *api.EventWrapper, error)
  • func (b *Broker) Unlock()
• type PublishResult
  • func (p PublishResult) Ack() *api.Ack
  • func (p PublishResult) IsAck() bool
  • func (p PublishResult) IsNack() bool
  • func (p PublishResult) Nack() *api.Nack
  • func (p PublishResult) Reply() *api.PublisherReply

Constants

const BufferSize = 16384

Variables

var (
	ErrBrokerNotRunning = errors.New("operation could not be completed: broker is not running")
	ErrUnknownID        = errors.New("no publisher or subscriber registered with specified id")
)

Types

type Broker

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

Every Ensign node is composed of a single Broker routine that collects events from publisher handlers, commits the events through consensus, writes the events to disk, and ensures that registered consumer groups receive any published events they are subscribed to. Essentially, the Broker fans in events from multiple, concurrent publisher streams, ensures the events are replicated and written, then fans out the events to one or more subscriber streams. The Broker uses an internal buffer that applies backpressure to the publisher streams when the buffer is full.

func New

func New(events store.EventStore) *Broker

func (*Broker) Close

func (b *Broker) Close(id rlid.RLID) error

Close either a publisher or subscriber so no events will be sent from the broker.

func (*Broker) Lock

func (b *Broker) Lock()

Locks both the pubmu and submu mutexes.

func (*Broker) NumPublishers

func (b *Broker) NumPublishers() int

func (*Broker) NumSubscribers

func (b *Broker) NumSubscribers() int

func (*Broker) Publish

func (b *Broker) Publish(publisherID rlid.RLID, event *api.EventWrapper) error

Publish an event from the specified publisher. When the event is committed, an acknowledgement or error is sent on the channel specified when registering.

func (*Broker) RLock

func (b *Broker) RLock()

RLocks the pubmu and submu mutexes.

func (*Broker) RUnlock

func (b *Broker) RUnlock()

RUnlocks both the pubmu and submu mutexes.

func (*Broker) Register

func (b *Broker) Register() (rlid.RLID, <-chan PublishResult, error)

Register a publisher to receive an ack/nack channel for events that are published using the publisher ID specified. If the broker is not running, an error is returned so that the publisher can shutdown the stream.

func (*Broker) Run

func (b *Broker) Run(errc chan<- error)

Run the broker; any fatal errors will be sent on the specified channel.

func (*Broker) Shutdown

func (b *Broker) Shutdown() error

Gracefully shutdown the broker. If a consensus or write operation is underway, then shutdown blocks until it is concluded. The broker then stops handling incoming events from publishers and closes all registered publishers and subscribers. This has the effect of closing any open event stream handlers.

func (*Broker) Subscribe

func (b *Broker) Subscribe(topics ...ulid.ULID) (rlid.RLID, <-chan *api.EventWrapper, error)

Subscribe to events filtered by topic ids. All recent events will be sent on the event wrapper channel once they are committed. If the broker is not running an error is returned so that the consumer group can shutdown the stream.

func (*Broker) Unlock

func (b *Broker) Unlock()

Unlocks both the pubmu and submu mutexex.

type PublishResult

type PublishResult struct {
	LocalID   []byte                 // The localID on the event sent by the publisher for client-side correlation
	Committed *timestamppb.Timestamp // The timestamp the event was committed (if it was committed)
	Code      api.Nack_Code          // The reason why the result errored; if not unknown the result is treated as an error
	Error     string                 // An error message, should be set if the nack code is set
}

PublishResult is sent back to the publisher stream that created an event to let it send an ack/nack back to to the client. If the event was correctly committed and emitted the result will contain an Ack message. If the event was unable to be processed by the broker (e.g. not committed, not written, etc.) then the result will contain a Nack message.

func (PublishResult) Ack

func (p PublishResult) Ack() *api.Ack

func (PublishResult) IsAck

func (p PublishResult) IsAck() bool

Returns true if the reply is an ack, false if it is a nack

func (PublishResult) IsNack

func (p PublishResult) IsNack() bool

Returns true if the reply is a nack, fals if it is an ack

func (PublishResult) Nack

func (p PublishResult) Nack() *api.Nack

func (PublishResult) Reply

func (p PublishResult) Reply() *api.PublisherReply

Reply composes a publisher reply to return to the client via the publish stream. If the Code is > 0 (e.g. is not NACK_UNKNOWN) then a Nack is returned; otherwise an Ack is returned. This method performs no data validation other than to set the error message to a standard message for the code if it is a nack.