chanbuf

Documentation

Index

• Variables
• func Publish(msg streams.Message) error
• func Shutdown()
• func Start()
• func Subscribe(stream string, handler streams.ReaderHandleFunc)
• type Bus
  • func NewBus(cfg Config) *Bus
  • func (b *Bus) Publish(msg streams.Message) error
  • func (b *Bus) Shutdown()
  • func (b *Bus) Start()
  • func (b *Bus) Subscribe(stream string, handler streams.ReaderHandleFunc)
• type Config
• type Reader
  • func NewReader(b *Bus) Reader
  • func (r Reader) Read(_ context.Context, task streams.ReadTask) error
• type Writer
  • func NewWriter(b *Bus) Writer
  • func (w Writer) Write(_ context.Context, msgBatch []streams.Message) error

Variables

var (
	// DefaultBus global Bus instance used by both Writer and Reader if passed Bus instance was nil.
	DefaultBus *Bus
	// DefaultBusConfig global Bus instance configuration used by both Writer and Reader if passed Bus instance was nil.
	DefaultBusConfig Config
)

Functions

func Publish

func Publish(msg streams.Message) error

Publish sends a message to one or many readers subscribed to msg.StreamName (aka. fan-out) in a fire-and-forget way meaning the internal writer only cares about writing to the queue whereas subscribers are independent processes.

This routine will fail if Bus is shut down.

func Shutdown

func Shutdown()

Shutdown disposes allocated resources and signals internal processes for a graceful shutdown.

This operation might block I/O as it waits for in-flight reader processes to finish -either by success or timeout-.

func Start

func Start()

Start spins up the Bus readers, blocking the I/O until Bus.Shutdown is called.

func Subscribe

func Subscribe(stream string, handler streams.ReaderHandleFunc)

Subscribe appends a reader handler to a stream.

Types

type Bus

type Bus struct {
	// contains filtered or unexported fields
}

Bus Go channel-backed concurrent-safe messaging bus which implements fan-out messaging pattern.

Zero-value IS NOT ready to use, please call NewBus routine instead.

func NewBus

func NewBus(cfg Config) *Bus

NewBus allocates a new Bus.

func (*Bus) Publish

func (b *Bus) Publish(msg streams.Message) error

Publish sends a message to one or many readers subscribed to msg.StreamName (aka. fan-out) in a fire-and-forget way meaning the internal writer only cares about writing to the queue whereas subscribers are independent processes.

This routine will fail if Bus is shut down.

func (*Bus) Shutdown

func (b *Bus) Shutdown()

Shutdown disposes allocated resources and signals internal processes for a graceful shutdown.

This operation might block I/O as it waits for in-flight reader processes to finish -either by success or timeout-.

func (*Bus) Start

func (b *Bus) Start()

Start spins up the Bus readers, blocking the I/O until Bus.Shutdown is called.

func (*Bus) Subscribe

func (b *Bus) Subscribe(stream string, handler streams.ReaderHandleFunc)

Subscribe appends a reader handler to a stream.

type Config

type Config struct {
	// Internal queue buffer capacity. Bus will allocate an unbuffered channel if <= 0.
	QueueBufferFactor int
	// streams.ReaderHandleFunc maximum process execution duration.
	ReaderHandlerTimeout time.Duration
	// logging instance used by internal processes.
	Logger *log.Logger
}

Config Bus configuration parameters.

type Reader

type Reader struct {
	// contains filtered or unexported fields
}

func NewReader

func NewReader(b *Bus) Reader

func (Reader) Read

func (r Reader) Read(_ context.Context, task streams.ReadTask) error

type Writer

type Writer struct {
	// contains filtered or unexported fields
}

func NewWriter

func NewWriter(b *Bus) Writer

func (Writer) Write

func (w Writer) Write(_ context.Context, msgBatch []streams.Message) error