queue

Documentation

Index

• type Adapter
  • func Adapt(sys swarm.System, service string, id string) *Adapter
  • func (q *Adapter) ConfIO(f func(msg *Msg) error) chan<- *Bag
  • func (q *Adapter) RecvIO(f func() (*Bag, error)) <-chan *Bag
  • func (q *Adapter) SendIO(f func(msg *Bag) error) chan<- *Bag
• type Bag
• type Msg
  • func (msg *Msg) Bytes() []byte
• type Policy
• type Queue
  • func New(sys swarm.System, id string, recv func() (<-chan *Bag, chan<- *Bag), ...) *Queue
  • func (q *Queue) Recv(cat swarm.Category) (<-chan swarm.Msg, chan<- swarm.Msg)
  • func (q *Queue) Send(cat swarm.Category) (chan<- swarm.Msg, <-chan swarm.Msg)
  • func (q *Queue) Sys() swarm.System

Types

type Adapter

type Adapter struct {
	System swarm.System
	ID     string

	Policy *Policy
	// contains filtered or unexported fields
}

Adapter to queueing systems

func Adapt

func Adapt(sys swarm.System, service string, id string) *Adapter

Adapt creates helper utility to adapt i/o with external queuing system

func (*Adapter) ConfIO

func (q *Adapter) ConfIO(f func(msg *Msg) error) chan<- *Bag

ConfIO create go routine to adapt async i/o over Golang channel to synchronous calls of queueing system interface

func (*Adapter) RecvIO

func (q *Adapter) RecvIO(f func() (*Bag, error)) <-chan *Bag

RecvIO create go routine to adapt async i/o over Golang channel to synchronous calls of queueing system interface

func (*Adapter) SendIO

func (q *Adapter) SendIO(f func(msg *Bag) error) chan<- *Bag

SendIO create go routine to adapt async i/o over Golang channel to synchronous calls of queueing system interface

type Bag

type Bag struct {
	// message attributes
	Target   string
	Source   string
	Category swarm.Category

	// message payload
	Object swarm.Msg

	//
	StdErr chan<- swarm.Msg
}

Bag is a product type of message and its attributes

type Msg

type Msg struct {
	Payload []byte
	Receipt string
}

Msg type defines external ingress message. It containers both payload and receipt to acknowledge

func (*Msg) Bytes

func (msg *Msg) Bytes() []byte

Bytes returns message payload (octet stream)

type Policy

type Policy struct {
	IO            backoff.Seq
	PollFrequency time.Duration
	TimeToFlight  time.Duration
}

queue system policy

type Queue

type Queue struct {
	sync.Mutex

	System swarm.System
	ID     string
	// contains filtered or unexported fields
}

Queue ...

func New

func New(
	sys swarm.System,
	id string,
	recv func() (<-chan *Bag, chan<- *Bag),
	send func() chan<- *Bag,
) *Queue

New creates

func (*Queue) Recv

func (q *Queue) Recv(cat swarm.Category) (<-chan swarm.Msg, chan<- swarm.Msg)

Recv creates endpoints to receive messages and acknowledge its consumption.

func (*Queue) Send

func (q *Queue) Send(cat swarm.Category) (chan<- swarm.Msg, <-chan swarm.Msg)

Send creates endpoints to send messages and receive errors.

func (*Queue) Sys

func (q *Queue) Sys() swarm.System