command

README

Go Command Bus

A command bus to demand all the things.

Installation

go get github.com/io-da/command

Overview

1. Commands
2. Handlers
3. Error Handlers
4. The Bus
   1. Tweaking Performance
   2. Shutting Down
   3. Available Errors
   4. Scheduled Commands
5. Benchmarks
6. Examples

Introduction

This library is intended for anyone looking to trigger application commands in a decoupled architecture.
The Bus provides the option to use workers (goroutines) to attempt handling the commands in non-blocking manner.
Clean and simple codebase.

Getting Started

Commands

Commands are any type that implements the Command interface. Ideally they should contain immutable data.

type Identifier int64

type Command interface {
    Identifier() Identifier
}

Handlers

Handlers are any type that implements the Handler interface. Handlers must be instantiated and provided to the Bus on initialization.
The Bus is initialized using the function bus.Initialize. The Bus will then use the Identifier of the handlers to know which Command to process.

type Handler interface {
    Handle(cmd Command) (any, error)
    Handles() Identifier
}

Error Handlers

Error handlers are any type that implements the ErrorHandler interface. Error handlers are optional (but advised) and provided to the Bus using the bus.SetErrorHandlers function.

type ErrorHandler interface {
    Handle(cmd Command, err error)
}

Any time an error occurs within the bus, it will be passed on to the error handlers. This strategy can be used for decoupled error handling.

Middlewares

Middlewares are any type that implements the InwardMiddleware or the OutwardMiddleware interface.
Middlewares are optional and provided to the Bus using the bus.SetInwardMiddlewares and bus.SetOutwardMiddlewares functions.
An InwardMiddleware middleware handles every command before it is provided to its respective handler.
An OutwardMiddleware middleware handles every command that was successfully processed by its respective handler. These middlewares are also provided with the data or error returned by the command handler. Allowing potential data/error handling, such as transformations.

The order in which the middlewares are provided to the Bus is always respected. Additionally, if a middleware returns an error, it interrupts the flow and the command is no longer passed along to the next step.

type InwardMiddleware interface {
    HandleInward(cmd Command) error
}

type OutwardMiddleware interface {
    HandleOutward(cmd Command, data any, err error) error
}

The Bus

Bus is the struct that will be used to trigger all the application's commands.
The Bus should be instantiated and initialized on application startup. The initialization is separated from the instantiation for dependency injection purposes.
The application should instantiate the Bus once and then use it's reference to trigger all the commands.
There can only be one Handler per Command.

Tweaking Performance

The number of workers for async commands can be adjusted.

bus.SetWorkerPoolSize(10)

If used, this function must be called before the Bus is initialized. And it specifies the number of goroutines used to handle async commands.
In some scenarios increasing the value can drastically improve performance.
It defaults to the value returned by runtime.GOMAXPROCS(0).

The buffer size of the async commands queue can also be adjusted.
Depending on the use case, this value may greatly impact performance.

bus.SetQueueBuffer(100)

If used, this function must be called before the Bus is initialized.
It defaults to 100.

Shutting Down

The Bus also provides a shutdown function that attempts to gracefully stop the command bus and all its routines.

bus.Shutdown()

This function will block until the bus is fully stopped.

Available Errors

Below is a list of errors that can occur when calling bus.HandleAsync or bus.Handle.

command.InvalidCommandError
command.BusNotInitializedError
command.BusIsShuttingDownError
command.HandlerNotFoundError

Scheduled Commands

Since 1.2, the bus also has built in support for github.com/io-da/schedule.
Using bus.Schedule, one may schedule a command to be processed at certain times or even following a cron like pattern.

Benchmarks

All the benchmarks are performed against batches of 1 million commands.
All the benchmarks contain a small overhead due to the usage of sync.WaitGroup for testing purposes.
The command handlers are calculating the fibonacci of 1000 for simulation purposes.
CPU: 11th Gen Intel(R) Core(TM) i9-11950H @ 2.60GHz

Benchmark Type	Time
Sync Commands	15828 ns/op
Async Commands	2808 ns/op

Examples

An optional constants list of Command identifiers (idiomatic enum) for consistency

const (
   Unidentified Identifier = iota
   FooCommand
   BarCommand
)

Example Commands

A simple struct command.

type fooCommand struct {
    bar string
}
func (*fooCommand) Identifier() Identifier {
    return FooCommand
}

A string command.

type barCommand string
func (barCommand) Identifier() Identifier {
    return BarCommand
}

Example Handlers

A couple of empty respective handlers.

type fooHandler struct{}

func (hdl *fooHandler) Handles() Identifier {
    return FooCommand
}

func (hdl *fooHandler) Handle(cmd Command) (data any, err error) {
    // handle FooCommand
    return
}

type barHandler struct{}

func (hdl *barHandler) Handles() Identifier {
    return BarCommand
}

func (hdl *barHandler) Handle(cmd Command) (data any, err error) {
    // handle BarCommand
    return
}

Putting it together

Initialization and usage of the exemplified commands and handlers

import (
    "github.com/io-da/command"
)

func main() {
    // instantiate the bus (returns *command.Bus)
    bus := command.NewBus()
    
    // initialize the bus with all the application's command handlers
    bus.Initialize(
        &fooHandler{},
        &barHandler{},
    )
    
    // trigger commands!
    // sync
    bus.Handle(&fooCommand{})
    // async
    bus.HandleAsync(barCommand{})
    // async await
    res, _ := bus.HandleAsync(&fooCommand{}) 
    // do something
    res.Await()
    // scheduled to run every day
    sch := schedule.As(schedule.Cron().EveryDay())
    bus.Schedule(&fooCommand{}, sch)
}

Contributing

Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.

Please make sure to update tests as appropriate.

License

MIT

Documentation

Index

• Constants
• type Async
  • func (res *Async) Await() error
  • func (res *Async) Get() (any, error)
• type Bus
  • func NewBus() *Bus
  • func (bus *Bus) Handle(cmd Command) (any, error)
  • func (bus *Bus) HandleAsync(cmd Command) (*Async, error)
  • func (bus *Bus) Initialize(hdls ...Handler) error
  • func (bus *Bus) RemoveScheduled(keys ...uuid.UUID)
  • func (bus *Bus) Schedule(cmd Command, sch *schedule.Schedule) (*uuid.UUID, error)
  • func (bus *Bus) SetErrorHandlers(hdls ...ErrorHandler)
  • func (bus *Bus) SetInwardMiddlewares(mdls ...InwardMiddleware)
  • func (bus *Bus) SetOutwardMiddlewares(mdls ...OutwardMiddleware)
  • func (bus *Bus) SetQueueBuffer(queueBuffer int)
  • func (bus *Bus) SetWorkerPoolSize(workerPoolSize int)
  • func (bus *Bus) Shutdown()
• type BusError
  • func (e BusError) Error() string
• type Command
• type ErrorHandler
• type Handler
• type Identifier
• type InwardMiddleware
• type OutwardMiddleware

Constants

const (
	InvalidCommandError       = BusError("command: invalid command")
	BusNotInitializedError    = BusError("command: the bus is not initialized")
	BusIsShuttingDownError    = BusError("command: the bus is shutting down")
	OneHandlerPerCommandError = BusError("command: there can only be one handler per command")
	HandlerNotFoundError      = BusError("command: no handler found for the command provided")
)

Types

type Async

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

Async is the struct returned from async commands.

func (*Async) Await

func (res *Async) Await() error

Await for the data from the return of the command

func (*Async) Get

func (res *Async) Get() (any, error)

Get retrieves the data from the return of the first command.

type Bus

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

Bus is the only struct exported and required for the command bus usage. The Bus should be instantiated using the NewBus function.

func NewBus

func NewBus() *Bus

NewBus instantiates the Bus struct. The Initialization of the Bus is performed separately (Initialize function) for dependency injection purposes.

func (*Bus) Handle

func (bus *Bus) Handle(cmd Command) (any, error)

Handle the command synchronously.

func (*Bus) HandleAsync

func (bus *Bus) HandleAsync(cmd Command) (*Async, error)

HandleAsync the command using the workers asynchronously. It also returns an *Async struct which allows clients to optionally await for the command to be processed.

func (*Bus) Initialize

func (bus *Bus) Initialize(hdls ...Handler) error

Initialize the command bus by providing the list of handlers. There can only be one handler per command.

func (*Bus) RemoveScheduled

func (bus *Bus) RemoveScheduled(keys ...uuid.UUID)

RemoveScheduled removes previously scheduled commands.

func (*Bus) Schedule

func (bus *Bus) Schedule(cmd Command, sch *schedule.Schedule) (*uuid.UUID, error)

Schedule allows commands to be scheduled to be executed asynchronously.

func (*Bus) SetErrorHandlers

func (bus *Bus) SetErrorHandlers(hdls ...ErrorHandler)

SetErrorHandlers may optionally be used to provide a list of error handlers. They will receive any error thrown during the command process. Error handlers may only be provided *before* the bus is initialized.

func (*Bus) SetInwardMiddlewares

func (bus *Bus) SetInwardMiddlewares(mdls ...InwardMiddleware)

SetInwardMiddlewares may optionally be used to provide a list of inward middlewares. They will receive and process every command that is about to be handled. *The order the middlewares are provided is always respected*. Middlewares may only be provided *before* the bus is initialized.

func (*Bus) SetOutwardMiddlewares

func (bus *Bus) SetOutwardMiddlewares(mdls ...OutwardMiddleware)

SetOutwardMiddlewares may optionally be used to provide a list of outward middlewares. They will receive and process every command that was handled. *The order the middlewares are provided is always respected*. Middlewares may only be provided *before* the bus is initialized.

func (*Bus) SetQueueBuffer

func (bus *Bus) SetQueueBuffer(queueBuffer int)

SetQueueBuffer may optionally be used to tweak the buffer size of the async commands queue. This value may have high impact on performance depending on the use case. It can only be adjusted *before* the bus is initialized. It defaults to 100.

func (*Bus) SetWorkerPoolSize

func (bus *Bus) SetWorkerPoolSize(workerPoolSize int)

SetWorkerPoolSize may optionally be used to tweak the worker pool size for async commands. It can only be adjusted *before* the bus is initialized. It defaults to the value returned by runtime.GOMAXPROCS(0).

func (*Bus) Shutdown

func (bus *Bus) Shutdown()

Shutdown the command bus gracefully. *Async commands accessed while shutting down will be disregarded*.

type BusError

type BusError string

BusError is used to create errors originating from the command bus

func (BusError) Error

func (e BusError) Error() string

Error returns the string message of the error.

type Command

type Command interface {
	Identifier() Identifier
}

Command is the interface that must be implemented by any type to be considered a command.

type ErrorHandler

type ErrorHandler interface {
	Handle(cmd Command, err error)
}

ErrorHandler must be implemented for a type to qualify as an error handler.

type Handler

type Handler interface {
	Handle(cmd Command) (any, error)
	Handles() Identifier
}

Handler must be implemented for a type to qualify as a command handler.

type Identifier

type Identifier int64

Identifier is used to create a consistent identity solution for commands

type InwardMiddleware

type InwardMiddleware interface {
	HandleInward(cmd Command) error
}

InwardMiddleware must be implemented for a type to qualify as an inward command middleware. An inward middleware process a command before being provided to the respective command handler.

type OutwardMiddleware

type OutwardMiddleware interface {
	HandleOutward(cmd Command, data any, err error) (any, error)
}

OutwardMiddleware must be implemented for a type to qualify as an outward command middleware. An outward middleware process the command after being provided to the respective command handler.