kernel

Documentation

Index

• func Dequeue[T any](k *Dequeuer, cat string, codec Decoder[T]) (<-chan swarm.Msg[T], chan<- swarm.Msg[T])
• func Enqueue[T any](k *Enqueuer, cat string, codec Encoder[T]) (chan<- T, <-chan T)
• type Bridge
  • func NewBridge(timeToFlight time.Duration) *Bridge
  • func (s *Bridge) Ack(ctx context.Context, digest string) error
  • func (s *Bridge) Ask(ctx context.Context) ([]swarm.Bag, error)
  • func (s *Bridge) Dispatch(seq []swarm.Bag) error
  • func (s *Bridge) Err(ctx context.Context, digest string, err error) error
• type Cathode
• type Decoder
• type Dequeuer
  • func NewDequeuer(cathode Cathode, config swarm.Config) *Dequeuer
  • func (k *Dequeuer) Await()
  • func (k *Dequeuer) Close()
• type Emitter
• type Encoder
• type Enqueuer
  • func NewEnqueuer(emitter Emitter, config swarm.Config) *Enqueuer
  • func (k *Enqueuer) Await()
  • func (k *Enqueuer) Close()
• type Kernel
  • func New(enqueuer *Enqueuer, dequeuer *Dequeuer) *Kernel
  • func (k *Kernel) Await()
  • func (k *Kernel) Close()
• type Router

Functions

func Dequeue

func Dequeue[T any](k *Dequeuer, cat string, codec Decoder[T]) (<-chan swarm.Msg[T], chan<- swarm.Msg[T])

Dequeue creates pair of channels within kernel to enqueue messages

func Enqueue

func Enqueue[T any](k *Enqueuer, cat string, codec Encoder[T]) (chan<- T, <-chan T)

Enqueue creates pair of channels within kernel to enqueue messages

Types

type Bridge

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

Bridge Lambda's main function to Cathode interface

func NewBridge

func NewBridge(timeToFlight time.Duration) *Bridge

func (*Bridge) Ack

func (s *Bridge) Ack(ctx context.Context, digest string) error

Acknowledge processed message, allowing lambda handler progress

func (*Bridge) Ask

func (s *Bridge) Ask(ctx context.Context) ([]swarm.Bag, error)

Ask converts input of Lambda handler to the context of the kernel

func (*Bridge) Dispatch

func (s *Bridge) Dispatch(seq []swarm.Bag) error

Dispatch the batch of messages in the context of Lambda handler.

lambda.Start(
	func(evt events.CloudWatchEvent) error {
		...
		bridge.Dispatch(bag)
	}
)

func (*Bridge) Err

func (s *Bridge) Err(ctx context.Context, digest string, err error) error

Acknowledge error, allowing lambda handler progress

type Cathode

type Cathode interface {
	Ack(ctx context.Context, digest string) error
	Err(ctx context.Context, digest string, err error) error
	Ask(ctx context.Context) ([]swarm.Bag, error)
}

Cathode defines on-the-wire protocol for swarm.Bag, covering the ingress.

type Decoder

type Decoder[T any] interface{ Decode([]byte) (T, error) }

Decode message from wire format

type Dequeuer

type Dequeuer struct {
	sync.WaitGroup
	sync.RWMutex

	// Kernel configuration
	Config swarm.Config

	// Cathode is the reader port on message broker
	Cathode Cathode
	// contains filtered or unexported fields
}

func NewDequeuer

func NewDequeuer(cathode Cathode, config swarm.Config) *Dequeuer

Creates instance of broker reader

func (*Dequeuer) Await

func (k *Dequeuer) Await()

Await reader to complete

func (*Dequeuer) Close

func (k *Dequeuer) Close()

Closes broker reader, gracefully shutdowns all I/O

type Emitter

type Emitter interface {
	Enq(context.Context, swarm.Bag) error
}

Emitter defines on-the-wire protocol for swarm.Bag, covering egress.

type Encoder

type Encoder[T any] interface{ Encode(T) ([]byte, error) }

Encodes message into wire format

type Enqueuer

type Enqueuer struct {
	sync.WaitGroup

	// Kernel configuration
	Config swarm.Config

	// Emitter is the writer port on message broker
	Emitter Emitter
	// contains filtered or unexported fields
}

Messaging Egress port

func NewEnqueuer

func NewEnqueuer(emitter Emitter, config swarm.Config) *Enqueuer

Creates instance of broker writer

func (*Enqueuer) Await

func (k *Enqueuer) Await()

Await enqueue

func (*Enqueuer) Close

func (k *Enqueuer) Close()

Close enqueuer

type Kernel

type Kernel struct {
	*Enqueuer
	*Dequeuer
}

func New

func New(enqueuer *Enqueuer, dequeuer *Dequeuer) *Kernel

func (*Kernel) Await

func (k *Kernel) Await()

func (*Kernel) Close

func (k *Kernel) Close()

type Router

type Router = interface {
	Route(context.Context, swarm.Bag) error
}