Range operator

iterable.go

@@ -8,26 +8,42 @@ type iterableFromChannel struct {
 	ch chan interface{}
 }
 
-func (it *iterableFromChannel) Iterator() Iterator {
-	return newIteratorFromChannel(it.ch)
+type iterableFromSlice struct {
+	s []interface{}
 }
 
-func newIterableFromChannel(ch chan interface{}) Iterable {
-	return &iterableFromChannel{
-		ch: ch,
-	}
+type iterableFromRange struct {
+	start int
+	count int
 }
 
-type iterableFromSlice struct {
-	s []interface{}
+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,
+	}
+}

iterator.go

@@ -15,6 +15,11 @@ type iteratorFromSlice struct {
 	s     []interface{}
 }
 
+type iteratorFromRange struct {
+	current int
+	end     int // Included
+}
+
 func (it *iteratorFromChannel) Next() bool {
 	if v, ok := <-it.ch; ok {
 		it.item = v
@@ -41,6 +46,15 @@ func (it *iteratorFromSlice) Value() interface{} {
 	return it.s[it.index]
 }
 
+func (it *iteratorFromRange) Next() bool {
+	it.current = it.current + 1
+	return it.current <= it.end
+}
+
+func (it *iteratorFromRange) Value() interface{} {
+	return it.current
+}
+
 func newIteratorFromChannel(ch chan interface{}) Iterator {
 	return &iteratorFromChannel{
 		ch: ch,
@@ -53,3 +67,10 @@ func newIteratorFromSlice(s []interface{}) Iterator {
 		s:     s,
 	}
 }
+
+func newIteratorFromRange(start, end int) Iterator {
+	return &iteratorFromRange{
+		current: start,
+		end:     end,
+	}
+}

observable_test.go

@@ -47,30 +47,6 @@ func TestEmptyOperator(t *testing.T) {
 	assert.Equal(t, "done", text)
 }
 
-func TestRange(t *testing.T) {
-	got := []interface{}{}
-	r, err := Range(1, 5)
-	if err != nil {
-		t.Fail()
-	}
-	r.Subscribe(handlers.NextFunc(func(i interface{}) {
-		got = append(got, i)
-	})).Block()
-	assert.Equal(t, []interface{}{1, 2, 3, 4, 5}, got)
-}
-
-func TestRangeWithNegativeCount(t *testing.T) {
-	r, err := Range(1, -5)
-	assert.NotNil(t, err)
-	assert.Nil(t, r)
-}
-
-func TestRangeWithMaximumExceeded(t *testing.T) {
-	r, err := Range(1<<31, 1)
-	assert.NotNil(t, err)
-	assert.Nil(t, r)
-}
-
 func TestJustOperator(t *testing.T) {
 	myStream := Just(1, 2.01, "foo", map[string]string{"bar": "baz"}, 'a')
 	//numItems := 5

observablecreate.go

@@ -30,6 +30,13 @@ func newObservableFromSlice(s []interface{}) Observable {
 	}
 }
 
+// newObservableFromRange creates an Observable from a range.
+func newObservableFromRange(start, count int) Observable {
+	return &observable{
+		iterable: newIterableFromRange(start, count),
+	}
+}
+
 func isClosed(ch <-chan interface{}) bool {
 	select {
 	case <-ch:
@@ -178,16 +185,7 @@ func Range(start, count int) (Observable, error) {
 		return nil, errors.New(errors.IllegalInputError, "max value is bigger than MaxInt32")
 	}
 
-	out := make(chan interface{})
-	go func() {
-		i := start
-		for i < count+start {
-			out <- i
-			i++
-		}
-		close(out)
-	}()
-	return newObservableFromChannel(out), nil
+	return newObservableFromRange(start, count), nil
 }
 
 // Just creates an Observable with the provided item(s).
@@ -233,8 +231,5 @@ func Start(f Supplier, fs ...Supplier) Observable {
 // Never create an Observable that emits no items and does not terminate
 func Never() Observable {
 	out := make(chan interface{})
-	go func() {
-		select {}
-	}()
 	return newObservableFromChannel(out)
 }

observablecreate_test.go

@@ -326,3 +326,24 @@ func TestFromStatelessIterable(t *testing.T) {
 	AssertThatObservable(t, obs, HasItems(0, 1, 2))
 	AssertThatObservable(t, obs, HasItems(0, 1, 2))
 }
+
+func TestRange(t *testing.T) {
+	obs, err := Range(5, 3)
+	if err != nil {
+		t.Fail()
+	}
+	AssertThatObservable(t, obs, HasItems(5, 6, 7, 8))
+	AssertThatObservable(t, obs, HasItems(5, 6, 7, 8))
+}
+
+func TestRangeWithNegativeCount(t *testing.T) {
+	r, err := Range(1, -5)
+	assert.NotNil(t, err)
+	assert.Nil(t, r)
+}
+
+func TestRangeWithMaximumExceeded(t *testing.T) {
+	r, err := Range(1<<31, 1)
+	assert.NotNil(t, err)
+	assert.Nil(t, r)
+}