watermill

README

Watermill

Watermill is a Go library for working efficiently with message streams. It is intended for building event driven applications, enabling event sourcing, RPC over messages, sagas and basically whatever else comes to your mind. You can use conventional pub/sub implementations like Kafka or RabbitMQ, but also HTTP or MySQL binlog if that fits your use case.

Note: Watermill should run reliably in a production environment, but it is still under heavy development and the public API may change before the 1.0.0 release.

Documentation: https://watermill.io/

Getting started guide: https://watermill.io/docs/getting-started/

Background

Building distributed and scalable services is rarely as easy as some may suggest. There is a lot of hidden knowledge that comes with writing such systems. Just like you don't need to know the whole TCP stack to create a HTTP REST server, you shouldn't need to study all of this knowledge to start with building message-driven applications.

Watermill's goal is to make communication with messages as easy to use as HTTP routers. It provides the tools needed to begin working with event-driven architecture and allows you to learn the details on the go.

At the heart of Watermill there is one simple interface:

func(*Message) ([]*Message, error)

Your handler receives a message and decides whether to publish new message(s) or return an error. What happens next is up to the middlewares you've chosen.

You can find more about our motivations in our Introducing Watermill blog post.

Features

• Easy to understand (see examples below).
• Universal - event-driven architecture, messaging, stream processing, CQRS - use it for whatever you need.
• Fast - (benchmarks coming soon)
• Flexible with middlewares and plugins.
• Resilient - using proven technologies and passing stress tests (results coming soon).

Pub/Subs

All publishers and subscribers have to implement an interface:

type Publisher interface {
	Publish(topic string, messages ...*Message) error
	Close() error
}

type Subscriber interface {
	Subscribe(topic string) (chan *Message, error)
	Close() error
}

All Pub/Subs implementation can be found in the documentation.

Examples

• Your first app - start here!
• Simple application with publisher and subscriber
• HTTP to Kafka
• ...and more!

Contributing

All contributions are very much welcome. If you'd like to help with Watermill development, please see open issues and submit your pull request via GitHub.

Support

If you didn't found answer to your question in the documentation, feel free to ask!

Please join us on the #watermill channel on the Gophers slack: You can get invite here.

Why the name?

It processes streams!

License

MIT License

Documentation

Overview

Watermill is a Golang library for working efficiently with message streams.

It is intended for building event driven applications, enabling event sourcing, RPC over messages, sagas and basically whatever else comes to your mind.

You can use conventional pub/sub implementations like Kafka or RabbitMQ, but also HTTP or MySQL binlog if that fits your use case.

Website with detailed documentation: https://watermill.io/

Getting started guide: https://watermill.io/docs/getting-started/

Index

• func NewShortUUID() string
• func NewULID() string
• func NewUUID() string
• type CaptureLoggerAdapter
  • func NewCaptureLogger() *CaptureLoggerAdapter
  • func (c *CaptureLoggerAdapter) Captured() map[LogLevel][]CapturedMessage
  • func (c *CaptureLoggerAdapter) Debug(msg string, fields LogFields)
  • func (c *CaptureLoggerAdapter) Error(msg string, err error, fields LogFields)
  • func (c *CaptureLoggerAdapter) Has(msg CapturedMessage) bool
  • func (c *CaptureLoggerAdapter) HasError(err error) bool
  • func (c *CaptureLoggerAdapter) Info(msg string, fields LogFields)
  • func (c *CaptureLoggerAdapter) Trace(msg string, fields LogFields)
  • func (c *CaptureLoggerAdapter) With(fields LogFields) LoggerAdapter
• type CapturedMessage
• type LogFields
  • func (l LogFields) Add(newFields LogFields) LogFields
  • func (l LogFields) Copy() LogFields
• type LogLevel
• type LoggerAdapter
  • func NewStdLogger(debug, trace bool) LoggerAdapter
  • func NewStdLoggerWithOut(out io.Writer, debug bool, trace bool) LoggerAdapter
• type NopLogger
  • func (NopLogger) Debug(msg string, fields LogFields)
  • func (NopLogger) Error(msg string, err error, fields LogFields)
  • func (NopLogger) Info(msg string, fields LogFields)
  • func (NopLogger) Trace(msg string, fields LogFields)
  • func (l NopLogger) With(fields LogFields) LoggerAdapter
• type StdLoggerAdapter
  • func (l *StdLoggerAdapter) Debug(msg string, fields LogFields)
  • func (l *StdLoggerAdapter) Error(msg string, err error, fields LogFields)
  • func (l *StdLoggerAdapter) Info(msg string, fields LogFields)
  • func (l *StdLoggerAdapter) Trace(msg string, fields LogFields)
  • func (l *StdLoggerAdapter) With(fields LogFields) LoggerAdapter

Functions

func NewShortUUID

func NewShortUUID() string

func NewULID

func NewULID() string

func NewUUID

func NewUUID() string

Types

type CaptureLoggerAdapter

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

func NewCaptureLogger

func NewCaptureLogger() *CaptureLoggerAdapter

func (*CaptureLoggerAdapter) Captured

func (c *CaptureLoggerAdapter) Captured() map[LogLevel][]CapturedMessage

func (*CaptureLoggerAdapter) Debug

func (c *CaptureLoggerAdapter) Debug(msg string, fields LogFields)

func (*CaptureLoggerAdapter) Error

func (c *CaptureLoggerAdapter) Error(msg string, err error, fields LogFields)

func (*CaptureLoggerAdapter) Has

func (c *CaptureLoggerAdapter) Has(msg CapturedMessage) bool

func (*CaptureLoggerAdapter) HasError

func (c *CaptureLoggerAdapter) HasError(err error) bool

func (*CaptureLoggerAdapter) Info

func (c *CaptureLoggerAdapter) Info(msg string, fields LogFields)

func (*CaptureLoggerAdapter) Trace

func (c *CaptureLoggerAdapter) Trace(msg string, fields LogFields)

func (*CaptureLoggerAdapter) With

func (c *CaptureLoggerAdapter) With(fields LogFields) LoggerAdapter

type CapturedMessage

type CapturedMessage struct {
	Level  LogLevel
	Fields LogFields
	Msg    string
	Err    error
}

type LogFields

type LogFields map[string]interface{}

func (LogFields) Add

func (l LogFields) Add(newFields LogFields) LogFields

func (LogFields) Copy

func (l LogFields) Copy() LogFields

type LogLevel

type LogLevel uint

const (
	TraceLogLevel LogLevel = iota + 1
	DebugLogLevel
	InfoLogLevel
	ErrorLogLevel
)

type LoggerAdapter

type LoggerAdapter interface {
	Error(msg string, err error, fields LogFields)
	Info(msg string, fields LogFields)
	Debug(msg string, fields LogFields)
	Trace(msg string, fields LogFields)
	With(fields LogFields) LoggerAdapter
}

func NewStdLogger

func NewStdLogger(debug, trace bool) LoggerAdapter

func NewStdLoggerWithOut

func NewStdLoggerWithOut(out io.Writer, debug bool, trace bool) LoggerAdapter

type NopLogger

type NopLogger struct{}

func (NopLogger) Debug

func (NopLogger) Debug(msg string, fields LogFields)

func (NopLogger) Error

func (NopLogger) Error(msg string, err error, fields LogFields)

func (NopLogger) Info

func (NopLogger) Info(msg string, fields LogFields)

func (NopLogger) Trace

func (NopLogger) Trace(msg string, fields LogFields)

func (NopLogger) With

func (l NopLogger) With(fields LogFields) LoggerAdapter

type StdLoggerAdapter

type StdLoggerAdapter struct {
	ErrorLogger *log.Logger
	InfoLogger  *log.Logger
	DebugLogger *log.Logger
	TraceLogger *log.Logger
	// contains filtered or unexported fields
}

func (*StdLoggerAdapter) Debug

func (l *StdLoggerAdapter) Debug(msg string, fields LogFields)

func (*StdLoggerAdapter) Error

func (l *StdLoggerAdapter) Error(msg string, err error, fields LogFields)

func (*StdLoggerAdapter) Info

func (l *StdLoggerAdapter) Info(msg string, fields LogFields)

func (*StdLoggerAdapter) Trace

func (l *StdLoggerAdapter) Trace(msg string, fields LogFields)

func (*StdLoggerAdapter) With

func (l *StdLoggerAdapter) With(fields LogFields) LoggerAdapter