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 - 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 - Create 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 - 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 - Create 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 - 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 - 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 - Create 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 - The standard interface of a broker type.