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 idiomatic 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 tech stack, Develop against an in-memory)
3. testability: (unit tests with no dependencies)

Getting started

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.

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

// spawn queue client
sys := swarm.New("test")
queue := swarm.Must(sqs.New(sys, "some-queue"))

// get Go channel to emit messages into queue
snd, _ := queue.Send("message-of-type-a")
snd <- swarm.Bytes("{\"type\": \"a\", \"some\": \"message\"}")

// get Go channel to recv messages from queue
rcv, ack := queue.Recv("message-of-type-a")
for msg := range rcv {
  // do something with message
  ack <- msg
}

sys.Stop()

See examples folder for executable examples and code snippets for your projects.

Supported queues

• Ephemeral (in-memory) built-in queue
  • sending/receiving message
• AWS EventBridge
  • sending message
  • receiving message through lambda handler
• AWS SQS
  • sending message
  • receiving message through lambda handler
  • 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 Bytes
  • func (b Bytes) Bytes() []byte
• type Category
• type Config
  • func WithContext(ctx context.Context) Config
• type Msg
• type Queue
  • func Must(q Queue, err error) Queue
• type System
  • func New(id string, opts ...Config) System

Types

type Bytes

type Bytes []byte

Bytes (octet stream) is a message to communicate via channels

func (Bytes) Bytes

func (b Bytes) Bytes() []byte

Bytes returns message payload (octet stream)

type Category

type Category string

Category of message (aka subject, topic)

type Config

type Config func(sys *system)

Config

func WithContext

func WithContext(ctx context.Context) Config

WithContext ...

type Msg

type Msg interface {
	Bytes() []byte
}

Msg type is an abstract container for octet stream exchange via channels

  ch <- swarm.Bytes("some message")
  ...
	for msg := range ch {
		msg.Bytes()
	}

type Queue

type Queue interface {
	/*
		Creates endpoints to receive messages and acknowledge its consumption.
	*/
	Recv(Category) (<-chan Msg, chan<- Msg)

	/*
		Creates endpoints to send messages and receive errors.
	*/
	Send(Category) (chan<- Msg, <-chan Msg)
}

Queue ...

func Must

func Must(q Queue, err error) Queue

Must ensures successful creation of queue

type System

type System interface {
	/*
		system ID
	*/
	ID() string

	/*
		spawn go routine in context of system
	*/
	Go(func(context.Context))

	/*
	 stop system and all active go routines
	*/
	Stop()

	/*
	 wait system
	*/
	Wait()
}

System ...

func New

func New(id string, opts ...Config) System

New creates new queueing system