topic

Documentation

Overview

Package topic implements a buffered channel with dynamic fanout size. Unlike the normal Go channels, messages sent to a Topic are duplicated to all receivers. Incoming messages are queued in-memory when a receiver is not ready. Users can add/remove receivers from a topic dynamically.

Index

• func Recent[T any](t *Topic[T]) (_ T, _ bool)
• type Receiver
  • func (r *Receiver[T]) Unsubscribe()
• type Topic
  • func New[T any]() *Topic[T]
  • func (t *Topic[T]) Close() error
  • func (t *Topic[T]) SendCh() chan<- T
  • func (t *Topic[T]) Subscribe(limit int) (*Receiver[T], <-chan T, error)

Functions

func Recent

func Recent[T any](t *Topic[T]) (_ T, _ bool)

Recent returns the latest message sent to the topic. Returns false if no message was sent to the topic.

Types

type Receiver

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

func (*Receiver[T]) Unsubscribe

func (r *Receiver[T]) Unsubscribe()

Unsubscribe removes a receiver from a topic. Buffered messages that were not yet received will be discarded.

type Topic

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

Topic implements a buffered channel with dynamic fanout.

func New

func New[T any]() *Topic[T]

New creates a new topic.

func (*Topic[T]) Close

func (t *Topic[T]) Close() error

Close destroys the topic. Blocking operations will return with os.ErrClosed error. All receivers are forcibly unsubscribed. Caller is blocked till all background goroutines complete.

func (*Topic[T]) SendCh

func (t *Topic[T]) SendCh() chan<- T

SendCh returns a channel for the Topic where users can send messages. topic in other select clauses. Returns nil if topic is closed.

func (*Topic[T]) Subscribe

func (t *Topic[T]) Subscribe(limit int) (*Receiver[T], <-chan T, error)

Subscribe adds a new receiver to the topic. Maximium number of messages bufferred in the queue is controlled by the limit. A zero limit indicates unbounded queue; with a positive limit N, queue buffers the most recent N messages and with a negative limit N queue buffers the oldest N messages.