multicast

README

multicast

Multiple writer, multiple listener message channel in Go.

• Any message from any writer goes to all listeners
• Supports adding and removing writers and listeners at runtime

Example

Initialization

m := multicast.New[string]()
defer m.Close()

Adding a Listener

listener, drain := m.NewListener(listenerCapacity)
defer drain(time.Second * 5)

go func() {
	defer listener.Done()

	for message := range listener.C() {
		// ... do something useful ...
	}
}()

Adding a Writer

writer, close := m.NewWriter()
defer close()

err := writer.Write(payload)
if err != nil {
	// ... handle error ...
}

Documentation

Index

• Variables
• type CloseFunc
• type DrainFunc
• type Listener
  • func (l *Listener[T]) C() <-chan T
  • func (l *Listener[T]) Done()
• type Multicast
  • func New[T any](opts ...Option) *Multicast[T]
  • func (m *Multicast[T]) Close()
  • func (m *Multicast[T]) NewListener(capacity int) (*Listener[T], DrainFunc)
  • func (m *Multicast[T]) NewWriter() (*Writer[T], CloseFunc)
• type Option
  • func WithOutboundQueueSize(size int) Option
• type Writer
  • func (w *Writer[T]) Write(message T) error

Variables

var ErrWriterClosed = errors.New("writer is closed")

ErrWriterClosed is returned when Write() is invoked on a closed Writer.

Types

type CloseFunc

type CloseFunc func()

CloseFunc represents a function to close a multicast Writer.

type DrainFunc

type DrainFunc func(time.Duration)

DrainFunc represents a function to wait for a multicast Listener to process its inbound queue.

type Listener

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

Listener represents a multicast listener.

func (*Listener[T]) C

func (l *Listener[T]) C() <-chan T

C returns the Listener's (receive-only) channel.

func (*Listener[T]) Done

func (l *Listener[T]) Done()

Done is used to signal that the Listener finished draining its inbound channel.

type Multicast

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

Multicast represents an object capable of broadcasting messages from multiple writers to multiple listeners.

func New

func New[T any](opts ...Option) *Multicast[T]

New returns a newly initialized Multicast of the specified type.

func (*Multicast[T]) Close

func (m *Multicast[T]) Close()

Close closes the Multicast.

func (*Multicast[T]) NewListener

func (m *Multicast[T]) NewListener(capacity int) (*Listener[T], DrainFunc)

NewListener returns a new message Listener for the multicast.

func (*Multicast[T]) NewWriter

func (m *Multicast[T]) NewWriter() (*Writer[T], CloseFunc)

NewWriter returns a new message Writer for the multicast.

type Option

type Option func(*config)

Option represents a Multicast configuration option.

func WithOutboundQueueSize

func WithOutboundQueueSize(size int) Option

WithOutboundQueueSize is a Multicast configuration option to set a non default value for the outbound queue (channel) buffer size.

type Writer

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

Writer represents a multicast writer.

func (*Writer[T]) Write

func (w *Writer[T]) Write(message T) error

Write writes a message to the Writer.