pubsub

README

pubsub

Golang implementation of publish-subscribe pattern using channels.

err := pubsub.NewTopic[int]("topic", pubsub.NewRingBuffer[int](16))
if err != nil {
    // ...
}

subscriber, err = pubsub.Subscribe[int]("topic")
if err != nil {
    // ...
}

publisher, err = pubsub.NewPublisher[int]("topic")
if err != nil {
    // ...
}

• Topics are explicitly created instead of implicit

• Messages are not persisted to disk

• Each subscriber has its own internal buffer to prevent deadlocks

• Does not deep copy structs

Documentation

Index

• Variables
• func Close[T any](name string)
• func NewTopic[T any](name string, history ...ICache[T]) error
• type EmptyCache
  • func (h *EmptyCache[T]) Iter(f func(T))
  • func (h *EmptyCache[T]) Push(item T)
• type ICache
  • func NewEmptyCache[T any]() ICache[T]
  • func NewRingCache[T any](bufferSize ...int) ICache[T]
• type Publisher
  • func NewPublisher[T any](name string) (*Publisher[T], error)
  • func (p *Publisher[T]) Publish(msg T) error
• type RingCache
  • func (r *RingCache[T]) Iter(f func(T))
  • func (r *RingCache[T]) Push(item T)
• type Subscriber
  • func NewSubscriber[T any](name string) (*Subscriber[T], error)
  • func (s *Subscriber[T]) Read() []T
  • func (s *Subscriber[T]) Recv() <-chan struct{}
  • func (s *Subscriber[T]) Unsubscribe()

Variables

var (
	ErrTopicDoesNotExist = errors.New("topic does not exist")
	ErrTopicExists       = errors.New("topic exists")
	ErrTypeMismatch      = errors.New("type mismatch")
)

Functions

func Close

func Close[T any](name string)

func NewTopic

func NewTopic[T any](name string, history ...ICache[T]) error

Types

type EmptyCache

type EmptyCache[T any] struct {
}

func (*EmptyCache[T]) Iter

func (h *EmptyCache[T]) Iter(f func(T))

func (*EmptyCache[T]) Push

func (h *EmptyCache[T]) Push(item T)

type ICache

type ICache[T any] interface {
	Push(item T)
	Iter(func(T))
}

func NewEmptyCache

func NewEmptyCache[T any]() ICache[T]

func NewRingCache

func NewRingCache[T any](bufferSize ...int) ICache[T]

type Publisher

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

func NewPublisher

func NewPublisher[T any](name string) (*Publisher[T], error)

func (*Publisher[T]) Publish

func (p *Publisher[T]) Publish(msg T) error

type RingCache

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

func (*RingCache[T]) Iter

func (r *RingCache[T]) Iter(f func(T))

func (*RingCache[T]) Push

func (r *RingCache[T]) Push(item T)

type Subscriber

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

func NewSubscriber

func NewSubscriber[T any](name string) (*Subscriber[T], error)

func (*Subscriber[T]) Read

func (s *Subscriber[T]) Read() []T

func (*Subscriber[T]) Recv

func (s *Subscriber[T]) Recv() <-chan struct{}

func (*Subscriber[T]) Unsubscribe

func (s *Subscriber[T]) Unsubscribe()