messaging

Documentation

Overview

Package messaging provides higher-order messaging primitives.

Index

• type Exchange
  • func (e Exchange[Req, Res]) Err(err error)
  • func (e Exchange[Req, Res]) Ok(res Res)
  • func (e Exchange[Req, Res]) Zero()
• type ExchangeQueue
  • func (q *ExchangeQueue[Req, Res]) Do(ctx context.Context, req Req) (Res, error)
  • func (q *ExchangeQueue[Req, Res]) New(req Req) (Exchange[Req, Res], <-chan Failable[Res])
  • func (q *ExchangeQueue[Req, Res]) Recv() <-chan Exchange[Req, Res]
  • func (q *ExchangeQueue[Req, Res]) Send() chan<- Exchange[Req, Res]
• type Failable
  • func (f Failable[T]) Get() (T, error)
• type None
• type Topic
  • func (t *Topic[T]) Close()
  • func (t *Topic[T]) Publish() chan<- T
  • func (t *Topic[T]) Subscribe(ch chan<- T)
  • func (t *Topic[T]) Unsubscribe(ch chan<- T)

Types

type Exchange

type Exchange[Req, Res any] struct {
	Request  Req
	Response chan<- Failable[Res]
}

Exchange encapsulates a request/response pair.

func (Exchange[Req, Res]) Err

func (e Exchange[Req, Res]) Err(err error)

Err sends an error response.

func (Exchange[Req, Res]) Ok

func (e Exchange[Req, Res]) Ok(res Res)

Ok sends a successful response.

func (Exchange[Req, Res]) Zero

func (e Exchange[Req, Res]) Zero()

Zero sends a successful zero-valued response.

type ExchangeQueue

type ExchangeQueue[Req, Res any] struct {
	// contains filtered or unexported fields
}

ExchangeQueue is a queue of request/response exchanges.

func (*ExchangeQueue[Req, Res]) Do

func (q *ExchangeQueue[Req, Res]) Do(ctx context.Context, req Req) (Res, error)

Do performs a synchronous request/response exchange.

func (*ExchangeQueue[Req, Res]) New

func (q *ExchangeQueue[Req, Res]) New(req Req) (Exchange[Req, Res], <-chan Failable[Res])

New creates a new exchange and response channel.

func (*ExchangeQueue[Req, Res]) Recv

func (q *ExchangeQueue[Req, Res]) Recv() <-chan Exchange[Req, Res]

Recv returns a channel that, when read, dequeues the next exchange.

func (*ExchangeQueue[Req, Res]) Send

func (q *ExchangeQueue[Req, Res]) Send() chan<- Exchange[Req, Res]

Send returns a channel that, when written, enqueues an exchange.

type Failable

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

Failable encapsulates a value of type T, or an error indicating that the value cannot be obtained.

func (Failable[T]) Get

func (f Failable[T]) Get() (T, error)

Get returns the value, or an error if the value cannot be obtained.

type None

type None struct{}

None is a placeholder type for when a response is not expected.

type Topic

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

Topic is a source of messages of type T.

func (*Topic[T]) Close

func (t *Topic[T]) Close()

Close closes the topic, preventing further messages from being published.

func (*Topic[T]) Publish

func (t *Topic[T]) Publish() chan<- T

Publish returns a channel that, when written, publishes a message to this topic.

func (*Topic[T]) Subscribe

func (t *Topic[T]) Subscribe(ch chan<- T)

Subscribe registers ch as a subscriber to messages from this topic.

func (*Topic[T]) Unsubscribe

func (t *Topic[T]) Unsubscribe(ch chan<- T)

Unsubscribe removes ch as a subscriber to messages from this topic.