message

Documentation

Index

• Constants
• Variables
• type Bus
• type DecoderFunc
  • func NewDecoder[T any]() DecoderFunc
• type DecoderRegistry
  • func NewRegistry() *DecoderRegistry
  • func (r *DecoderRegistry) Decode(kind string, buf []byte) (any, error)
  • func (r *DecoderRegistry) Register(kind string, decoderFunc DecoderFunc) *DecoderRegistry
• type Envelope
• type Handler
• type InMemoryBus
  • func NewInMemoryBus() *InMemoryBus
  • func (i *InMemoryBus) Subject(ctx context.Context, subject string) (Mailbox, error)
• type InMemoryMailbox
  • func NewInMemoryMailbox() *InMemoryMailbox
  • func (i *InMemoryMailbox) Emit(ctx context.Context, message Message) error
  • func (i *InMemoryMailbox) ReadC(ctx context.Context) (<-chan Message, context.CancelFunc)
• type Mailbox
• type Message
  • func New(kind string, payload any) Message
  • func Unmarshall(buf []byte) (*Message, error)
  • func (m Message) CreatedAt() time.Time
  • func (m Message) Marshall() ([]byte, error)
  • func (m Message) Node() []byte
• type NoOpBus
  • func NewNoopBus() *NoOpBus
  • func (NoOpBus) Emit(_ context.Context, _ Message) error
  • func (NoOpBus) ReadC(_ context.Context) (<-chan Message, context.CancelFunc)
  • func (b NoOpBus) Subject(_ context.Context, _ string) (Mailbox, error)

Constants

const (
	InMemoryBusMaximumInFlightMessage = 4096
)

Variables

var GlobalDecoderRegistry = NewRegistry()

Types

type Bus

type Bus interface {
	Subject(ctx context.Context, subject string) (Mailbox, error)
}

type DecoderFunc

type DecoderFunc func(buf []byte) (any, error)

func NewDecoder

func NewDecoder[T any]() DecoderFunc

type DecoderRegistry

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

func NewRegistry

func NewRegistry() *DecoderRegistry

func (*DecoderRegistry) Decode

func (r *DecoderRegistry) Decode(kind string, buf []byte) (any, error)

func (*DecoderRegistry) Register

func (r *DecoderRegistry) Register(kind string, decoderFunc DecoderFunc) *DecoderRegistry

type Envelope

type Envelope struct {
	Message Message `json:"message"`
	Payload []byte  `json:"payload"`
}

type Handler

type Handler func(ctx context.Context, message Message) (Message, error)

type InMemoryBus

type InMemoryBus struct{}

func NewInMemoryBus

func NewInMemoryBus() *InMemoryBus

func (*InMemoryBus) Subject

func (i *InMemoryBus) Subject(ctx context.Context, subject string) (Mailbox, error)

type InMemoryMailbox

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

func NewInMemoryMailbox

func NewInMemoryMailbox() *InMemoryMailbox

func (*InMemoryMailbox) Emit

func (i *InMemoryMailbox) Emit(ctx context.Context, message Message) error

func (*InMemoryMailbox) ReadC

func (i *InMemoryMailbox) ReadC(ctx context.Context) (<-chan Message, context.CancelFunc)

type Mailbox

type Mailbox interface {
	Emit(ctx context.Context, message Message) error
	ReadC(ctx context.Context) (<-chan Message, context.CancelFunc)
}

type Message

type Message struct {
	ID      xid.ID `json:"id"`
	Kind    string `json:"kind"`
	Payload any    `json:"-"`
}

func New

func New(kind string, payload any) Message

func Unmarshall

func Unmarshall(buf []byte) (*Message, error)

func (Message) CreatedAt

func (m Message) CreatedAt() time.Time

func (Message) Marshall

func (m Message) Marshall() ([]byte, error)

func (Message) Node

func (m Message) Node() []byte

type NoOpBus

type NoOpBus struct{}

func NewNoopBus

func NewNoopBus() *NoOpBus

func (NoOpBus) Emit

func (NoOpBus) Emit(_ context.Context, _ Message) error

func (NoOpBus) ReadC

func (NoOpBus) ReadC(_ context.Context) (<-chan Message, context.CancelFunc)

func (NoOpBus) Subject

func (b NoOpBus) Subject(_ context.Context, _ string) (Mailbox, error)