swarm

README

swarm

Go channels for queueing 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 queues into the idiomatic native Golang code.

Inspiration

See design pattern to learn how to improve:

1. readability: application uses pure Go code instead of vendor specific interfaces (learning time)
2. portability: application is portable between various queuing systems 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 (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.

The latest version of the library is available at main branch. 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.

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

go get -u github.com/fogfish/swarm

1. Import required package in your code

import (
  "github.com/fogfish/swarm/queue"
  "github.com/fogfish/swarm/queue/sqs" // <-- replace with your queue 
)

Quick Example

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

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

// Create instance of the client to queueing system, for example AWS SQS
sys := sqs.NewSystem("swarm-example-sqs")
q := sqs.Must(sqs.New(sys, "swarm-test"))

// Get Go channel to emit messages into queue and receive failed messages
enq, dlq := queue.Enqueue[Note](q)

// Get Go channel to recv messages from queue and acknowledge them
deq, ack := queue.Dequeue[Note](q)

// Spawn queue listeners. At this point the system spawns the transport
// routines so that channels are ready for the communications
if err := sys.Listen(); err != nil {
  panic(err)
}

// Enqueue message to the queue
enq <- Note{ID: "note", Text: "some text"}

// Dequeue message from queue
for msg := range deq {
  // do something with message and then acknowledge it
  ack <- msg
}

sys.Stop()

See examples folder for executable examples, code snippets for your projects and receipts to build serverless applications.

Getting started with serverless

See example of serverless event consumer. The library provides AWS CDK Golang constructs to spawn consumers.

package main

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

func main() {
  eventbridge.NewServerlessApp("swarm-example-eventbridge").
    CreateEventBus().
    CreateSink(
      &eventbridge.SinkProps{
        Queue: "swarm-test",
        Lambda: &scud.FunctionGoProps{
          SourceCodePackage: "github.com/fogfish/swarm",
          SourceCodeLambda:  "examples/eventbridge/recv",
        },
      },
    ).
    Synth(nil)
}

Supported queues

• AWS EventBridge
  • sending message
  • receiving message using aws lambda
  • aws cdk construct
• AWS SQS Serverless
  • sending message
  • receiving message using aws lambda
  • aws cdk construct
• AWS SQS
  • sending message
  • receiving message
• AWS SNS
  • sending message
• AWS Kinesis
  • sending message
  • receiving message through lambda handler
  • receiving message
• Redis
  • sending message
  • receiving message
• MQTT API
  • sending message
  • receiving 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

How To Abstract Queue

TBD

License

Documentation

Index

• type Bag
• type BagStdErr
• type Dequeue
• type Enqueue
• type Msg
• type Policy
  • func DefaultPolicy() *Policy
  • func NewPolicy() *Policy
  • func (p *Policy) WithBackoffIO(policy backoff.Seq) *Policy
  • func (p *Policy) WithPollFrequency(t time.Duration) *Policy
  • func (p *Policy) WithQueueCapacity(capacity int) *Policy
  • func (p *Policy) WithTimeToFlight(t time.Duration) *Policy
• type Queue
• type System

Types

type Bag

type Bag struct {
	System   string
	Queue    string
	Category string

	Object []byte
	Digest string
}

Bag type is an abstract container for octet stream, exchanged via channels. It is used by queuing transport to abstract message on the wire.

type BagStdErr

type BagStdErr struct {
	Bag
	StdErr func(error)
}

BagStdErr is an envelop used to enqueue messages to broker

type Dequeue

type Dequeue interface {
	// Listent activates transport
	Listen() error

	// Close transport
	Close() error

	// Deq returns a channel to dequeue messages
	Deq() chan *Bag

	// Ack returns a channel to acknowledge message processign
	Ack() chan *Bag
}

Dequeue is an abstraction of transport protocol to receive messages from queueing system (broker). Dequeue is an abstraction of transport client.

type Enqueue

type Enqueue interface {
	// Listent activates transport
	Listen() error

	// Close transport
	Close() error

	// Enq returns a channel to enqueue messages
	Enq() chan *BagStdErr

	// Enqueue synchronously message
	EnqSync(*Bag) error
}

Enqueue is an abstraction of transport protocols to send messages for the processing to queueing system (broker). Enqueue is an abstraction of transport client.

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 Policy

type Policy struct {
	BackoffIO     backoff.Seq
	PollFrequency time.Duration
	TimeToFlight  time.Duration
	QueueCapacity int
}

Policy defines behavior of queue

func DefaultPolicy

func DefaultPolicy() *Policy

default policy for the queue

func NewPolicy

func NewPolicy() *Policy

func (*Policy) WithBackoffIO

func (p *Policy) WithBackoffIO(policy backoff.Seq) *Policy

WithBackoffIO configures retry of queue I/O

swarm.NewPolicy().
  WithBackoffIO(backoff.Exp(...))

func (*Policy) WithPollFrequency

func (p *Policy) WithPollFrequency(t time.Duration) *Policy

WithPollFrequency configures frequency of polling loop

swarm.NewPolicy().
  WithPollFrequency(5*time.Seconds)

func (*Policy) WithQueueCapacity

func (p *Policy) WithQueueCapacity(capacity int) *Policy

func (*Policy) WithTimeToFlight

func (p *Policy) WithTimeToFlight(t time.Duration) *Policy

type Queue

type Queue interface {
	// Sync queue buffers
	Sync()
}

Queue ...

type System

type System interface {
	ID() string

	// Create new queueing endpoint from transport
	Queue(string, Enqueue, Dequeue, *Policy) Queue

	// Listent activates all transport
	Listen() error

	// Close system and all queues
	Close()

	// Wait for system to be stopped
	Wait()
}

System of Queues controls group related queues.