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

• 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

Types

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.