eventbus

README

eventbus

A lightweight observable. A Go channel cannot have an unbounded buffer, and choosing a size is a pain; this library grows the buffer as needed.

• unbounded buffer
• one publisher, many subscribers
• thread-safe
• subscribers never block each other
• stable event order

Provenance

eventbus is a Satellite module of wand. It is a snapshot of goob v0.4.0 (commit 8c75c65), imported without git history and maintained here by headlesslab; see NOTICE. The module path changed, the exported identifiers did not.

Replace github.com/ysmood/goob with github.com/headlesslab/eventbus and goob. with eventbus.; New, NewPipe, Observable, Event, ... keep their names.

The test suite uses leakcheck and therefore runs with GODEBUG=tracebackancestors=1000.

Usage

import "github.com/headlesslab/eventbus"

The tests are the documentation: see examples_test.go.

Versioning

v0.x: no breaking change without a major version bump. Releases are cut by hand when the module changes; wand pins each release exactly.

License

MIT, see LICENSE.

Documentation

Overview

Package eventbus is a lightweight observable: one publisher, many subscribers, an unbounded buffer and a stable event order.

Example (Basic)

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/headlesslab/eventbus"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
	defer cancel()

	// create an observable instance
	ob := eventbus.New(ctx)

	events := ob.Subscribe(context.TODO())

	// publish events without blocking
	ob.Publish(1)
	ob.Publish(2)
	ob.Publish(3)

	// consume events
	for e := range events {
		fmt.Print(e)
	}

}

Output:
123

Index

• func NewPipe(ctx context.Context) (Write func(Event), Events <-chan Event)
• type Event
• type Events
• type Observable
  • func New(ctx context.Context) *Observable
  • func (ob *Observable) Len() int
  • func (ob *Observable) Publish(e Event)
  • func (ob *Observable) Subscribe(ctx context.Context) Events

Examples

• Package (Basic)

Functions

func NewPipe

func NewPipe(ctx context.Context) (Write func(Event), Events <-chan Event)

NewPipe instance. Pipe the Event via Write to Events. Events uses an internal buffer so it won't block Write.

Types

type Event

type Event interface{}

Event interface

type Events

type Events <-chan Event

Events channel

type Observable

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

Observable hub

func New

func New(ctx context.Context) *Observable

New observable instance

func (*Observable) Len

func (ob *Observable) Len() int

Len of the subscribers

func (*Observable) Publish

func (ob *Observable) Publish(e Event)

Publish message to the queue

func (*Observable) Subscribe

func (ob *Observable) Subscribe(ctx context.Context) Events

Subscribe message