Merge f0669f3 into 613168f

assert.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 
 	"github.com/reactivex/rxgo/handlers"
-	"github.com/reactivex/rxgo/optional"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -215,7 +214,7 @@ func AssertThatOptionalSingle(t *testing.T, optionalSingle OptionalSingle, asser
 		assert.Fail(t, "error while retrieving OptionalSingle results")
 	}
 
-	if optional, ok := v.(optional.Optional); ok {
+	if optional, ok := v.(Optional); ok {
 		checkIsEmpty, empty := ass.isEmptyFunc()
 		if checkIsEmpty {
 			if empty {

assert_test.go

@@ -3,8 +3,6 @@ package rxgo
 import (
 	"errors"
 	"testing"
-
-	"github.com/reactivex/rxgo/optional"
 )
 
 func TestAssertThatObservableHasItems(t *testing.T) {
@@ -37,21 +35,21 @@ func TestAssertThatSingleNotError(t *testing.T) {
 }
 
 func TestAssertThatOptionalSingleIsEmpty(t *testing.T) {
-	AssertThatOptionalSingle(t, newOptionalSingleFrom(optional.Empty()), IsEmpty())
+	AssertThatOptionalSingle(t, newOptionalSingleFrom(EmptyOptional()), IsEmpty())
 }
 
 func TestAssertThatOptionalSingleIsNotEmpty(t *testing.T) {
-	AssertThatOptionalSingle(t, newOptionalSingleFrom(optional.Of(1)), IsNotEmpty())
+	AssertThatOptionalSingle(t, newOptionalSingleFrom(Of(1)), IsNotEmpty())
 }
 
 func TestAssertThatOptionalSingleHasValue(t *testing.T) {
-	AssertThatOptionalSingle(t, newOptionalSingleFrom(optional.Of(1)), HasValue(1))
+	AssertThatOptionalSingle(t, newOptionalSingleFrom(Of(1)), HasValue(1))
 }
 
 func TestAssertThatOptionalSingleHasRaisedAnError(t *testing.T) {
-	AssertThatOptionalSingle(t, newOptionalSingleFrom(optional.Of(errors.New("foo"))), HasRaisedAnError())
+	AssertThatOptionalSingle(t, newOptionalSingleFrom(Of(errors.New("foo"))), HasRaisedAnError())
 }
 
 func TestAssertThatOptionalSingleHasRaisedError(t *testing.T) {
-	AssertThatOptionalSingle(t, newOptionalSingleFrom(optional.Of(errors.New("foo"))), HasRaisedError(errors.New("foo")))
+	AssertThatOptionalSingle(t, newOptionalSingleFrom(Of(errors.New("foo"))), HasRaisedError(errors.New("foo")))
 }

connectableobservable.go

@@ -1,6 +1,7 @@
 package rxgo
 
 import (
+	"context"
 	"sync"
 
 	"github.com/reactivex/rxgo/handlers"
@@ -24,8 +25,8 @@ func newConnectableObservableFromObservable(observable Observable) ConnectableOb
 	}
 }
 
-func (c *connectableObservable) Iterator() Iterator {
-	return c.observable.Iterator()
+func (c *connectableObservable) Iterator(ctx context.Context) Iterator {
+	return c.observable.Iterator(context.Background())
 }
 
 func (c *connectableObservable) All(predicate Predicate) Single {
@@ -75,9 +76,9 @@ func (c *connectableObservable) BufferWithTimeOrCount(timespan Duration, count i
 func (c *connectableObservable) Connect() Observer {
 	out := NewObserver()
 	go func() {
-		it := c.observable.Iterator()
+		it := c.observable.Iterator(context.Background())
 		for {
-			if item, err := it.Next(); err == nil {
+			if item, err := it.Next(context.Background()); err == nil {
 				c.observersMutex.Lock()
 				for _, observer := range c.observers {
 					c.observersMutex.Unlock()
@@ -284,6 +285,10 @@ func (c *connectableObservable) TakeWhile(apply Predicate) Observable {
 	return c.observable.TakeWhile(apply)
 }
 
+func (c *connectableObservable) Timeout(duration Duration) Observable {
+	return c.observable.Timeout(duration)
+}
+
 func (c *connectableObservable) ToChannel(opts ...options.Option) Channel {
 	return c.observable.ToChannel(opts...)
 }

errors.go

@@ -0,0 +1,50 @@
+package rxgo
+
+// CancelledIteratorError is triggered when an iterator is canceled
+type CancelledIteratorError struct {
+}
+
+// EndOfIteratorError is triggered when an iterator is complete
+type EndOfIteratorError struct {
+}
+
+// IllegalInputError is triggered when the observable receives an illegal input
+type IllegalInputError struct {
+	reason string
+}
+
+// IndexOutOfBoundError is triggered when the observable cannot access to the specified index
+type IndexOutOfBoundError struct {
+}
+
+// NoSuchElementError is triggered when an optional does not contain any element
+type NoSuchElementError struct {
+}
+
+// TimeoutError is triggered when a timeout occurs
+type TimeoutError struct {
+}
+
+func (e *CancelledIteratorError) Error() string {
+	return "CancelledIteratorError"
+}
+
+func (e *EndOfIteratorError) Error() string {
+	return "EndOfIteratorError"
+}
+
+func (e *IllegalInputError) Error() string {
+	return e.reason
+}
+
+func (e *IndexOutOfBoundError) Error() string {
+	return "IndexOutOfBoundError"
+}
+
+func (e *NoSuchElementError) Error() string {
+	return "NoSuchElementError"
+}
+
+func (e *TimeoutError) Error() string {
+	return "TimeoutError"
+}