pubsub

Documentation

Index

• Variables
• func EmitEvent(w io.Writer, event Event) error
• func EmitEvents(ctx context.Context, w http.ResponseWriter, events <-chan Event)
• type Channel
• type ClientID
• type CountDownLatch
  • func NewCountDownLatch(count int) *CountDownLatch
  • func (l *CountDownLatch) CountDown()
  • func (l *CountDownLatch) Wait(duration time.Duration) bool
• type ErrWriter
  • func (ErrWriter) Write(p []byte) (n int, err error)
• type Event
• type Registry
  • func (r *Registry) Publish(channel Channel, event Event)
  • func (r *Registry) Subscribe(channel Channel, stream chan<- Event) (ClientID, error)
  • func (r *Registry) SubscribeMatching(fn func(Channel, Event) bool, stream chan<- Event) (ClientID, error)
  • func (r *Registry) Unsubscribe(id ClientID)

Variables

var PingEvent = Event{Kind: "ping"}

Functions

func EmitEvent

func EmitEvent(w io.Writer, event Event) error

EmitEvent marshals an event to JSON and sends it as a SSE message to the provided io.Writer. If the provided io.Writer implements the http.Flusher interface than the writer will be flushed after the write occurs.

func EmitEvents

func EmitEvents(ctx context.Context, w http.ResponseWriter, events <-chan Event)

EmitEvents will loop and send events to the provided HTTP response. The events will be formatted according to the W3C working draft for Server-Sent Events found at: https://www.w3.org/TR/2009/WD-eventsource-20090421. This method is useful for implementing the server side component of a SSE endpoint.

EmitEvents will block until either the events channel is closed, the provided context is done, or an error occurs while emitting an event.

If no events are available on the events channel for 30 seconds then a ping event will be synthesized and emitted automatically in order to keep the connection with the client alive.

Types

type Channel

type Channel string

Channel represents the segment of clients that a subscription is for or that an event should be delivered to.

type ClientID

type ClientID string

ClientID represents the identifier of a client that has subscribed to the registry.

type CountDownLatch

type CountDownLatch struct {
	sync.Mutex
	// contains filtered or unexported fields
}

CountDownLatch is a synchronization aid that allows one or more goroutines to wait until a set of operations completes.

func NewCountDownLatch

func NewCountDownLatch(count int) *CountDownLatch

func (*CountDownLatch) CountDown

func (l *CountDownLatch) CountDown()

func (*CountDownLatch) Wait

func (l *CountDownLatch) Wait(duration time.Duration) bool

type ErrWriter

type ErrWriter struct{}

Writer whose Write method always returns an error.

func (ErrWriter) Write

func (ErrWriter) Write(p []byte) (n int, err error)

type Event

type Event struct {
	Kind    string      `json:"kind"`
	Payload interface{} `json:"payload,omitempty"`
}

Event encapsulates an event that can be sent to all subscribed clients of a registry.

type Registry

type Registry struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Registry manages the event streams for subscribed clients across different channels. Internally the registry keeps track of each subscribed client for a channel using an identifier. This identifier can be used by the client to unsubscribe to stop receiving any future events. The registry is safe to access from multiple goroutines.

func (*Registry) Publish

func (r *Registry) Publish(channel Channel, event Event)

Publish sends an event to all subscribed clients of a given channel. If a client's stream is full the event will be skipped.

func (*Registry) Subscribe

func (r *Registry) Subscribe(channel Channel, stream chan<- Event) (ClientID, error)

Subscribe adds a new client stream for a particular channel. The provided stream will be associated with the channel and begin to receive all events for that channel. A unique client identifier will be returned so that in the future the client can be unsubscribed to stop events from being received.

If desired, the provided stream can be used externally to target events for a single client. It can also be passed into the subscribe method with events already in it so that a set of initialization events can be sent to the client before any published events.

NOTE: The passed in stream should not be closed prior to the client being unsubscribed from the registry.

func (*Registry) SubscribeMatching

func (r *Registry) SubscribeMatching(fn func(Channel, Event) bool, stream chan<- Event) (ClientID, error)

SubscribeMatching adds a new client stream for all events published that are for a channel that matches a provided function. The provided stream will be associated with the subscription and begin to receive all matching events once the subscribe call returns. A unique client identifier will be returned so that in the future the client can be unsubscribed to stop events from being received.

If desired, the provided stream can be used externally to target events for a single client. It can also be passed into the subscribe method with events already in it so that a set of initialization events can be sent to the client before any published events.

NOTE: The passed in stream should not be closed prior to the client being unsubscribed from the registry.

func (*Registry) Unsubscribe

func (r *Registry) Unsubscribe(id ClientID)

Unsubscribe removes a client from a particular channel. Once this method returns no further events will be received on that client's stream.