go-bus

go-bus

go-bus is a transport agnostic service bus implementation written in go. go-bus aims to provide high-level messaging capabilities to your services with close to no configuration at all and enables your service architecture to asynchronously communicate in an event-driven way.

Build & Test:

Please note: This library is not production ready yet.

Please read the wiki for more information.

Getting started

Create a service bus endpoint with RabbitMQ transport

endpoint := servicebus.Create("awesomeService",
    rabbitmq.Create("amqp://guest:guest@localhost:5672/"))

You can then configure incoming messages you want to handle like this:

endpoint.Message("CreateUser").
    AsIncoming().
    Handle(createUserHandler)

func createUserHandler(ctx *servicebus.IncomingMessageContext) {
    fmt.Println("received CreateUser!")
    ctx.Ack()
}

By default messages need to be handled with acknowledgement.

ctx.Ack()       // Tell the server that the msg has been acknowledged. Message will be removed from queue.
ctx.Discard()   // Tell the server that the msg will not be acknowledged, but also do not requeue. Can be used in cases where you decide the msg is of no value to you.
ctx.Retry()     // Tell the server that the msg will not be acknowledged, but the msg will be requeued. This should be used in cases of transient errors like an unreachable service or database.
ctx.Fail()      // Tell the server that the msg will not be acknowledged and move the msg to the error queue. This should be used in cases of unrecoverable errors like for example schema or serialization issues.

In case of a request/response pattern, you may send a reply to the originator of the received msg with

ctx.Reply("YourReplyMessage", yourMessageObj)

You may bind the message payload to a matching struct

type User struct {
    UserName string
    Email    string
}

func createUserHandler(ctx *servicebus.IncomingMessageContext) {
    user := new(CreateUser)
    ctx.Bind(user)
    fmt.Println(user.UserName + " created!")
    ctx.Ack()
}

(De)-Serialization is currently done by using the standard go json lib.

Start the service bus endpoint to send and receive messages.

endpoint.Start()

You can then publish events to 0 or many subscribers...

endpoint.Publish("UserCreated", &UserCreated{
    UserName: "Name",
    Email: "Name@email.com",
})

...send commands to a single endpoint...

endpoint.Send("CreateUser", "SomeOtherAwesomeService", &CreateUser{
    UserName: "Name",
    Email: "Name@email.com",
})

...or send messages to the local endpoint

endpoint.SendLocal("CreateUser", &CreateUser{
    UserName: "Name",
    Email: "Name@email.com",
})

A simple local development example could finally look like this

type CreateUser struct {
    UserName string
    Email    string
}

func main() {
    endpoint := servicebus.Create("awesomeService",
        rabbitmq.Create("amqp://guest:guest@localhost:5672/"))

    endpoint.Message("CreateUser").
        AsIncoming().
        Handle(createUserHandler)

    err := endpoint.Start()
    if err != nil {
        panic(err)
    }

    err = endpoint.SendLocal("CreateUser", &CreateUser{
    	UserName: "Name",
    	Email: "Name@email.com",
    })
    if err != nil {
        fmt.Println(err)
    }
}

func createUserHandler(ctx *servicebus.IncomingMessageContext) {
    user := new(CreateUser)
    err := ctx.Bind(user)
    if err != nil {
        fmt.Println(err)
    }

    fmt.Println(user.UserName + " created!")
    
    ctx.Ack()
}