pubsub

README

pubsub

GoDoc Reference

Package pubsub implements an unbounded channel for the pub/sub pattern.

Documentation

Overview

Package pubsub implements an unbounded channel for the pub/sub pattern.

Index

• type FilterFunc
• type Queue
  • func NewQueue[T any]() *Queue[T]
  • func NewQueueWithCapacity[T any](cap int) *Queue[T]
  • func (q *Queue[T]) Dequeue()
  • func (q *Queue[T]) Enqueue(v T)
  • func (q *Queue[T]) IsEmpty() bool
  • func (q *Queue[T]) Pending() (T, bool)
  • func (q *Queue[T]) PendingOrZero() T
• type Subscriber
  • func NewSubscriber[T any]() *Subscriber[T]
  • func (s *Subscriber[T]) Listen(ctx context.Context, src <-chan T) error
  • func (s *Subscriber[T]) Subscribe(ch chan<- T, filter FilterFunc[T])
  • func (s *Subscriber[T]) Unsubscribe(ch chan<- T)

Types

type FilterFunc

type FilterFunc[T any] func(T) bool

FilterFunc is a filter function for any type. If the function returns true, the message will be sent to the subscriber. If the function is nil, all messages will be sent to the subscriber.

type Queue

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

Queue implements a simple first-in-first-out queue. It is not thread-safe. A zero-value Queue is a valid queue.

func NewQueue

func NewQueue[T any]() *Queue[T]

NewQueue creates a new queue with a default capacity of 16.

func NewQueueWithCapacity

func NewQueueWithCapacity[T any](cap int) *Queue[T]

NewQueueWithCapacity creates a new queue with the given capacity. The capacity is set to be at least 16.

func (*Queue[T]) Dequeue

func (q *Queue[T]) Dequeue()

Dequeue pops the first element from the queue and discards it.

func (*Queue[T]) Enqueue

func (q *Queue[T]) Enqueue(v T)

Enqueue queues the given item.

func (*Queue[T]) IsEmpty

func (q *Queue[T]) IsEmpty() bool

IsEmpty returns true if the queue is empty.

func (*Queue[T]) Pending

func (q *Queue[T]) Pending() (T, bool)

Pending returns the pending item in the queue and a boolean indicating whether the item was found.

func (*Queue[T]) PendingOrZero

func (q *Queue[T]) PendingOrZero() T

PendingOrZero returns the pending item in the queue or the zero value if the queue is empty.

type Subscriber

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

Subscriber is a subscriber that subscribes to a Pipe. A zero-value Subscriber is a valid Subscriber.

func NewSubscriber

func NewSubscriber[T any]() *Subscriber[T]

NewSubscriber creates a new Subscriber.

func (*Subscriber[T]) Listen

func (s *Subscriber[T]) Listen(ctx context.Context, src <-chan T) error

Listen starts broadcasting messages received from the given src channel. It blocks until the src channel is closed or ctx is canceled.

func (*Subscriber[T]) Subscribe

func (s *Subscriber[T]) Subscribe(ch chan<- T, filter FilterFunc[T])

Subscribe subscribes ch to incoming messages from the given recipient. Calls to Subscribe should always be paired with Unsubscribe. It is recommended to use defer.

Subscribe panics if it's called on a Subscriber w/ a src that's already closed.

func (*Subscriber[T]) Unsubscribe

func (s *Subscriber[T]) Unsubscribe(ch chan<- T)

Unsubscribe unsubscribes ch from incoming messages from all its recipients. Once unsubscribed, ch will be closed.