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

Documentation

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

• New

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

Example

package main

import (
	"context"
	"fmt"

	"github.com/ysmood/goob"
)

func main() {
	ob := goob.New()

	ctx, unsubscribe := context.WithCancel(context.Background())
	defer unsubscribe()
	s := ob.Subscribe(ctx)

	go func() {
		ob.Publish(1)
		ob.Publish(2)
		ob.Publish(3)
	}()

	for e := range s {
		fmt.Println(e)

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

}

Output:
1
2
3

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