Disposed type

Author: teivah

Diff for: factory_connectable_test.go

@@ -40,6 +40,24 @@ func Test_Connectable_IterableChannel_Composed(t *testing.T) {
 	testConnectableComposed(t, obs)
 }
 
+func Test_Connectable_IterableChannel_Disposed(t *testing.T) {
+	ch := make(chan Item, 10)
+	go func() {
+		ch <- Of(1)
+		ch <- Of(2)
+		ch <- Of(3)
+		close(ch)
+	}()
+	obs := &ObservableImpl{
+		iterable: newChannelIterable(ch, WithPublishStrategy()),
+	}
+	obs.Connect()()
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	time.Sleep(50 * time.Millisecond)
+	Assert(ctx, t, obs, IsEmpty())
+}
+
 func Test_Connectable_IterableChannel_WithoutConnect(t *testing.T) {
 	ch := make(chan Item, 10)
 	go func() {
@@ -82,6 +100,24 @@ func Test_Connectable_IterableCreate_Composed(t *testing.T) {
 	testConnectableComposed(t, obs)
 }
 
+func Test_Connectable_IterableCreate_Disposed(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	obs := &ObservableImpl{
+		iterable: newCreateIterable([]Producer{func(_ context.Context, ch chan<- Item) {
+			ch <- Of(1)
+			ch <- Of(2)
+			ch <- Of(3)
+			cancel()
+		}}, WithPublishStrategy(), WithContext(ctx)),
+	}
+	obs.Connect()()
+	ctx, cancel = context.WithTimeout(context.Background(), 550*time.Millisecond)
+	defer cancel()
+	time.Sleep(50 * time.Millisecond)
+	Assert(ctx, t, obs, IsEmpty())
+}
+
 func Test_Connectable_IterableCreate_WithoutConnect(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

Diff for: iterable_channel.go

@@ -1,6 +1,7 @@
 package rxgo
 
 import (
+	"context"
 	"sync"
 )
 
@@ -29,7 +30,7 @@ func (i *channelIterable) Observe(opts ...Option) <-chan Item {
 	}
 
 	if option.isConnectOperation() {
-		i.connect()
+		i.connect(option.buildContext())
 		return nil
 	}
 
@@ -40,27 +41,37 @@ func (i *channelIterable) Observe(opts ...Option) <-chan Item {
 	return ch
 }
 
-func (i *channelIterable) connect() {
+func (i *channelIterable) connect(ctx context.Context) {
 	i.mutex.Lock()
 	if !i.producerAlreadyCreated {
-		go i.produce()
+		go i.produce(ctx)
 		i.producerAlreadyCreated = true
 	}
 	i.mutex.Unlock()
 }
 
-func (i *channelIterable) produce() {
-	for item := range i.next {
+func (i *channelIterable) produce(ctx context.Context) {
+	defer func() {
 		i.mutex.RLock()
 		for _, subscriber := range i.subscribers {
-			subscriber <- item
+			close(subscriber)
 		}
 		i.mutex.RUnlock()
-	}
+	}()
 
-	i.mutex.RLock()
-	for _, subscriber := range i.subscribers {
-		close(subscriber)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case item, ok := <-i.next:
+			if !ok {
+				return
+			}
+			i.mutex.RLock()
+			for _, subscriber := range i.subscribers {
+				subscriber <- item
+			}
+			i.mutex.RUnlock()
+		}
 	}
-	i.mutex.RUnlock()
 }

Diff for: iterable_create.go

@@ -1,6 +1,7 @@
 package rxgo
 
 import (
+	"context"
 	"sync"
 )
 
@@ -46,7 +47,7 @@ func (i *createIterable) Observe(opts ...Option) <-chan Item {
 	}
 
 	if option.isConnectOperation() {
-		i.connect()
+		i.connect(option.buildContext())
 		return nil
 	}
 
@@ -57,27 +58,37 @@ func (i *createIterable) Observe(opts ...Option) <-chan Item {
 	return ch
 }
 
-func (i *createIterable) connect() {
+func (i *createIterable) connect(ctx context.Context) {
 	i.mutex.Lock()
 	if !i.producerAlreadyCreated {
-		go i.produce()
+		go i.produce(ctx)
 		i.producerAlreadyCreated = true
 	}
 	i.mutex.Unlock()
 }
 
-func (i *createIterable) produce() {
-	for item := range i.next {
+func (i *createIterable) produce(ctx context.Context) {
+	defer func() {
 		i.mutex.RLock()
 		for _, subscriber := range i.subscribers {
-			subscriber <- item
+			close(subscriber)
 		}
 		i.mutex.RUnlock()
-	}
+	}()
 
-	i.mutex.RLock()
-	for _, subscriber := range i.subscribers {
-		close(subscriber)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case item, ok := <-i.next:
+			if !ok {
+				return
+			}
+			i.mutex.RLock()
+			for _, subscriber := range i.subscribers {
+				subscriber <- item
+			}
+			i.mutex.RUnlock()
+		}
 	}
-	i.mutex.RUnlock()
 }

Diff for: observable.go

@@ -10,7 +10,6 @@ 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
@@ -23,6 +22,7 @@ type Observable interface {
 	BufferWithCount(count int, opts ...Option) Observable
 	BufferWithTime(timespan Duration, opts ...Option) Observable
 	BufferWithTimeOrCount(timespan Duration, count int, opts ...Option) Observable
+	Connect() Disposable
 	Contains(equal Predicate, opts ...Option) Single
 	Count(opts ...Option) Single
 	Debounce(timespan Duration, opts ...Option) Observable

Diff for: observable_operator.go

@@ -542,9 +542,10 @@ func (o *ObservableImpl) BufferWithTimeOrCount(timespan Duration, count int, opt
 }
 
 // Connect instructs a connectable Observable to begin emitting items to its subscribers.
-func (o *ObservableImpl) Connect() Observable {
-	o.Observe(connect())
-	return o
+func (o *ObservableImpl) Connect() Disposable {
+	ctx, cancel := context.WithCancel(context.Background())
+	o.Observe(WithContext(ctx), connect())
+	return Disposable(cancel)
 }
 
 // Contains determines whether an Observable emits a particular item or not.

Diff for: types.go

@@ -37,6 +37,8 @@ type (
 	Supplier func(ctx context.Context) Item
 	// Disposed is a notification channel indicating when an Observable is closed.
 	Disposed <-chan struct{}
+	// Disposable is a function to be called in order to dispose a subscription.
+	Disposable context.CancelFunc
 
 	// NextFunc handles a next item in a stream.
 	NextFunc func(interface{})