goob

README

Overview

A lightweight observable lib. The problem of go channel is that is doesn't support unlimited buffer size, it's a pain to decide what size to use, this lib will dynamically handle it.

• unlimited buffer size
• one publisher to multiple subscribers
• thread-safe
• subscribers never block each other
• publishing order is the same as listening order

Examples

See examples_test.go.

Documentation

Overview

Example (Basic)

package main

import (
	"context"
	"fmt"

	"github.com/ysmood/goob"
)

func main() {
	// create an observable instance
	ob := goob.New()

	// use context primitive to unsubscribe observable
	ctx, unsubscribe := context.WithCancel(context.Background())
	defer unsubscribe()

	// create 2 subscribers
	s1 := ob.Subscribe(ctx)
	s2 := ob.Subscribe(ctx)

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

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

		if e.(int) == 3 {
			break
		}
	}

	// s2 consume events with goob.Each helper, it will auto cast the type
	goob.Each(s2, func(i int) bool {
		fmt.Print(i)
		return i == 3
	})

}

Output:
123123

Index

• func Each(s <-chan Event, iteratee interface{})
• type Event
• type Observable
  • func New() *Observable
  • func (ob *Observable) Count() int
  • func (ob *Observable) Filter(ctx context.Context, iteratee interface{}) *Observable
  • func (ob *Observable) Map(ctx context.Context, iteratee interface{}) *Observable
  • func (ob *Observable) Publish(e Event)
  • func (ob *Observable) Subscribe(ctx context.Context) <-chan Event

Examples

• Package (Basic)

Functions

func Each

func Each(s <-chan Event, iteratee interface{})

Each event, if iteratee returns true, the iteration will stop

Types

type Event

type Event interface{}

Event interface

type Observable

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

Observable hub

func New

func New() *Observable

New observable instance

func (*Observable) Count

func (ob *Observable) Count() int

Count of the subscribers

func (*Observable) Filter

func (ob *Observable) Filter(ctx context.Context, iteratee interface{}) *Observable

Filter events

func (*Observable) Map

func (ob *Observable) Map(ctx context.Context, iteratee interface{}) *Observable

Map events

func (*Observable) Publish

func (ob *Observable) Publish(e Event)

Publish message to the queue

func (*Observable) Subscribe

func (ob *Observable) Subscribe(ctx context.Context) <-chan Event

Subscribe message, the ctx is used to cancel the subscription