multi

Documentation

Index

• type MultiDestination
  • func NewMultiDestination[T any](dests []kawa.Destination[T]) MultiDestination[T]
  • func (md MultiDestination[T]) Send(ctx context.Context, ack func(), msgs ...kawa.Message[T]) error
• type MultiSource
  • func NewMultiSource[T any](sources []kawa.Source[T]) MultiSource[T]
  • func (ms MultiSource[T]) Recv(ctx context.Context) (kawa.Message[T], func(), error)
  • func (ms MultiSource[T]) Run(ctx context.Context) error

Types

type MultiDestination

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

func NewMultiDestination

func NewMultiDestination[T any](dests []kawa.Destination[T]) MultiDestination[T]

TODO: options for ack behavior?

func (MultiDestination[T]) Send

func (md MultiDestination[T]) Send(ctx context.Context, ack func(), msgs ...kawa.Message[T]) error

type MultiSource

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

MultiSource multiplexes multiple sources into one. It was thrown together quickly and may have some performance issues. It definitely needs some work on proper error handling, and concurrency issues on closing.

func NewMultiSource

func NewMultiSource[T any](sources []kawa.Source[T]) MultiSource[T]

TODO: options for ack behavior?

func (MultiSource[T]) Recv

func (ms MultiSource[T]) Recv(ctx context.Context) (kawa.Message[T], func(), error)

func (MultiSource[T]) Run

func (ms MultiSource[T]) Run(ctx context.Context) error

Run assumes the wrapped sources are already running, it spawns a go-routine for each source being wrapped, and in a loop reads its Recv method, then makes that message available on the Recv method for the multi source. Sources will be "competing" to send events on the shared channel, which means that faster sources have the potential to "starve" slower ones. At our current scale, this shouldn't be an issue.