Go Command Bus
A command bus to demand all the things.
go get github.com/io-da/command
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.
Commands are any type that implements the Command interface. Ideally they should contain immutable data.
It is also possible to provide a closure to the bus.
type Identifier int64
type Command interface {
Identifier() Identifier
}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 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 are any type that implements the Middleware interface.
type Middleware interface {
Handle(cmd Command, next Next) (any, error)
}Middlewares are optional and provided to the Bus using the bus.SetMiddlewares function.
A Middleware handles every command before it is provided to its respective handler.
Middlewares may be used to modify or completely prevent the command execution.
Middlewares are responsible for the execution of the next step:
func (Middleware) Handle(cmd Command, next Next) (any, error) {
// do something before...
data, err := next(cmd)
// do something after...
return data, err
}The order in which the middlewares are provided to the Bus is always respected.
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.
The Bus provides multiple ways to handle commands.
The bus processes the command immediately and returns the result from the handler.
data, err := bus.Handle(&FooBar{})
The bus processes the command using workers. It is no-blocking.
It is possible however to Await for the command to finish being processed.as, _ := bus.HandleAsync(&FooBar{}) // do something data, err := as.Await()
The bus processes the provided commands using workers. It is no-blocking.
It is possible however to Await for these commands to finish being processed. The return type is a *AsyncList. This struct exposes a few methods:
- Await waits for all the commands to finish processing similarly to a regular *Async type. However, it returns
[]any, error. The order of the returned results matches the order of the provided commands. The error joins all the resulting errors.- AwaitIterator returns an iterator to process the results in the order they are handled. This means that if multiple commands are issued, the fastest ones will be received first. The value that is returned by each iteration is a AsyncResult. Aside from the Data and Error, this struct also exposes the Index matching the order of the issued commands.
asl, _ := bus.HandleAsyncList(&FooBar{}, &FooBar2{}, &FooBar3{}) // do something data, err := asl.Await()
The bus also accepts closure commands to be provided.
These will be handled similarly to any other command. Using the type Closure.as, _ := bus.HandleAsync(Closure(func() (data any, err error) { return "foo bar", nil })) // do something data, err := as.Await()
The bus will use a schedule processor to handle the provided command according to a *Schedule struct.
More information about *Schedule can be found here.uuid, err := bus.Schedule(&FooBar{}, schedule.At(time.Now())) // if the scheduled command needs to be removed during runtime. bus.RemoveScheduled(uuid)
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.
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.
Below is a list of errors that can occur when calling bus.Initialize, bus.Handle, bus.HandleAsync and bus.Schedule.
command.InvalidCommandError
command.BusNotInitializedError
command.BusIsShuttingDownError
command.OneHandlerPerCommandError
command.HandlerNotFoundError
command.EmptyAwaitListError
command.InvalidClosureCommandErrorSince 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.
All the benchmarks are performed with command handlers calculating the fibonacci of 100.
CPU: Apple M3 Pro
| Benchmark Type | Time |
|---|---|
| Sync Commands | 660 ns/op |
| Async Commands | 425 ns/op |
For reference, here is the benchmark of the function used (fibonacci of 100) for simulation purposes, when run directly:
| Benchmark Type | Time |
|---|---|
| fibonacci(100) | 643 ns/op |
These results imply that the infrastructure induces a small overhead of about 10-20ns.
Naturally, the reason for the async's method improved performance is the fact that the executions of the fibonacci function are being spread over multiple workers.
An optional constants list of Command identifiers (idiomatic enum) for consistency
const (
FooCommand Identifier = "FooCommand"
BarCommand Identifier = "BarCommand"
)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
}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
}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)
}The change log is generated using the tool git-chglog.
Thanks to their amazing work, the CHANGELOG.md can be easily regenerated using the following command:
git-chglog -o CHANGELOG.mdPull 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.