Init: first commit 🎉

observable/iterable.go

@@ -0,0 +1,18 @@
+package observable
+
+import (
+	"errors"
+)
+
+type Iterable <-chan interface{}
+
+func NewIterable(any interface{}) (Iterable, error) {
+	switch any := any.(type) {
+	case chan interface{}:
+		return Iterable(any), nil
+	case <-chan interface{}:
+		return Iterable(any), nil
+	default:
+		return nil, errors.New("type error")
+	}
+}

observable/observable.go

@@ -0,0 +1,68 @@
+package observable
+
+import (
+	"errors"
+	"sync"
+)
+
+type Observable struct {
+	iterable Iterable
+	listener *sync.Map
+	done     bool
+	doneLock sync.RWMutex
+}
+
+func (o *Observable) process() {
+	for item := range o.iterable {
+		o.listener.Range(func(key, value interface{}) bool {
+			elm := value.(*Subscriber)
+			elm.Emit(item)
+			return true
+		})
+	}
+	o.close()
+}
+
+func (o *Observable) close() {
+	o.doneLock.Lock()
+	o.done = true
+	o.doneLock.Unlock()
+
+	o.listener.Range(func(key, value interface{}) bool {
+		elm := value.(*Subscriber)
+		elm.Close()
+		return true
+	})
+}
+
+func (o *Observable) Subscribe() (Subscription, error) {
+	o.doneLock.RLock()
+	done := o.done
+	o.doneLock.RUnlock()
+	if done == true {
+		return nil, errors.New("Observable is closed")
+	}
+	subscriber := newSubscriber()
+	o.listener.Store(subscriber.Out(), subscriber)
+	return subscriber.Out(), nil
+}
+
+func (o *Observable) UnSubscribe(sub Subscription) {
+	elm, exist := o.listener.Load(sub)
+	if !exist {
+		println("not exist")
+		return
+	}
+	subscriber := elm.(*Subscriber)
+	o.listener.Delete(subscriber.Out())
+	subscriber.Close()
+}
+
+func NewObservable(any Iterable) *Observable {
+	observable := &Observable{
+		iterable: any,
+		listener: &sync.Map{},
+	}
+	go observable.process()
+	return observable
+}

observable/observable_test.go

@@ -0,0 +1,117 @@
+package observable
+
+import (
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+)
+
+func iterator(item []interface{}) chan interface{} {
+	ch := make(chan interface{})
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		for _, elm := range item {
+			ch <- elm
+		}
+		close(ch)
+	}()
+	return ch
+}
+
+func TestObservable(t *testing.T) {
+	iter := iterator([]interface{}{1, 2, 3, 4, 5})
+	src := NewObservable(iter)
+	data, err := src.Subscribe()
+	if err != nil {
+		t.Error(err)
+	}
+	count := 0
+	for {
+		_, open := <-data
+		if !open {
+			break
+		}
+		count = count + 1
+	}
+	if count != 5 {
+		t.Error("Revc number error")
+	}
+}
+
+func TestObservable_MutilSubscribe(t *testing.T) {
+	iter := iterator([]interface{}{1, 2, 3, 4, 5})
+	src := NewObservable(iter)
+	ch1, _ := src.Subscribe()
+	ch2, _ := src.Subscribe()
+	count := 0
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	waitCh := func(ch <-chan interface{}) {
+		for {
+			_, open := <-ch
+			if !open {
+				break
+			}
+			count = count + 1
+		}
+		wg.Done()
+	}
+	go waitCh(ch1)
+	go waitCh(ch2)
+	wg.Wait()
+	if count != 10 {
+		t.Error("Revc number error")
+	}
+}
+
+func TestObservable_UnSubscribe(t *testing.T) {
+	iter := iterator([]interface{}{1, 2, 3, 4, 5})
+	src := NewObservable(iter)
+	data, err := src.Subscribe()
+	if err != nil {
+		t.Error(err)
+	}
+	src.UnSubscribe(data)
+	_, open := <-data
+	if open {
+		t.Error("Revc number error")
+	}
+}
+
+func TestObservable_SubscribeGoroutineLeak(t *testing.T) {
+	// waiting for other goroutine recycle
+	time.Sleep(120 * time.Millisecond)
+	init := runtime.NumGoroutine()
+	iter := iterator([]interface{}{1, 2, 3, 4, 5})
+	src := NewObservable(iter)
+	max := 100
+
+	var list []Subscription
+	for i := 0; i < max; i++ {
+		ch, _ := src.Subscribe()
+		list = append(list, ch)
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(max)
+	waitCh := func(ch <-chan interface{}) {
+		for {
+			_, open := <-ch
+			if !open {
+				break
+			}
+		}
+		wg.Done()
+	}
+
+	for _, ch := range list {
+		go waitCh(ch)
+	}
+	wg.Wait()
+	now := runtime.NumGoroutine()
+	if init != now {
+		t.Errorf("Goroutine Leak: init %d now %d", init, now)
+	}
+}

observable/subscriber.go

@@ -0,0 +1,35 @@
+package observable
+
+import (
+	"sync"
+
+	"gopkg.in/eapache/channels.v1"
+)
+
+type Subscription <-chan interface{}
+
+type Subscriber struct {
+	buffer *channels.InfiniteChannel
+	once   sync.Once
+}
+
+func (s *Subscriber) Emit(item interface{}) {
+	s.buffer.In() <- item
+}
+
+func (s *Subscriber) Out() Subscription {
+	return s.buffer.Out()
+}
+
+func (s *Subscriber) Close() {
+	s.once.Do(func() {
+		s.buffer.Close()
+	})
+}
+
+func newSubscriber() *Subscriber {
+	sub := &Subscriber{
+		buffer: channels.NewInfiniteChannel(),
+	}
+	return sub
+}

observable/util.go

@@ -0,0 +1,15 @@
+package observable
+
+func mergeWithBytes(ch <-chan interface{}, buf []byte) chan interface{} {
+	out := make(chan interface{})
+	go func() {
+		defer close(out)
+		if len(buf) != 0 {
+			out <- buf
+		}
+		for elm := range ch {
+			out <- elm
+		}
+	}()
+	return out
+}