goob

README

Overview

A lightweight observable lib. Go channel doesn't support unlimited buffer size, it's a pain to decide what size to use, this lib will handle it dynamically.

• unlimited buffer size
• one publisher to multiple subscribers
• thread-safe
• subscribers never block each other
• stable event order

Examples

See examples_test.go.

Benchmark

goos: darwin
goarch: amd64
pkg: github.com/ysmood/goob
cpu: Intel(R) Core(TM) i7-8750H CPU @ 2.20GHz
BenchmarkPublish-12    	 7493547	       143.9 ns/op	      86 B/op	       0 allocs/op
BenchmarkConsume-12    	 4258910	       275.5 ns/op	       0 B/op	       0 allocs/op

Documentation

Overview

Example (Basic)

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/ysmood/goob"
)

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

	// create an observable instance
	ob := goob.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