support for subscribes (#52)

Author: chenjunbiao

observable.go

@@ -2,6 +2,7 @@ package y3
 
 import (
 	"io"
+	"sync"
 
 	"github.com/yomorun/y3-codec-golang/pkg/encoding"
 )
@@ -21,11 +22,52 @@ type observableImpl struct {
 }
 
 type iterableImpl struct {
-	channel chan interface{}
+	next                   chan interface{}
+	subscribers            []chan interface{}
+	mutex                  sync.RWMutex
+	producerAlreadyCreated bool
 }
 
 func (i *iterableImpl) Observe() <-chan interface{} {
-	return i.channel
+	i.connect()
+	ch := make(chan interface{})
+	i.mutex.Lock()
+	i.subscribers = append(i.subscribers, ch)
+	i.mutex.Unlock()
+	return ch
+}
+
+func (i *iterableImpl) connect() {
+	i.mutex.Lock()
+	if !i.producerAlreadyCreated {
+		go i.produce()
+		i.producerAlreadyCreated = true
+	}
+	i.mutex.Unlock()
+}
+
+func (i *iterableImpl) produce() {
+	defer func() {
+		i.mutex.RLock()
+		for _, subscriber := range i.subscribers {
+			close(subscriber)
+		}
+		i.mutex.RUnlock()
+	}()
+
+	for {
+		select {
+		case item, ok := <-i.next:
+			if !ok {
+				return
+			}
+			i.mutex.RLock()
+			for _, subscriber := range i.subscribers {
+				subscriber <- item
+			}
+			i.mutex.RUnlock()
+		}
+	}
 }
 
 func (o *observableImpl) Observe() <-chan interface{} {
@@ -36,6 +78,7 @@ func (o *observableImpl) Observe() <-chan interface{} {
 func FromStream(reader io.Reader) Observable {
 
 	f := func(next chan interface{}) {
+		defer close(next)
 		for {
 			buf := make([]byte, 3*1024)
 			n, err := reader.Read(buf)
@@ -173,8 +216,10 @@ func decodeLength(buf []byte) (length int32, err error) {
 
 func createObservable(f func(next chan interface{})) Observable {
 	next := make(chan interface{})
+	subscribers := make([]chan interface{}, 0)
+
 	go f(next)
-	return &observableImpl{iterable: &iterableImpl{channel: next}}
+	return &observableImpl{iterable: &iterableImpl{next: next, subscribers: subscribers}}
 }
 
 //filter root data from the stream

observable_test.go

@@ -11,15 +11,28 @@ import (
 
 func TestObservable(t *testing.T) {
 	buf := []byte{0x81, 0x16, 0xb0, 0x14, 0x10, 0x4, 0x79, 0x6f, 0x6d, 0x6f, 0x11, 0x2, 0x43, 0xe4, 0x92, 0x8, 0x13, 0x2, 0x41, 0xf0, 0x14, 0x2, 0x42, 0x20, 0x81, 0x16, 0xb0, 0x14, 0x10, 0x4, 0x79, 0x6f, 0x6d, 0x6f, 0x11, 0x2, 0x43, 0xe4, 0x92, 0x8, 0x13, 0x2, 0x41, 0xf0, 0x14, 0x2, 0x42, 0x20, 0x81, 0x16, 0xb0, 0x14, 0x10, 0x4, 0x79, 0x6f, 0x6d, 0x6f, 0x11, 0x2, 0x43, 0xe4, 0x92, 0x8, 0x13, 0x2, 0x41, 0xf0, 0x14, 0x2, 0x42, 0x20}
-	var err error = nil
-	var count int = 0
+	var err1 error = nil
+	var err2 error = nil
+	var count1 int = 0
+	var count2 int = 0
 
-	callback := func(v []byte) (interface{}, error) {
+	callback1 := func(v []byte) (interface{}, error) {
 		if (v[0] == 17) && (v[1] == 2) && (v[2] == 67) && (v[3] == 228) {
-			count++
-			return "ok", nil
+			count1++
+			return "ok1", nil
 		} else {
-			err = errors.New("fail")
+			err1 = errors.New("fail")
+			return nil, errors.New("fail")
+		}
+
+	}
+
+	callback2 := func(v []byte) (interface{}, error) {
+		if (v[0] == 19) && (v[1] == 2) && (v[2] == 65) && (v[3] == 240) {
+			count2++
+			return "ok2", nil
+		} else {
+			err2 = errors.New("fail")
 			return nil, errors.New("fail")
 		}
 
@@ -29,15 +42,26 @@ func TestObservable(t *testing.T) {
 
 	source := FromStream(reader)
 
-	consumer := source.Subscribe(0x11).OnObserve(callback)
+	consumer1 := source.Subscribe(0x11).OnObserve(callback1)
 
-	for range consumer {
-		if count == 3 || err != nil {
+	consumer2 := source.Subscribe(0x13).OnObserve(callback2)
+
+	for range consumer1 {
+		if count1 == 3 || err1 != nil {
+			break
+		}
+	}
+
+	for range consumer2 {
+		if count2 == 3 || err2 != nil {
 			break
 		}
 	}
-	assert.NoError(t, err, fmt.Sprintf("subscribe error:%v", err))
-	assert.Equal(t, 3, count, fmt.Sprintf("testing observable %v: %v", 3, count))
-	testPrintf("count=%v, observable_result=%v, err=%v\n", 3, count, err)
+
+	assert.NoError(t, err1, fmt.Sprintf("subscribe2 error:%v", err1))
+	assert.Equal(t, 3, count1, fmt.Sprintf("testing observable1 %v: %v", 3, count1))
+	assert.NoError(t, err2, fmt.Sprintf("subscribe2 error:%v", err2))
+	assert.Equal(t, 3, count2, fmt.Sprintf("testing observable2 %v: %v", 3, count2))
+	testPrintf("count1=%v,count2=%v, observable_result=%v, err1=%v,err2=%v\n", 3, 3, "ok", err1, err2)
 
 }

tester.go

@@ -33,8 +33,9 @@ func testDecoder(observe byte, buf []byte, callback func(v []byte) (interface{},
 // Init create a channel for testing
 func (t *observableTester) Init(callback func(v []byte) (interface{}, error)) *observableTester {
 	t.sourceChannel = make(chan interface{})
+	subscribers := make([]chan interface{}, 0)
 
-	t.source = &observableImpl{iterable: &iterableImpl{channel: t.sourceChannel}}
+	t.source = &observableImpl{iterable: &iterableImpl{next: t.sourceChannel, subscribers: subscribers}}
 
 	consumer := t.source.Subscribe(t.observe).OnObserve(callback)