broker

Documentation

Overview

Package broker implements types used for routing inputs to outputs in non-trivial arrangements, such as fan-out or fan-in models.

Index

• func OptDynamicFanInSetOnAdd(onAddFunc func(label string)) func(*DynamicFanIn)
• func OptDynamicFanInSetOnRemove(onRemoveFunc func(label string)) func(*DynamicFanIn)
• func OptDynamicFanOutSetOnAdd(onAddFunc func(label string)) func(*DynamicFanOut)
• func OptDynamicFanOutSetOnRemove(onRemoveFunc func(label string)) func(*DynamicFanOut)
• type DynamicFanIn
  • func NewDynamicFanIn(inputs map[string]DynamicInput, logger log.Modular, stats metrics.Type, ...) (*DynamicFanIn, error)
  • func (d *DynamicFanIn) CloseAsync()
  • func (d *DynamicFanIn) MessageChan() <-chan types.Message
  • func (d *DynamicFanIn) SetInput(ident string, input DynamicInput, timeout time.Duration) error
  • func (d *DynamicFanIn) StartListening(responseChan <-chan types.Response) error
  • func (d *DynamicFanIn) WaitForClose(timeout time.Duration) error
• type DynamicFanOut
  • func NewDynamicFanOut(outputs map[string]DynamicOutput, logger log.Modular, stats metrics.Type, ...) (*DynamicFanOut, error)
  • func (d *DynamicFanOut) CloseAsync()
  • func (d *DynamicFanOut) ResponseChan() <-chan types.Response
  • func (d *DynamicFanOut) SetOutput(ident string, output DynamicOutput, timeout time.Duration) error
  • func (d *DynamicFanOut) StartReceiving(msgs <-chan types.Message) error
  • func (d *DynamicFanOut) WaitForClose(timeout time.Duration) error
• type DynamicInput
• type DynamicOutput
• type FanIn
  • func NewFanIn(inputs []types.Producer, stats metrics.Type) (*FanIn, error)
  • func (i *FanIn) CloseAsync()
  • func (i *FanIn) MessageChan() <-chan types.Message
  • func (i *FanIn) StartListening(responseChan <-chan types.Response) error
  • func (i *FanIn) WaitForClose(timeout time.Duration) error
• type FanOut
  • func NewFanOut(conf FanOutConfig, outputs []types.Consumer, logger log.Modular, ...) (*FanOut, error)
  • func (o *FanOut) CloseAsync()
  • func (o *FanOut) ResponseChan() <-chan types.Response
  • func (o *FanOut) StartReceiving(msgs <-chan types.Message) error
  • func (o *FanOut) WaitForClose(timeout time.Duration) error
• type FanOutConfig
  • func NewFanOutConfig() FanOutConfig
• type MockType
  • func (m MockType) CloseAsync()
  • func (m MockType) WaitForClose(time.Duration) error
• type RoundRobin
  • func NewRoundRobin(outputs []types.Consumer, stats metrics.Type) (*RoundRobin, error)
  • func (o *RoundRobin) CloseAsync()
  • func (o *RoundRobin) ResponseChan() <-chan types.Response
  • func (o *RoundRobin) StartReceiving(msgs <-chan types.Message) error
  • func (o *RoundRobin) WaitForClose(timeout time.Duration) error
• type Type

Functions

func OptDynamicFanInSetOnAdd

func OptDynamicFanInSetOnAdd(onAddFunc func(label string)) func(*DynamicFanIn)

OptDynamicFanInSetOnAdd sets the function that is called whenever a dynamic input is added.

func OptDynamicFanInSetOnRemove

func OptDynamicFanInSetOnRemove(onRemoveFunc func(label string)) func(*DynamicFanIn)

OptDynamicFanInSetOnRemove sets the function that is called whenever a dynamic input is removed.

func OptDynamicFanOutSetOnAdd

func OptDynamicFanOutSetOnAdd(onAddFunc func(label string)) func(*DynamicFanOut)

OptDynamicFanOutSetOnAdd sets the function that is called whenever a dynamic output is added.

func OptDynamicFanOutSetOnRemove

func OptDynamicFanOutSetOnRemove(onRemoveFunc func(label string)) func(*DynamicFanOut)

OptDynamicFanOutSetOnRemove sets the function that is called whenever a dynamic output is removed.

Types

type DynamicFanIn

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

DynamicFanIn is a broker that implements types.Producer and manages a map of inputs to unique string identifiers, routing them through a single message channel. Inputs can be added and removed dynamically as the broker runs.

func NewDynamicFanIn

func NewDynamicFanIn(
	inputs map[string]DynamicInput,
	logger log.Modular,
	stats metrics.Type,
	options ...func(*DynamicFanIn),
) (*DynamicFanIn, error)

NewDynamicFanIn creates a new DynamicFanIn type by providing an initial map map of inputs.

func (*DynamicFanIn) CloseAsync

func (d *DynamicFanIn) CloseAsync()

CloseAsync shuts down the DynamicFanIn broker and stops processing requests.

func (*DynamicFanIn) MessageChan

func (d *DynamicFanIn) MessageChan() <-chan types.Message

MessageChan returns the channel used for consuming messages from this broker.

func (*DynamicFanIn) SetInput

func (d *DynamicFanIn) SetInput(ident string, input DynamicInput, timeout time.Duration) error

SetInput attempts to add a new input to the dynamic input broker. If an input already exists with the same identifier it will be closed and removed. If either action takes longer than the timeout period an error will be returned.

A nil input is safe and will simply remove the previous input under the indentifier, if there was one.

func (*DynamicFanIn) StartListening

func (d *DynamicFanIn) StartListening(responseChan <-chan types.Response) error

StartListening assigns a new responses channel for the broker to read.

func (*DynamicFanIn) WaitForClose

func (d *DynamicFanIn) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the DynamicFanIn broker has closed down.

type DynamicFanOut

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

DynamicFanOut is a broker that implements types.Consumer and broadcasts each message out to a dynamic map of outputs.

func NewDynamicFanOut

func NewDynamicFanOut(
	outputs map[string]DynamicOutput,
	logger log.Modular,
	stats metrics.Type,
	options ...func(*DynamicFanOut),
) (*DynamicFanOut, error)

NewDynamicFanOut creates a new DynamicFanOut type by providing outputs.

func (*DynamicFanOut) CloseAsync

func (d *DynamicFanOut) CloseAsync()

CloseAsync shuts down the DynamicFanOut broker and stops processing requests.

func (*DynamicFanOut) ResponseChan

func (d *DynamicFanOut) ResponseChan() <-chan types.Response

ResponseChan returns the response channel.

func (*DynamicFanOut) SetOutput

func (d *DynamicFanOut) SetOutput(ident string, output DynamicOutput, timeout time.Duration) error

SetOutput attempts to add a new output to the dynamic output broker. If an output already exists with the same identifier it will be closed and removed. If either action takes longer than the timeout period an error will be returned.

A nil output argument is safe and will simply remove the previous output under the indentifier, if there was one.

func (*DynamicFanOut) StartReceiving

func (d *DynamicFanOut) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a new messages channel for the broker to read.

func (*DynamicFanOut) WaitForClose

func (d *DynamicFanOut) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the DynamicFanOut broker has closed down.

type DynamicInput

type DynamicInput interface {
	types.MessageSender
	types.ResponderListener
	types.Closable
}

DynamicInput is an interface of input types that must be closable.

type DynamicOutput

type DynamicOutput interface {
	types.MessageReceiver
	types.Responder
	types.Closable
}

DynamicOutput is an interface of output types that must be closable.

type FanIn

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

FanIn is a broker that implements types.Producer, takes an array of inputs and routes them through a single message channel.

func NewFanIn

func NewFanIn(inputs []types.Producer, stats metrics.Type) (*FanIn, error)

NewFanIn creates a new FanIn type by providing inputs.

func (*FanIn) CloseAsync

func (i *FanIn) CloseAsync()

CloseAsync shuts down the FanIn broker and stops processing requests.

func (*FanIn) MessageChan

func (i *FanIn) MessageChan() <-chan types.Message

MessageChan returns the channel used for consuming messages from this broker.

func (*FanIn) StartListening

func (i *FanIn) StartListening(responseChan <-chan types.Response) error

StartListening assigns a new responses channel for the broker to read.

func (*FanIn) WaitForClose

func (i *FanIn) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the FanIn broker has closed down.

type FanOut

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

FanOut is a broker that implements types.Consumer and broadcasts each message out to an array of outputs.

func NewFanOut

func NewFanOut(
	conf FanOutConfig, outputs []types.Consumer, logger log.Modular, stats metrics.Type,
) (*FanOut, error)

NewFanOut creates a new FanOut type by providing outputs.

func (*FanOut) CloseAsync

func (o *FanOut) CloseAsync()

CloseAsync shuts down the FanOut broker and stops processing requests.

func (*FanOut) ResponseChan

func (o *FanOut) ResponseChan() <-chan types.Response

ResponseChan returns the response channel.

func (*FanOut) StartReceiving

func (o *FanOut) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a new messages channel for the broker to read.

func (*FanOut) WaitForClose

func (o *FanOut) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the FanOut broker has closed down.

type FanOutConfig

type FanOutConfig struct {
}

FanOutConfig is config values for the fan out type.

func NewFanOutConfig

func NewFanOutConfig() FanOutConfig

NewFanOutConfig creates a FanOutConfig fully populated with default values.

type MockType

type MockType struct {
}

MockType implements the broker.Type interface.

func (MockType) CloseAsync

func (m MockType) CloseAsync()

CloseAsync does nothing.

func (MockType) WaitForClose

func (m MockType) WaitForClose(time.Duration) error

WaitForClose does nothing.

type RoundRobin

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

RoundRobin is a broker that implements types.Consumer and sends each message out to a single consumer chosen from an array in round-robin fashion. Consumers that apply backpressure will block all consumers.

func NewRoundRobin

func NewRoundRobin(outputs []types.Consumer, stats metrics.Type) (*RoundRobin, error)

NewRoundRobin creates a new RoundRobin type by providing consumers.

func (*RoundRobin) CloseAsync

func (o *RoundRobin) CloseAsync()

CloseAsync shuts down the RoundRobin broker and stops processing requests.

func (*RoundRobin) ResponseChan

func (o *RoundRobin) ResponseChan() <-chan types.Response

ResponseChan returns the response channel.

func (*RoundRobin) StartReceiving

func (o *RoundRobin) StartReceiving(msgs <-chan types.Message) error

StartReceiving assigns a new messages channel for the broker to read.

func (*RoundRobin) WaitForClose

func (o *RoundRobin) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the RoundRobin broker has closed down.

type Type

type Type interface {
	types.Closable
}

Type is the standard interface of a broker type.