Merge 3f219cb into 6007666

connectableobservable.go

@@ -8,21 +8,28 @@ import (
 )
 
 type ConnectableObservable interface {
+	Iterable
 	Connect() Observer
 	Subscribe(handler handlers.EventHandler, opts ...options.Option) Observer
 }
 
 type connectableObservable struct {
+	iterator   Iterator
 	observable Observable
 	observers  []Observer
 }
 
-func NewConnectableObservable(observable Observable) ConnectableObservable {
+func newConnectableObservableFromObservable(observable Observable) ConnectableObservable {
 	return &connectableObservable{
 		observable: observable,
+		iterator:   observable.Iterator(),
 	}
 }
 
+func (c *connectableObservable) Iterator() Iterator {
+	return c.iterator
+}
+
 func (c *connectableObservable) Subscribe(handler handlers.EventHandler, opts ...options.Option) Observer {
 	ob := CheckEventHandler(handler)
 	c.observers = append(c.observers, ob)
@@ -32,12 +39,13 @@ func (c *connectableObservable) Subscribe(handler handlers.EventHandler, opts ..
 func (c *connectableObservable) Connect() Observer {
 	source := make([]interface{}, 0)
 
+	it := c.iterator
 	for {
-		item, err := c.observable.Next()
-		if err != nil {
+		if item, err := it.Next(); err == nil {
+			source = append(source, item)
+		} else {
 			break
 		}
-		source = append(source, item)
 	}
 
 	var wg sync.WaitGroup

flatmap.go

@@ -26,9 +26,7 @@ func (o *observable) flatMap(
 
 	go flatteningFunc(out, o, apply, maxInParallel)
 
-	return &observable{
-		ch: out,
-	}
+	return newObservableFromChannel(out)
 }
 
 func flatObservedSequence(out chan interface{}, o Observable, apply func(interface{}) Observable, maxInParallel uint) {
@@ -43,21 +41,22 @@ func flatObservedSequence(out chan interface{}, o Observable, apply func(interfa
 
 	count = 0
 
+	it := o.Iterator()
 	for {
-		element, err := o.Next()
-		if err != nil {
-			break
-		}
-		sequence = apply(element)
-		count++
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			sequence.Subscribe(emissionObserver).Block()
-		}()
+		if item, err := it.Next(); err == nil {
+			sequence = apply(item)
+			count++
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				sequence.Subscribe(emissionObserver).Block()
+			}()
 
-		if count%maxInParallel == 0 {
-			wg.Wait()
+			if count%maxInParallel == 0 {
+				wg.Wait()
+			}
+		} else {
+			break
 		}
 	}
 

iterable.go

@@ -0,0 +1,65 @@
+package rxgo
+
+type Iterable interface {
+	Iterator() Iterator
+}
+
+type iterableFromChannel struct {
+	ch chan interface{}
+}
+
+type iterableFromSlice struct {
+	s []interface{}
+}
+
+type iterableFromRange struct {
+	start int
+	count int
+}
+
+type iterableFromFunc struct {
+	f func(chan interface{})
+}
+
+func (it *iterableFromFunc) Iterator() Iterator {
+	out := make(chan interface{})
+	go it.f(out)
+	return newIteratorFromChannel(out)
+}
+
+func (it *iterableFromChannel) Iterator() Iterator {
+	return newIteratorFromChannel(it.ch)
+}
+
+func (it *iterableFromSlice) Iterator() Iterator {
+	return newIteratorFromSlice(it.s)
+}
+
+func (it *iterableFromRange) Iterator() Iterator {
+	return newIteratorFromRange(it.start-1, it.start+it.count)
+}
+
+func newIterableFromChannel(ch chan interface{}) Iterable {
+	return &iterableFromChannel{
+		ch: ch,
+	}
+}
+
+func newIterableFromSlice(s []interface{}) Iterable {
+	return &iterableFromSlice{
+		s: s,
+	}
+}
+
+func newIterableFromRange(start, count int) Iterable {
+	return &iterableFromRange{
+		start: start,
+		count: count,
+	}
+}
+
+func newIterableFromFunc(f func(chan interface{})) Iterable {
+	return &iterableFromFunc{
+		f: f,
+	}
+}

iterator.go

@@ -1,6 +1,67 @@
 package rxgo
 
-// Iterator type is implemented by Iterable.
+import "github.com/reactivex/rxgo/errors"
+
 type Iterator interface {
 	Next() (interface{}, error)
 }
+
+type iteratorFromChannel struct {
+	ch chan interface{}
+}
+
+type iteratorFromSlice struct {
+	index int
+	s     []interface{}
+}
+
+type iteratorFromRange struct {
+	current int
+	end     int // Included
+}
+
+func (it *iteratorFromChannel) Next() (interface{}, error) {
+	if next, ok := <-it.ch; ok {
+		return next, nil
+	}
+
+	return nil, errors.New(errors.EndOfIteratorError)
+}
+
+func (it *iteratorFromSlice) Next() (interface{}, error) {
+	it.index = it.index + 1
+	if it.index < len(it.s) {
+		return it.s[it.index], nil
+	} else {
+		return nil, errors.New(errors.EndOfIteratorError)
+	}
+}
+
+func (it *iteratorFromRange) Next() (interface{}, error) {
+	it.current = it.current + 1
+	if it.current <= it.end {
+		return it.current, nil
+	} else {
+		return nil, errors.New(errors.EndOfIteratorError)
+	}
+}
+
+func newIteratorFromChannel(ch chan interface{}) Iterator {
+	return &iteratorFromChannel{
+		ch: ch,
+	}
+}
+
+func newIteratorFromSlice(s []interface{}) Iterator {
+	return &iteratorFromSlice{
+		index: -1,
+		s:     s,
+	}
+}
+
+func newIteratorFromRange(start, end int) Iterator {
+	return &iteratorFromRange{
+		current: start,
+		end:     end,
+	}
+}