Introducing Watermill - Go event-driven applications library
Watermill is a Go library for working efficiently with message streams. It is intended as a library for building event-driven applications, enabling event sourcing, CQRS, RPC over messages, sagas.
Why?
Lack of standard messaging library
There are many third party and standard library tools which help to implement standardized RPC or HTTP communication in Golang. There are also multiple third party HTTP routers and frameworks.
But when you want to build a message-driven application, there are no libraries which are infrastructure-agnostic and not opinionated.
Watermill is a library which will work with any implementation of Publisher and Subscriber interfaces:
type Publisher interface {
Publish(topic string, messages ...*Message) error
Close() error
}
type Subscriber interface {
Subscribe(topic string) (chan *Message, error)
Close() error
}
For now, Watermill comes with the following implementations:
- Kafka Pub/Sub
- Go channel (intraprocess) Pub/Sub
- HTTP Subscriber
In close future, there are plans for RabbitMQ (AMQP) Pub/Sub, Google Cloud PubSub and MySQL Binlog subscriber.
High entry level
When implementing a HTTP REST API nowadays, how do you go about it? Do you send it via TCP like this:
if _, ok := w.(io.ByteWriter); !ok {
bw = bufio.NewWriter(w)
w = bw
}
_, err = fmt.Fprintf(w, "%s %s HTTP/1.1\r\n", valueOrDefault(r.Method, "GET"), ruri)
if err != nil {
return err
}
// Header lines
_, err = fmt.Fprintf(w, "Host: %s\r\n", host)
if err != nil {
return err
}
// [74 lines later]
err = tw.writeBody(w)
// [a lot of crazy stuff in writeBody]
or just by calling:
helloHandler := func(w http.ResponseWriter, req *http.Request) {
io.WriteString(w, "Hello, world!\n")
}
For the same reason, you probably shouldn’t use your PubSub’s raw driver (for example for Kafka). Instead of this, you should use something that will deal with stuff like acknowledgment, errors handling and publishing created messages in a universal and consistent manner. It is also a lot easier for less experienced developers.
Watermill works with high level handler functions, with abstraction level similar to Golang HTTP handlers:
type HandlerFunc func(msg *Message) ([]*Message, error)
All you need to do to process a message is to create a function with the HandlerFunc signature:
func process(msg *message.Message) ([]*message.Message, error) {
log.Println("received message: ", string(msg.Payload))
producedMessages := []*message.Message{
message.NewMessage(uuid.NewV4().String(), []byte(time.Now().String()),
}
return producedMessages, nil
}
Extensibility and elasticity
Every organization and project is different. This is why Watermill is built so that every part that may differ between organizations can be configured.
As said before, you can use any Pub/Sub or implement your own when needed. If you are using different message format, you can also inject your own message marshaller.
Nowadays, any modern service must support metrics and distributed tracing. When building a message-driven application, often we need a posion queue or retrying support. Also, sometimes we need to limit the rate of message processing with throttling.
Every one of these functionalities can be implemented by a middleware:
type HandlerMiddleware func(h HandlerFunc) HandlerFunc
Watermill is shipped with ready-to-use middlewares for these functionalities.
Also, error handling can be different for some services. By default, we want to send an Ack when the processing was successful. But sometimes, we want to send an Ack just after receiving the message. It can be also done by using a middleware:
func InstantAck(h message.HandlerFunc) message.HandlerFunc {
return func(message *message.Message) ([]*message.Message, error) {
message.Ack()
return h(message)
}
}
Fast
Use of message-driven architecture and Golang is often dictated by high performance requirements. For that reason, Watermill can process hundreds of thousands of messages per second with low latency. But there’s still a lot of work to be done in this field. Performance updates and benchmarks are on the way!
How to start?
In https://watermill.io/ you can find detailed documentation of Watermill. I recommend to start with the Getting Started guide.
Future plans
For now, Watermill is in beta phase. Currently, our number one priority is to create a good documentation.
We have also plans to implement some more components basing on our Messages Router and Pub/Sub:
As mentioned before, we are planning to implement RabbitMQ (AMQP) Pub/Sub, Google Cloud PubSub and MySQL Binlog subscriber.
If you need something that Watermill is missing, feel free to add a new issue. If one of the previously mentioned features is important for you, please add a thumbs up in GitHub issue.
