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/runZeroInc/go-rod/pkg/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/runZeroInc/go-rod/pkg/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 any

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