Connectable first version

Author: teivah

factory.go

@@ -185,9 +185,9 @@ func Empty() Observable {
 }
 
 // FromChannel creates a cold observable from a channel.
-func FromChannel(next <-chan Item) Observable {
+func FromChannel(next <-chan Item, opts ...Option) Observable {
 	return &ObservableImpl{
-		iterable: newChannelIterable(next),
+		iterable: newChannelIterable(next, opts...),
 	}
 }
 

factory_connectable_test.go

@@ -0,0 +1,121 @@
+package rxgo
+
+import (
+	"context"
+	"fmt"
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/sync/errgroup"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+)
+
+func testConnectableSingle(t *testing.T, obs Observable) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	eg, _ := errgroup.WithContext(ctx)
+
+	expected := []interface{}{1, 2, 3}
+
+	nbConsumers := 3
+	wg := sync.WaitGroup{}
+	wg.Add(nbConsumers)
+	// Before Connect() is called we create multiple observers
+	// We check all observers receive the same items
+	for i := 0; i < nbConsumers; i++ {
+		eg.Go(func() error {
+			observer := obs.Observe(WithContext(ctx))
+			wg.Done()
+			got, err := collect(ctx, observer)
+			if err != nil {
+				return err
+			}
+			if !reflect.DeepEqual(got, expected) {
+				return fmt.Errorf("expected: %v, got: %v", expected, got)
+			}
+			return nil
+		})
+	}
+
+	wg.Wait()
+	obs.Connect()
+	assert.NoError(t, eg.Wait())
+}
+
+func testConnectableComposed(t *testing.T, obs Observable) {
+	obs = obs.Map(func(_ context.Context, i interface{}) (interface{}, error) {
+		return i.(int) + 1, nil
+	}, WithPublishStrategy())
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	eg, _ := errgroup.WithContext(ctx)
+
+	expected := []interface{}{2, 3, 4}
+
+	nbConsumers := 3
+	wg := sync.WaitGroup{}
+	wg.Add(nbConsumers)
+	// Before Connect() is called we create multiple observers
+	// We check all observers receive the same items
+	for i := 0; i < nbConsumers; i++ {
+		eg.Go(func() error {
+			observer := obs.Observe(WithContext(ctx))
+			wg.Done()
+
+			got, err := collect(ctx, observer)
+			if err != nil {
+				return err
+			}
+			if !reflect.DeepEqual(got, expected) {
+				return fmt.Errorf("expected: %v, got: %v", expected, got)
+			}
+			return nil
+		})
+	}
+
+	wg.Wait()
+	obs.Connect()
+	assert.NoError(t, eg.Wait())
+}
+
+func testConnectableWithoutConnect(t *testing.T, obs Observable) {
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	Assert(ctx, t, obs, IsEmpty())
+}
+
+func Test_Connectable_IterableChannel_Single(t *testing.T) {
+	ch := make(chan Item, 10)
+	go func() {
+		ch <- Of(1)
+		ch <- Of(2)
+		ch <- Of(3)
+		close(ch)
+	}()
+	testConnectableSingle(t, FromChannel(ch, WithPublishStrategy()))
+}
+
+func Test_Connectable_IterableChannel_Composed(t *testing.T) {
+	ch := make(chan Item, 10)
+	go func() {
+		ch <- Of(1)
+		ch <- Of(2)
+		ch <- Of(3)
+		close(ch)
+	}()
+	testConnectableComposed(t, FromChannel(ch, WithPublishStrategy()))
+}
+
+func Test_Connectable_IterableChannel_WithoutConnect(t *testing.T) {
+	ch := make(chan Item, 10)
+	go func() {
+		ch <- Of(1)
+		ch <- Of(2)
+		ch <- Of(3)
+		close(ch)
+	}()
+	obs := FromChannel(ch, WithPublishStrategy(), WithBufferedChannel(10))
+	testConnectableWithoutConnect(t, obs)
+}

factory_test.go

@@ -9,6 +9,25 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func collect(ctx context.Context, ch <-chan Item) ([]interface{}, error) {
+	s := make([]interface{}, 0)
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case item, ok := <-ch:
+			if !ok {
+				return s, nil
+			}
+			if item.Error() {
+				s = append(s, item.E)
+			} else {
+				s = append(s, item.V)
+			}
+		}
+	}
+}
+
 func Test_Amb1(t *testing.T) {
 	obs := Amb([]Observable{testObservable(1, 2, 3), Empty()})
 	Assert(context.Background(), t, obs, HasItems(1, 2, 3))

go.mod

@@ -6,4 +6,5 @@ require (
 	github.com/cenkalti/backoff/v4 v4.0.0
 	github.com/emirpasic/gods v1.12.0
 	github.com/stretchr/testify v1.4.0
+	golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e
 )

go.sum

@@ -10,6 +10,8 @@ github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=

iterable_channel.go

@@ -1,13 +1,66 @@
 package rxgo
 
+import (
+	"sync"
+)
+
 type channelIterable struct {
-	next <-chan Item
+	next                   <-chan Item
+	opts                   []Option
+	subscribers            []chan Item
+	mutex                  sync.RWMutex
+	producerAlreadyCreated bool
 }
 
-func newChannelIterable(next <-chan Item) Iterable {
-	return &channelIterable{next: next}
+func newChannelIterable(next <-chan Item, opts ...Option) Iterable {
+	return &channelIterable{
+		next:        next,
+		subscribers: make([]chan Item, 0),
+		opts:        opts,
+	}
 }
 
-func (i *channelIterable) Observe(_ ...Option) <-chan Item {
-	return i.next
+func (i *channelIterable) Observe(opts ...Option) <-chan Item {
+	mergedOptions := append(i.opts, opts...)
+	option := parseOptions(mergedOptions...)
+
+	if !option.isConnectable() {
+		return i.next
+	}
+
+	if option.isConnectOperation() {
+		i.connect()
+		return nil
+	}
+
+	ch := option.buildChannel()
+	i.mutex.Lock()
+	i.subscribers = append(i.subscribers, ch)
+	i.mutex.Unlock()
+	return ch
+}
+
+func (i *channelIterable) connect() {
+	i.mutex.Lock()
+	if !i.producerAlreadyCreated {
+		go i.produce()
+		i.producerAlreadyCreated = true
+	}
+	i.mutex.Unlock()
+}
+
+func (i *channelIterable) produce() {
+	for item := range i.next {
+		i.mutex.RLock()
+		for _, subscriber := range i.subscribers {
+			subscriber <- item
+		}
+		i.mutex.RUnlock()
+	}
+
+	i.mutex.RLock()
+	for _, subscriber := range i.subscribers {
+		close(subscriber)
+	}
+	i.mutex.RUnlock()
 }

observable.go

@@ -10,6 +10,7 @@ import (
 // Observable is the standard interface for Observables.
 type Observable interface {
 	Iterable
+	Connect() Observable
 	All(predicate Predicate, opts ...Option) Single
 	AverageFloat32(opts ...Option) Single
 	AverageFloat64(opts ...Option) Single
@@ -117,7 +118,7 @@ func observable(iterable Iterable, operatorFactory func() operator, forceSeq, by
 	return &ObservableImpl{
 		iterable: newFactoryIterable(func(propagatedOptions ...Option) <-chan Item {
 			mergedOptions := append(opts, propagatedOptions...)
-			option = parseOptions(mergedOptions...)
+			option := parseOptions(mergedOptions...)
 
 			next := option.buildChannel()
 			ctx := option.buildContext()
@@ -196,10 +197,10 @@ func optionalSingle(iterable Iterable, operatorFactory func() operator, forceSeq
 }
 
 func runSeq(ctx context.Context, next chan Item, iterable Iterable, operatorFactory func() operator, option Option, opts ...Option) {
+	observe := iterable.Observe(opts...)
 	go func() {
 		op := operatorFactory()
 		stopped := false
-		observe := iterable.Observe(opts...)
 		operator := operatorOptions{
 			stop: func() {
 				if option.getErrorStrategy() == StopOnError {

observable_operator.go

@@ -541,6 +541,11 @@ func (o *ObservableImpl) BufferWithTimeOrCount(timespan Duration, count int, opt
 	return customObservableOperator(f, opts...)
 }
 
+func (o *ObservableImpl) Connect() Observable {
+	o.Observe(connect())
+	return o
+}
+
 // Contains determines whether an Observable emits a particular item or not.
 func (o *ObservableImpl) Contains(equal Predicate, opts ...Option) Single {
 	return single(o, func() operator {

options.go

@@ -15,6 +15,8 @@ type Option interface {
 	buildContext() context.Context
 	getBackPressureStrategy() BackpressureStrategy
 	getErrorStrategy() OnErrorStrategy
+	isConnectable() bool
+	isConnectOperation() bool
 }
 
 type funcOption struct {
@@ -27,6 +29,8 @@ type funcOption struct {
 	backPressureStrategy BackpressureStrategy
 	onErrorStrategy      OnErrorStrategy
 	propagate            bool
+	connectable          bool
+	connectOperation     bool
 }
 
 func (fdo *funcOption) toPropagate() bool {
@@ -63,6 +67,14 @@ func (fdo *funcOption) getErrorStrategy() OnErrorStrategy {
 	return fdo.onErrorStrategy
 }
 
+func (fdo *funcOption) isConnectable() bool {
+	return fdo.connectable
+}
+
+func (fdo *funcOption) isConnectOperation() bool {
+	return fdo.connectOperation
+}
+
 func (fdo *funcOption) apply(do *funcOption) {
 	fdo.f(do)
 }
@@ -131,3 +143,15 @@ func WithErrorStrategy(strategy OnErrorStrategy) Option {
 		options.onErrorStrategy = strategy
 	})
 }
+
+func WithPublishStrategy() Option {
+	return newFuncOption(func(options *funcOption) {
+		options.connectable = true
+	})
+}
+
+func connect() Option {
+	return newFuncOption(func(options *funcOption) {
+		options.connectOperation = true
+	})
+}