swarm

README

swarm

Go channels for distributed queueing and event-driven systems

---

Today's wrong abstractions lead to complexity on maintainability in the future. Usage of synchronous interfaces to reflect asynchronous nature of messaging queues is a good example of inaccurate abstraction. Usage of pure Go channels is a proper solution to distills asynchronous semantic of queueing systems into the idiomatic native Golang code.

Inspiration

The library encourages developers to use Golang struct for asynchronous communication with peers. It helps engineers to define domain models, write correct, maintainable code. This library (swarm) uses generic programming style to abstract queueing systems into the idiomatic Golang channels chan<- T and <-chan T. See the design pattern Golang channels for distributed event-driven architecture to learn philosophy and use-cases:

1. readability: application uses pure Go code instead of vendor specific interfaces (learning time)
2. portability: application is portable between various queuing systems or event brokers in same manner as sockets abstracts networking stacks (exchange queueing transport "on-the-fly" to resolve evolution of requirements)
3. testability: unit testing focuses on pure biz logic, simplify dependency injections and mocking (pure unit tests).
4. distribution: idiomatic architecture to build distributed topologies and scale-out Golang applications (clustering).
5. serverless: of-the-shelf portable patterns for serverless applications (infrastructure as a code, aws cdk).

Getting started

The library requires Go 1.18 or later due to usage of generics.

The latest version of the library is available at main branch of this repository. All development, including new features and bug fixes, take place on the main branch using forking and pull requests as described in contribution guidelines. The stable version is available via Golang modules.

Use go get to retrieve the library and add it as dependency to your application.

go get -u github.com/fogfish/swarm

• Getting Started
  • Produce (enqueue) messages
  • Consume (dequeue) messages
  • Configure library behavior
  • Message Delivery Guarantees
  • Delayed Guarantee vs Guarantee
  • Order of Messages
  • Octet streams
  • Generic events
  • Error handling
  • Internal channel architecture
  • Serverless
  • Supported queuing system and event brokers

Produce (enqueue) messages

Please see and try examples. Its cover all basic use-cases with runnable code snippets, check the design pattern Distributed event-driven Golang channels for deep-dive into library philosophy.

The following code snippet shows a typical flow of producing the messages using the library.

import (
  "github.com/fogfish/swarm/broker/sqs"
  "github.com/fogfish/swarm/queue"
)

// Use pure Golang struct to define semantic of messages and events
type Note struct {
  ID   string `json:"id"`
  Text string `json:"text"`
}

// Spawn a new instance of the messaging broker
q := swarm.Must(sqs.New("name-of-the-queue"), /* config options */)

// creates pair Golang channels dedicated for publishing
// messages of type Note through the messaging broker. The first channel
// is dedicated to emit messages. The second one is the dead letter queue that
// contains failed transmissions. 
enq, dlq := queue.Enqueue[Note](q)

// Enqueue message of type Note
enq <- Note{ID: "note", Text: "some text"}

// Close the broker and release all resources
q.Close()

Consume (dequeue) messages

Please see and try examples. Its cover all basic use-cases with runnable code snippets, check the design pattern Distributed event-driven Golang channels for deep-dive into library philosophy.

The following code snippet shows a typical flow of consuming the messages using the library.

import (
  "github.com/fogfish/swarm/broker/sqs"
  "github.com/fogfish/swarm/queue"
)

// Use pure Golang struct to define semantic of messages and events
type Note struct {
  ID   string `json:"id"`
  Text string `json:"text"`
}

// Spawn a new instance of the messaging broker
q := swarm.Must(sqs.New("name-of-the-queue", /* config options */))

// Create pair Golang channels dedicated for consuming
// messages of type Note from the messaging broker. The first channel
// is dedicated to receive messages. The second one is the channel to
// acknowledge consumption  
deq, ack := queue.Dequeue[Note](q)

// consume messages and then acknowledge it
for msg := range deq {
  ack <- msg
}

// Await messages from the broker
q.Await()

Configure library behavior

The library uses "option pattern" for the configuration. See all available configuration options, which are passed into the broker. Please note that each configuration options has With prefix:

q, err := sqs.New("name-of-the-queue",
  swarm.WithSource("name-of-my-component"),
  swarm.WithRetryConstant(10 * time.Millisecond, 3),
  swarm.WithPollFrequency(10 * time.Second),
  /* ... */
)

Message Delivery Guarantees

Usage of Golang channels as an abstraction raises a concern about grade of service on the message delivery guarantees. The library ensures exactly same grade of service as the underlying queueing system or event broker. Messages are delivered according to the promise once they are accepted by the remote side of queuing system. The library's built-in retry logic protects losses from temporary unavailability of the remote peer. However, Golang channels are sophisticated "in-memory buffers", which introduce a lag of few milliseconds between scheduling a message to the channel and dispatching message to the remote peer. Use one of the following policy to either accept or protect from the loss all the in-the-flight messages in case of catastrophic failures.

At Most Once is best effort policy, where a message is published without any formal acknowledgement of receipt, and it isn't replayed. Some messages can be lost as subscribers are not required to acknowledge receipt.

The library implements asymmetric approaches. The enqueue path uses buffered Golang channels for emitter and dead-letter queues. The dequeue path also uses buffered Golang channels for delivery message to consumer. The messages are automatically acknowledged to the broker upon successful scheduling. This means that information will be lost if the consumer crashes before it has finished processing the message.

// Spawn a new instance of the messaging broker using At Most Once policy.
// The policy defines the capacity of Golang channel.
q, err := sqs.New("name-of-the-queue",
  swarm.WithPolicyAtMostOnce(1000),
)

// for compatibility reasons two channels are returned on the enqueue path but
// dead-letter-queue is nil
enq, dlq := queue.Enqueue[Note](q)

// for compatibility reasons two channels are returned on the dequeue path but
// ack channel acts as /dev/null discards any sent message
deq, ack := queue.Dequeue[Note](q)

At Least Once is the default policy used by the library. The policy assume usage of "acknowledgement" protocol, which guarantees a message will be re-sent until it is formally acknowledged by a recipient. Messages should never be lost but it might be delivered more than once causing duplicate work to consumer.

The library implements also asymmetric approaches. The enqueue path uses unbuffered Golang channels to emit messages and handle dead-letter queue, which leads to a delayed guarantee. The delayed guarantee in this context implies that enqueueing of other messages is blocked until dead-letter queue is resolved. Alternatively, the application can use synchronous protocol to enqueue message. The dequeue path also uses unbuffered Golang channels for delivery message to consumer and acknowledge its processing. The acknowledgement of message by consumer guarantee reliable delivery of the message but might cause duplicates.

// Spawn a new instance of the messaging broker using At Least Once policy.
// At Least Once policy is the default one, no needs to explicitly declare it.
// Use it only if you need to define other capacity for dequeue channel than
// the default one, which creates unbuffered channel
q, err := sqs.New("name-of-the-queue",
  swarm.WithPolicyAtLeastOnce(1000),
)

// all channels behaves as specified earlier.
enq, dlq := queue.Enqueue[Note](q)
deq, ack := queue.Dequeue[Note](q)

Exactly Once is not supported by the library yet.

Delayed Guarantee vs Guarantee

Usage of "At Least Once" policy (unbuffered channels) provides the delayed guarantee for producers. Let's consider the following example. If queue broker fails to send message A then the channel enq is blocked at sending message B until the program consumes message A from the dead-letter queue channel.

enq, dlq := queue.Enqueue[*User](q)

enq <- &User{ID: "A", Text: "some text by A"} // failed to send
enq <- &User{ID: "B", Text: "some text by B"} // blocked until dlq is processed 
enq <- &User{ID: "C", Text: "some text by C"}

The delayed guarantee is efficient on batch processing, pipelining but might cause complication at transactional processing. Therefore, the library also support a synchronous variant to producing a message:

// Creates "synchronous" variant of the queue
user := queue.New[User](q)

// Synchronously enqueue the message. It ensure that message is scheduled for
// delivery to remote peer once function successfully returns.
if err := user.Enqueue(&User{ID: "A", Text: "some text by A"}); err != nil {
  // handle error
}

Order of Messages

The library guarantee ordering of the messages when they are produced over same Golang channel. Let's consider a following example:

user, _ := queue.Enqueue[*User](q)
note, _ := queue.Enqueue[*Note](q)

user <- &User{ID: "A", Text: "some text by A"}
note <- &Note{ID: "B", Text: "some note A"}
user <- &User{ID: "C", Text: "some text by A"}

The library guarantees following clauses A before C and C after A because both messages are produced to single channel note. It do not guarantee clauses A before B, B before C or C after B because multiple channels are used.

Octet Streams

The library support slices of bytes []byte as message type. It opens an opportunity for the many encoding options like JSON, Gob, etc.

import (
  queue "github.com/fogfish/swarm/queue/bytes"
)

enq, dlq := queue.Enqueue(q, "Note")
deq, ack := queue.Dequeue(q, "Note")

Please see example about binary consumer/producer.

Generic events

Event defines immutable fact(s) placed into the queueing system. Event resembles the concept of Action as it is defined by schema.org.

An action performed by a direct agent and indirect participants upon a direct object.

This type supports development of event-driven solutions that treat data as a collection of immutable facts, which are queried and processed in real-time. These applications processes logical log of events, each event defines a change to current state of the object, i.e. which attributes were inserted, updated or deleted (a kind of diff). The event identifies the object that was changed together with using unique identifier.

The library support this concept through generic type swarm.Event[T] using the Higher-Kinded Type Classes abstraction. This abstraction allows to "overload" well-defined behavior of swarm.Event[T] with application specific type:

import (
  "github.com/fogfish/swarm"
  queue "github.com/fogfish/swarm/queue/events"
)

// declares the application specific event type.
type EventCreateNote swarm.Event[*Note]

func (EventCreateNote) HKT1(swarm.EventType) {}
func (EventCreateNote) HKT2(*Note)           {}

// creates Golang channels to produce / consume messages
enq, dlq := queue.Enqueue[*Note, EventCreateNote](q)
deq, ack := queue.Dequeue[*Note, EventCreateNote](q)

Please see example about event-driven consumer/producer.

Error Handling

The error handling on channel level is governed either by dead-letter queue or acknowledge protocol. The library provides swarm.WithStdErr configuration option to pass the side channel to consume global errors. Use it as top level error handler.

stderr := make(chan error)
q := queue.Must(sqs.New("swarm-test", swarm.WithStdErr(stderr)))

for err := range stderr {
  // error handling loop
}

Internal channel architecture

Serverless

The library support development of serverless event-driven application using AWS service. The library provides AWS CDK Golang constructs to spawn consumers. See example of serverless consumer and corresponding AWS CDK application.

package main

import (
  "github.com/fogfish/scud"
  "github.com/fogfish/swarm/queue/eventbridge"
)

func main() {
  app := eventbridge.NewServerlessApp()

  stack := app.NewStack("swarm-example-eventbridge")
  stack.NewEventBus()

  stack.NewSink(
    &eventbridge.SinkProps{
      Source: []string{"swarm-example-eventbridge"},
      Lambda: &scud.FunctionGoProps{
        SourceCodePackage: "github.com/fogfish/swarm",
        SourceCodeLambda:  "examples/eventbridge/dequeue",
      },
    },
  )

  app.Synth(nil)
}

Note: AWS Event Bridge has a feature that allows to match execution of consumer to the pattern of JSON object. Use swarm.Event[T] type to build reliable matching of incoming events:

/*
enq <- &swarm.Event[*User]{
  Agent:  "swarm:example",
  Participant: "user",
  Object: &User{ID: "user", Text: "some text"},
}
*/

stack.NewSink(
  &eventbridge.SinkProps{
    Pattern: map[string]interface{}{
      "@type": []string{"[user:Event[*swarm.User]]"},
      "agent": []string{"[swarm:example]"},
      "participant": []string{"[user]"},
    },
    /* ... */
  },
)

Supported queuing system and event brokers

• AWS EventBridge (serverless only)
  • produce message
  • consume message using aws lambda
  • aws cdk construct
• AWS SQS (serverless)
  • produce message
  • consume message using aws lambda
  • aws cdk construct
• AWS SQS
  • produce message
  • consume message
• AWS SNS
  • produce message
• AWS Kinesis (serverless)
  • produce message
  • consume message using aws lambda
• AWS Kinesis
  • produce message
  • consume message
• Redis
  • produce message
  • consume message
• MQTT API
  • produce message
  • consume message

Please let us know via GitHub issues your needs about queuing technologies.

How To Contribute

The library is Apache Version 2.0 licensed and accepts contributions via GitHub pull requests:

1. Fork it
2. Create your feature branch (git checkout -b my-new-feature)
3. Commit your changes (git commit -am 'Added some feature')
4. Push to the branch (git push origin my-new-feature)
5. Create new Pull Request

The build and testing process requires Go version 1.16 or later.

build and test library.

git clone https://github.com/fogfish/swarm
cd swarm
go test ./...

commit message

The commit message helps us to write a good release note, speed-up review process. The message should address two question what changed and why. The project follows the template defined by chapter Contributing to a Project of Git book.

bugs

If you experience any issues with the library, please let us know via GitHub issues. We appreciate detailed and accurate reports that help us to identity and replicate the issue.

benchmarking

cd queue/sqs
go test -run=^$ -bench=. -benchtime 100x

Bring Your Own Queue

TBD

License

Documentation

Index

• type Bag
• type Broker
• type Channel
• type Channels
  • func NewChannels() *Channels
  • func (chs *Channels) Attach(id string, ch Channel)
  • func (chs *Channels) Close()
  • func (chs *Channels) Length() int
  • func (chs *Channels) Sync()
• type Config
  • func NewConfig() Config
• type Dequeue
• type Enqueue
• type Event
  • func (Event[T]) HKT1(EventType)
  • func (Event[T]) HKT2(T)
• type EventKind
• type EventType
• type EvtDeqCh
  • func NewEvtDeqCh[T any, E EventKind[T]](n int) EvtDeqCh[T, E]
  • func (ch *EvtDeqCh[T, E]) Close()
  • func (ch *EvtDeqCh[T, E]) Sync()
• type EvtEnqCh
  • func NewEvtEnqCh[T any, E EventKind[T]](n int) EvtEnqCh[T, E]
  • func (ch *EvtEnqCh[T, E]) Close()
  • func (ch *EvtEnqCh[T, E]) Sync()
• type Msg
• type MsgDeqCh
  • func NewMsgDeqCh[T any](n int) MsgDeqCh[T]
  • func (ch *MsgDeqCh[T]) Close()
  • func (ch *MsgDeqCh[T]) Sync()
• type MsgEnqCh
  • func NewMsgEnqCh[T any](n int) MsgEnqCh[T]
  • func (ch *MsgEnqCh[T]) Close()
  • func (ch *MsgEnqCh[T]) Sync()
• type Option
  • func WithHookCommit(hook func()) Option
  • func WithNetworkTimeout(t time.Duration) Option
  • func WithPolicyAtLeastOnce(n int) Option
  • func WithPolicyAtMostOnce(n int) Option
  • func WithPollFrequency(t time.Duration) Option
  • func WithRetry(backoff Retry) Option
  • func WithRetryConstant(t time.Duration, n int) Option
  • func WithRetryExponential(t time.Duration, n int, f float64) Option
  • func WithRetryLinear(t time.Duration, n int) Option
  • func WithRetryNo() Option
  • func WithService(service any) Option
  • func WithSource(agent string) Option
  • func WithStdErr(stderr chan<- error) Option
  • func WithTimeToFlight(t time.Duration) Option
• type Policy
• type Retry

Types

type Bag

type Bag struct {
	Category string
	Event    any
	Object   []byte
	Digest   string
}

Bag is an abstract container for octet stream. Bag is used by the transport to abstract message on the wire.

type Broker

type Broker interface {
	Config() Config
	Close()
	DSync()
	Await()
	Enqueue(string, Channel) Enqueue
	Dequeue(string, Channel) Dequeue
}

type Channel

type Channel interface {
	Sync()
	Close()
}

Channel is abstract concept of channel(s)

type Channels

type Channels struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Channels

func NewChannels

func NewChannels() *Channels

func (*Channels) Attach

func (chs *Channels) Attach(id string, ch Channel)

func (*Channels) Close

func (chs *Channels) Close()

func (*Channels) Length

func (chs *Channels) Length() int

func (*Channels) Sync

func (chs *Channels) Sync()

type Config

type Config struct {
	// Instance of AWS Service, ...
	Service any

	// Source is a direct performer of the event.
	// A software service that emits action to the stream.
	Source string

	// Quality of Service Policy
	Policy Policy

	// Queue capacity
	EnqueueCapacity int
	DequeueCapacity int

	// Retry Policy for service calls
	Backoff Retry

	// Standard Error I/O channel
	StdErr chan<- error

	// Frequency to poll broker api
	PollFrequency time.Duration

	// Time To Flight is a time required by the client to acknowledge the message
	TimeToFlight time.Duration

	// Timeout for any network operations
	NetworkTimeout time.Duration

	// Commit hook (executed after each loop iteration)
	HookCommit func()
}

func NewConfig

func NewConfig() Config

type Dequeue

type Dequeue interface {
	Deq(string) (Bag, error)
	Ack(Bag) error
}

type Enqueue

type Enqueue interface {
	Enq(Bag) error
}

type Event

type Event[T any] struct {
	//
	// Unique identity for event
	// It is automatically defined by the library upon the transmission
	ID string `json:"@id,omitempty"`

	//
	// Canonical IRI that defines a type of action.
	// It is automatically defined by the library upon the transmission
	Type curie.IRI `json:"@type,omitempty"`

	//
	// Direct performer of the event, a software service that emits action to the stream.
	Agent curie.IRI `json:"agent,omitempty"`

	//
	// Indirect participants, a user who initiated an event.
	Participant curie.IRI `json:"participant,omitempty"`

	//
	// ISO8601 timestamps when action has been created
	// It is automatically defined by the library upon the transmission
	Created string `json:"created,omitempty"`

	//
	// The digest of received event (used internally to ack processing)
	Digest string `json:"-"`

	//
	// The object upon which the event is carried out.
	Object T `json:"object,omitempty"`
}

Event defines immutable fact(s) placed into the queueing system. Event resembles the concept of Action as it is defined by schema.org.

> An action performed by a direct agent and indirect participants upon a direct object.

This type supports development of event-driven solutions that treat data as a collection of immutable facts, which are queried and processed in real-time. These applications processes logical log of events, each event defines a change to current state of the object, i.e. which attributes were inserted, updated or deleted (a kind of diff). The event identifies the object that was changed together with using unique identifier.

func (Event[T]) HKT1

func (Event[T]) HKT1(EventType)

func (Event[T]) HKT2

func (Event[T]) HKT2(T)

type EventKind

type EventKind[A any] pure.HKT[EventType, A]

type EventType

type EventType any

type EvtDeqCh

type EvtDeqCh[T any, E EventKind[T]] struct {
	Msg chan *E // channel to recv message
	Ack chan *E // channel to send acknowledgement
}

msgRecv is the pair of channel, exposed by the queue to clients to recv messages

func NewEvtDeqCh

func NewEvtDeqCh[T any, E EventKind[T]](n int) EvtDeqCh[T, E]

func (*EvtDeqCh[T, E]) Close

func (ch *EvtDeqCh[T, E]) Close()

func (*EvtDeqCh[T, E]) Sync

func (ch *EvtDeqCh[T, E]) Sync()

type EvtEnqCh

type EvtEnqCh[T any, E EventKind[T]] struct {
	Msg  chan *E // channel to send message out
	Err  chan *E // channel to recv failed messages
	Busy sync.Mutex
}

EvtEnqCh is the pair of channel, exposed by the queue to clients to send messages

func NewEvtEnqCh

func NewEvtEnqCh[T any, E EventKind[T]](n int) EvtEnqCh[T, E]

func (*EvtEnqCh[T, E]) Close

func (ch *EvtEnqCh[T, E]) Close()

func (*EvtEnqCh[T, E]) Sync

func (ch *EvtEnqCh[T, E]) Sync()

type Msg

type Msg[T any] struct {
	Object T
	Digest string
}

Msg is a generic envelop type for incoming messages. It contains both decoded object and its digest used to acknowledge message.

type MsgDeqCh

type MsgDeqCh[T any] struct {
	Msg chan *Msg[T] // channel to recv message
	Ack chan *Msg[T] // channel to send acknowledgement
}

msgRecv is the pair of channel, exposed by the queue to clients to recv messages

func NewMsgDeqCh

func NewMsgDeqCh[T any](n int) MsgDeqCh[T]

func (*MsgDeqCh[T]) Close

func (ch *MsgDeqCh[T]) Close()

func (*MsgDeqCh[T]) Sync

func (ch *MsgDeqCh[T]) Sync()

type MsgEnqCh

type MsgEnqCh[T any] struct {
	Msg  chan T // channel to send message out
	Err  chan T // channel to recv failed messages
	Busy sync.Mutex
}

MsgEnqCh is the pair of channel, exposed by the queue for enqueuing the messages

func NewMsgEnqCh

func NewMsgEnqCh[T any](n int) MsgEnqCh[T]

func (*MsgEnqCh[T]) Close

func (ch *MsgEnqCh[T]) Close()

func (*MsgEnqCh[T]) Sync

func (ch *MsgEnqCh[T]) Sync()

type Option

type Option func(conf *Config)

Configuration option for queueing broker

func WithHookCommit

func WithHookCommit(hook func()) Option

func WithNetworkTimeout

func WithNetworkTimeout(t time.Duration) Option

Timeout for Network I/O

func WithPolicyAtLeastOnce

func WithPolicyAtLeastOnce(n int) Option

AtLeastOnce policy ensures delivery of the message to broker

The policy only impacts behavior of Golang channels created by the broker

func WithPolicyAtMostOnce

func WithPolicyAtMostOnce(n int) Option

AtMostOnce is best effort policy, where a message is published without any formal acknowledgement of receipt, and it isn't replayed.

The policy only impacts behavior of Golang channels created by the broker

func WithPollFrequency

func WithPollFrequency(t time.Duration) Option

Frequency to poll broker api

func WithRetry

func WithRetry(backoff Retry) Option

Custom retry policy

func WithRetryConstant

func WithRetryConstant(t time.Duration, n int) Option

Retry operation for N times, with T wait time in between

func WithRetryExponential

func WithRetryExponential(t time.Duration, n int, f float64) Option

Retry operation for N times, with exponential increments by T on each step

func WithRetryLinear

func WithRetryLinear(t time.Duration, n int) Option

Retry operation for N times, with linear increments by T on each step

func WithRetryNo

func WithRetryNo() Option

No retires

func WithService

func WithService(service any) Option

Configure AWS Service for broker instance

func WithSource

func WithSource(agent string) Option

Source is a direct performer of the event. A software service that emits action to the stream.

func WithStdErr

func WithStdErr(stderr chan<- error) Option

Configure Channel for global errors

func WithTimeToFlight

func WithTimeToFlight(t time.Duration) Option

Time To Flight for message from broker API to consumer

type Policy

type Policy int

Grade of Service Policy

const (
	PolicyAtMostOnce Policy = iota
	PolicyAtLeastOnce
	PolicyExactlyOnce
)

type Retry

type Retry interface {
	Retry(f func() error) error
}