StartWith operator. Minor test migration to use assertion API. (#157)

assert.go

@@ -18,22 +18,24 @@ type Assertion interface {
 	hasValueFunc() (bool, interface{})
 	hasRaisedErrorFunc() (bool, error)
 	hasRaisedAnErrorFunc() bool
+	hasNotRaisedAnErrorFunc() bool
 	isEmptyFunc() (bool, bool)
 }
 
 type assertion struct {
-	f                     func(*assertion)
-	checkHasItems         bool
-	hasItems              []interface{}
-	checkHasSize          bool
-	hasSize               int
-	checkHasValue         bool
-	hasValue              interface{}
-	checkHasRaisedError   bool
-	hasRaisedError        error
-	checkHasRaisedAnError bool
-	checkIsEmpty          bool
-	isEmpty               bool
+	f                        func(*assertion)
+	checkHasItems            bool
+	hasItems                 []interface{}
+	checkHasSize             bool
+	hasSize                  int
+	checkHasValue            bool
+	hasValue                 interface{}
+	checkHasRaisedError      bool
+	hasRaisedError           error
+	checkHasRaisedAnError    bool
+	checkHasNotRaisedAnError bool
+	checkIsEmpty             bool
+	isEmpty                  bool
 }
 
 func (ass *assertion) hasItemsFunc() (bool, []interface{}) {
@@ -56,6 +58,10 @@ func (ass *assertion) hasRaisedAnErrorFunc() bool {
 	return ass.checkHasRaisedAnError
 }
 
+func (ass *assertion) hasNotRaisedAnErrorFunc() bool {
+	return ass.checkHasNotRaisedAnError
+}
+
 func (ass *assertion) isEmptyFunc() (bool, bool) {
 	return ass.checkIsEmpty, ass.isEmpty
 }
@@ -133,6 +139,13 @@ func HasRaisedAnError() Assertion {
 	})
 }
 
+// HasNotRaisedAnError checks that a single does not raise an error.
+func HasNotRaisedAnError() Assertion {
+	return newAssertion(func(a *assertion) {
+		a.checkHasNotRaisedAnError = true
+	})
+}
+
 // AssertThatObservable asserts the result of an Observable against a list of assertions.
 func AssertThatObservable(t *testing.T, observable Observable, assertions ...Assertion) {
 	ass := parseAssertions(assertions...)
@@ -185,6 +198,11 @@ func AssertThatSingle(t *testing.T, single Single, assertions ...Assertion) {
 	if checkHasRaisedError {
 		assert.Equal(t, value, err)
 	}
+
+	checkHasNotRaisedError := ass.hasNotRaisedAnErrorFunc()
+	if checkHasNotRaisedError {
+		assert.Nil(t, err)
+	}
 }
 
 // AssertThatOptionalSingle asserts the result of an OptionalSingle against a list of assertions.

assert_test.go

@@ -32,6 +32,10 @@ func TestAssertThatSingleError(t *testing.T) {
 		HasRaisedAnError(), HasRaisedError(errors.New("foo")))
 }
 
+func TestAssertThatSingleNotError(t *testing.T) {
+	AssertThatSingle(t, newSingleFrom(1), HasNotRaisedAnError())
+}
+
 func TestAssertThatOptionalSingleIsEmpty(t *testing.T) {
 	AssertThatOptionalSingle(t, newOptionalSingleFrom(optional.Empty()), IsEmpty())
 }

iterable/iterable_test.go

@@ -4,14 +4,9 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/reactivex/rxgo"
 	"github.com/stretchr/testify/assert"
 )
 
-func TestIterableImplementsIterator(t *testing.T) {
-	assert.Implements(t, (*rxgo.Iterator)(nil), Iterable(nil))
-}
-
 func TestCreateHomogenousIterable(t *testing.T) {
 	ch := make(chan interface{})
 	items := []interface{}{}

observable.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/reactivex/rxgo/errors"
 	"github.com/reactivex/rxgo/handlers"
+	"github.com/reactivex/rxgo/iterable"
 	"github.com/reactivex/rxgo/optional"
 	"github.com/reactivex/rxgo/options"
 )
@@ -53,6 +54,9 @@ type Observable interface {
 	Skip(nth uint) Observable
 	SkipLast(nth uint) Observable
 	SkipWhile(apply Predicate) Observable
+	StartWithItems(items ...interface{}) Observable
+	StartWithIterable(iterable iterable.Iterable) Observable
+	StartWithObservable(observable Observable) Observable
 	Subscribe(handler handlers.EventHandler, opts ...options.Option) Observer
 	SumFloat32() Single
 	SumFloat64() Single
@@ -1280,3 +1284,65 @@ func (o *observable) SumFloat64() Single {
 	}()
 	return NewSingleFromChannel(out)
 }
+
+// StartWithItems returns an Observable that emits the specified items before it begins to emit items emitted
+// by the source Observable.
+func (o *observable) StartWithItems(items ...interface{}) Observable {
+	out := make(chan interface{})
+	go func() {
+		for _, item := range items {
+			out <- item
+		}
+
+		for item := range o.ch {
+			out <- item
+		}
+
+		close(out)
+	}()
+	return &observable{ch: out}
+}
+
+// StartWithIterable returns an Observable that emits the items in a specified Iterable before it begins to
+// emit items emitted by the source Observable.
+func (o *observable) StartWithIterable(iterable iterable.Iterable) Observable {
+	out := make(chan interface{})
+	go func() {
+		for {
+			item, err := iterable.Next()
+			if err != nil {
+				break
+			}
+			out <- item
+		}
+
+		for item := range o.ch {
+			out <- item
+		}
+
+		close(out)
+	}()
+	return &observable{ch: out}
+}
+
+// StartWithObservable returns an Observable that emits the items in a specified Observable before it begins to
+// emit items emitted by the source Observable.
+func (o *observable) StartWithObservable(obs Observable) Observable {
+	out := make(chan interface{})
+	go func() {
+		for {
+			item, err := obs.Next()
+			if err != nil {
+				break
+			}
+			out <- item
+		}
+
+		for item := range o.ch {
+			out <- item
+		}
+
+		close(out)
+	}()
+	return &observable{ch: out}
+}

observable_test.go

@@ -1399,15 +1399,11 @@ func TestAll(t *testing.T) {
 		}
 	}
 
-	got1, err := Just(1, 2, 3).All(predicateAllInt).
-		Subscribe(nil).Block()
-	assert.Nil(t, err)
-	assert.Equal(t, true, got1)
+	AssertThatSingle(t, Just(1, 2, 3).All(predicateAllInt),
+		HasValue(true), HasNotRaisedAnError())
 
-	got2, err := Just(1, "x", 3).All(predicateAllInt).
-		Subscribe(nil).Block()
-	assert.Nil(t, err)
-	assert.Equal(t, false, got2)
+	AssertThatSingle(t, Just(1, "x", 3).All(predicateAllInt),
+		HasValue(false), HasNotRaisedAnError())
 }
 
 func TestContain(t *testing.T) {
@@ -1805,3 +1801,87 @@ func TestSumFloat64(t *testing.T) {
 	AssertThatSingle(t, Just("x").SumFloat64(), HasRaisedAnError())
 	AssertThatSingle(t, Empty().SumFloat64(), HasValue(float64(0)))
 }
+
+func TestStartWithItems(t *testing.T) {
+	obs := Just(1, 2, 3).StartWithItems(10, 20)
+	AssertThatObservable(t, obs, HasItems(10, 20, 1, 2, 3))
+}
+
+func TestStartWithItemsWithError(t *testing.T) {
+	obs := Just(1, 2, 3).StartWithItems(10, errors.New(""))
+	AssertThatObservable(t, obs, HasItems(10), HasRaisedError(errors.New("")))
+}
+
+func TestStartWithItemsFromEmpty(t *testing.T) {
+	obs := Empty().StartWithItems(1, 2)
+	AssertThatObservable(t, obs, HasItems(1, 2))
+}
+
+func TestStartWithItemsWithoutItems(t *testing.T) {
+	obs := Just(1, 2, 3).StartWithItems()
+	AssertThatObservable(t, obs, HasItems(1, 2, 3))
+}
+
+func TestStartWithIterable(t *testing.T) {
+	ch := make(chan interface{})
+	it, err := iterable.New(ch)
+	if err != nil {
+		t.Fail()
+	}
+	obs := Just(1, 2, 3).StartWithIterable(it)
+	ch <- 10
+	close(ch)
+	AssertThatObservable(t, obs, HasItems(10, 1, 2, 3))
+}
+
+func TestStartWithIterableWithError(t *testing.T) {
+	ch := make(chan interface{})
+	it, err := iterable.New(ch)
+	if err != nil {
+		t.Fail()
+	}
+	obs := Just(1, 2, 3).StartWithIterable(it)
+	ch <- errors.New("")
+	close(ch)
+	AssertThatObservable(t, obs, IsEmpty(), HasRaisedError(errors.New("")))
+}
+
+func TestStartWithIterableFromEmpty(t *testing.T) {
+	ch := make(chan interface{})
+	it, err := iterable.New(ch)
+	if err != nil {
+		t.Fail()
+	}
+	obs := Empty().StartWithIterable(it)
+	ch <- 1
+	close(ch)
+	AssertThatObservable(t, obs, HasItems(1))
+}
+
+func TestStartWithIterableWithoutItems(t *testing.T) {
+	ch := make(chan interface{})
+	it, err := iterable.New(ch)
+	if err != nil {
+		t.Fail()
+	}
+	obs := Just(1, 2, 3).StartWithIterable(it)
+	close(ch)
+	AssertThatObservable(t, obs, HasItems(1, 2, 3))
+}
+
+func TestStartWithObservable(t *testing.T) {
+	obs := Just(10, 20)
+	just := Just(1, 2, 3).StartWithObservable(obs)
+	AssertThatObservable(t, just, HasItems(10, 20, 1, 2, 3))
+}
+
+func TestStartWithObservableWithError(t *testing.T) {
+	obs := Just(10, errors.New(""))
+	just := Just(1, 2, 3).StartWithObservable(obs)
+	AssertThatObservable(t, just, HasItems(10), HasRaisedError(errors.New("")))
+}
+
+func TestStartWithObservableFromEmpty(t *testing.T) {
+	obs := Just(1, 2, 3).StartWithObservable(Empty())
+	AssertThatObservable(t, obs, HasItems(1, 2, 3))
+}