Create connectable observable

Author: teivah

factory_connectable_test.go

@@ -116,6 +116,39 @@ func Test_Connectable_IterableChannel_WithoutConnect(t *testing.T) {
 		ch <- Of(3)
 		close(ch)
 	}()
-	obs := FromChannel(ch, WithPublishStrategy(), WithBufferedChannel(10))
+	obs := FromChannel(ch, WithPublishStrategy())
 	testConnectableWithoutConnect(t, obs)
 }
+
+func Test_Connectable_IterableCreate_Single(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	testConnectableSingle(t, Create([]Producer{func(_ context.Context, ch chan<- Item) {
+		ch <- Of(1)
+		ch <- Of(2)
+		ch <- Of(3)
+		cancel()
+	}}, WithPublishStrategy(), WithContext(ctx)))
+}
+
+func Test_Connectable_IterableCreate_Composed(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	testConnectableComposed(t, Create([]Producer{func(_ context.Context, ch chan<- Item) {
+		ch <- Of(1)
+		ch <- Of(2)
+		ch <- Of(3)
+		cancel()
+	}}, WithPublishStrategy(), WithContext(ctx)))
+}
+
+func Test_Connectable_IterableCreate_WithoutConnect(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	testConnectableWithoutConnect(t, Create([]Producer{func(_ context.Context, ch chan<- Item) {
+		ch <- Of(1)
+		ch <- Of(2)
+		ch <- Of(3)
+		cancel()
+	}}, WithPublishStrategy(), WithContext(ctx)))
+}

iterable_create.go

@@ -5,8 +5,11 @@ import (
 )
 
 type createIterable struct {
-	opts []Option
-	next <-chan Item
+	next                   <-chan Item
+	opts                   []Option
+	subscribers            []chan Item
+	mutex                  sync.RWMutex
+	producerAlreadyCreated bool
 }
 
 func newCreateIterable(fs []Producer, opts ...Option) Iterable {
@@ -33,7 +36,47 @@ func newCreateIterable(fs []Producer, opts ...Option) Iterable {
 		next: next,
 	}
 }
+func (i *createIterable) Observe(opts ...Option) <-chan Item {
+	mergedOptions := append(i.opts, opts...)
+	option := parseOptions(mergedOptions...)
 
-func (i *createIterable) Observe(_ ...Option) <-chan Item {
-	return i.next
+	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 *createIterable) connect() {
+	i.mutex.Lock()
+	if !i.producerAlreadyCreated {
+		go i.produce()
+		i.producerAlreadyCreated = true
+	}
+	i.mutex.Unlock()
+}
+
+func (i *createIterable) 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()
 }