Golang Channel

A Pub/Sub implemented on Golang goroutines and channels.

Golang Channel

Full source: github.com/ThreeDotsLabs/watermill/pubsub/gochannel/pubsub.go

// ...
// GoChannel is the simplest Pub/Sub implementation.
// It is based on Golang's channels which are sent within the process.
//
// GoChannel has no global state,
// that means that you need to use the same instance for Publishing and Subscribing!
//
// When GoChannel is persistent, messages order is not guaranteed.
type GoChannel struct {
// ...

Characteristics

FeatureImplementsNote
ConsumerGroupsno
ExactlyOnceDeliveryyes
GuaranteedOrderyes
Persistentno

Configuration

You can inject configuration via the constructor.

Full source: github.com/ThreeDotsLabs/watermill/pubsub/gochannel/pubsub.go

// ...
func NewGoChannel(config Config, logger watermill.LoggerAdapter) *GoChannel {
    if logger == nil {
        logger = watermill.NopLogger{}
    }

    return &GoChannel{
        config: config,

        subscribers:            make(map[string][]*subscriber),
        subscribersByTopicLock: sync.Map{},
        logger: logger.With(watermill.LogFields{
// ...

Publishing

Full source: github.com/ThreeDotsLabs/watermill/pubsub/gochannel/pubsub.go

// ...
// Publish in GoChannel is NOT blocking until all consumers consume.
// Messages will be send in background.
//
// Messages may be persisted or not, depending of persistent attribute.
func (g *GoChannel) Publish(topic string, messages ...*message.Message) error {
// ...

Subscribing

Full source: github.com/ThreeDotsLabs/watermill/pubsub/gochannel/pubsub.go

// ...
// Subscribe returns channel to which all published messages are sent.
// Messages are not persisted. If there are no subscribers and message is produced it will be gone.
//
// There are no consumer groups support etc. Every consumer will receive every produced message.
func (g *GoChannel) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error) {
// ...

Marshaler

No marshaling is needed when sending messages within the process.