Code quality

assert_test.go → assert.go

@@ -2,20 +2,22 @@ package rxgo
 
 import (
 	"context"
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
-type rxAssert interface {
-	apply(*rxAssertImpl)
+// RxAssert lists the Observable assertions.
+type RxAssert interface {
+	apply(*rxAssert)
 	itemsToBeChecked() (bool, []interface{})
 	noItemsToBeChecked() bool
 	raisedErrorToBeChecked() (bool, error)
 	notRaisedErrorToBeChecked() bool
 }
 
-type rxAssertImpl struct {
-	f                      func(*rxAssertImpl)
+type rxAssert struct {
+	f                      func(*rxAssert)
 	checkHasItems          bool
 	checkHasNoItems        bool
 	items                  []interface{}
@@ -24,67 +26,72 @@ type rxAssertImpl struct {
 	checkHasNotRaisedError bool
 }
 
-func (ass *rxAssertImpl) apply(do *rxAssertImpl) {
+func (ass *rxAssert) apply(do *rxAssert) {
 	ass.f(do)
 }
 
-func (ass *rxAssertImpl) itemsToBeChecked() (bool, []interface{}) {
+func (ass *rxAssert) itemsToBeChecked() (bool, []interface{}) {
 	return ass.checkHasItems, ass.items
 }
 
-func (ass *rxAssertImpl) noItemsToBeChecked() bool {
+func (ass *rxAssert) noItemsToBeChecked() bool {
 	return ass.checkHasNoItems
 }
 
-func (ass *rxAssertImpl) raisedErrorToBeChecked() (bool, error) {
+func (ass *rxAssert) raisedErrorToBeChecked() (bool, error) {
 	return ass.checkHasRaisedError, ass.error
 }
 
-func (ass *rxAssertImpl) notRaisedErrorToBeChecked() bool {
+func (ass *rxAssert) notRaisedErrorToBeChecked() bool {
 	return ass.checkHasNotRaisedError
 }
 
-func newAssertion(f func(*rxAssertImpl)) *rxAssertImpl {
-	return &rxAssertImpl{
+func newAssertion(f func(*rxAssert)) *rxAssert {
+	return &rxAssert{
 		f: f,
 	}
 }
 
-func hasItems(items ...interface{}) rxAssert {
-	return newAssertion(func(a *rxAssertImpl) {
+// HasItems checks that the observable produces the corresponding items.
+func HasItems(items ...interface{}) RxAssert {
+	return newAssertion(func(a *rxAssert) {
 		a.checkHasItems = true
 		a.items = items
 	})
 }
 
-func hasNoItems() rxAssert {
-	return newAssertion(func(a *rxAssertImpl) {
+// HasNoItems checks that the observable has not produce any item.
+func HasNoItems() RxAssert {
+	return newAssertion(func(a *rxAssert) {
 		a.checkHasNoItems = true
 	})
 }
 
-func hasRaisedError(err error) rxAssert {
-	return newAssertion(func(a *rxAssertImpl) {
+// HasRaisedError checks that the observable has produce a specific error.
+func HasRaisedError(err error) RxAssert {
+	return newAssertion(func(a *rxAssert) {
 		a.checkHasRaisedError = true
 		a.error = err
 	})
 }
 
-func hasNotRaisedError() rxAssert {
-	return newAssertion(func(a *rxAssertImpl) {
+// HasNotRaisedError checks that the observable has not raised any error.
+func HasNotRaisedError() RxAssert {
+	return newAssertion(func(a *rxAssert) {
 		a.checkHasRaisedError = true
 	})
 }
 
-func parseAssertions(assertions ...rxAssert) rxAssert {
-	ass := new(rxAssertImpl)
+func parseAssertions(assertions ...RxAssert) RxAssert {
+	ass := new(rxAssert)
 	for _, assertion := range assertions {
 		assertion.apply(ass)
 	}
 	return ass
 }
 
-func assertObservable(t *testing.T, ctx context.Context, observable Observable, assertions ...rxAssert) {
+// AssertObservable asserts the result of an Observable against a list of assertions.
+func AssertObservable(ctx context.Context, t *testing.T, observable Observable, assertions ...RxAssert) {
 	ass := parseAssertions(assertions...)
 
 	got := make([]interface{}, 0)

factory.go

@@ -1,5 +1,6 @@
 package rxgo
 
+// Empty creates an Observable with no item and terminate immediately.
 func Empty() Observable {
 	next := make(chan Item)
 	close(next)
@@ -8,12 +9,14 @@ func Empty() Observable {
 	}
 }
 
+// FromChannel creates a cold observable from a channel.
 func FromChannel(next <-chan Item) Observable {
 	return &observable{
 		iterable: newChannelIterable(next),
 	}
 }
 
+// Just creates an Observable with the provided items.
 func Just(item Item, items ...Item) Observable {
 	if len(items) > 0 {
 		items = append([]Item{item}, items...)

factory_test.go

@@ -7,17 +7,17 @@ import (
 
 func Test_Empty(t *testing.T) {
 	obs := Empty()
-	assertObservable(t, context.Background(), obs, hasNoItems())
+	AssertObservable(context.Background(), t, obs, HasNoItems())
 }
 
 func Test_FromChannel(t *testing.T) {
 	next := channelValue(1, 2, 3, closeCmd)
 	obs := FromChannel(next)
-	assertObservable(t, context.Background(), obs, hasItems(1, 2, 3), hasNotRaisedError())
+	AssertObservable(context.Background(), t, obs, HasItems(1, 2, 3), HasNotRaisedError())
 }
 
 func Test_Just(t *testing.T) {
 	obs := Just(FromValue(1), FromValue(2), FromValue(3))
-	assertObservable(t, context.Background(), obs, hasItems(1, 2, 3), hasNotRaisedError())
-	assertObservable(t, context.Background(), obs, hasItems(1, 2, 3), hasNotRaisedError())
+	AssertObservable(context.Background(), t, obs, HasItems(1, 2, 3), HasNotRaisedError())
+	AssertObservable(context.Background(), t, obs, HasItems(1, 2, 3), HasNotRaisedError())
 }

iterable.go

@@ -1,5 +1,6 @@
 package rxgo
 
+// Iterable is the interface returning an iterable channel.
 type Iterable interface {
 	Next() <-chan Item
 }

observable.go

@@ -4,6 +4,7 @@ import (
 	"context"
 )
 
+// Observable is the basic observable interface.
 type Observable interface {
 	Iterable
 	Filter(ctx context.Context, apply Predicate) Observable

observable_test.go

@@ -2,26 +2,24 @@ package rxgo
 
 import (
 	"context"
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func Test_Filter(t *testing.T) {
 	obs := FromChannel(channelValue(1, 2, 3, 4, closeCmd)).Filter(context.Background(),
 		func(i interface{}) bool {
-			if i.(int)%2 == 0 {
-				return true
-			}
-			return false
+			return i.(int)%2 == 0
 		})
-	assertObservable(t, context.Background(), obs, hasItems(2, 4), hasNotRaisedError())
+	AssertObservable(context.Background(), t, obs, HasItems(2, 4), HasNotRaisedError())
 }
 
 func Test_ForEach(t *testing.T) {
 	count := 0
 	var gotErr error
 	done := make(chan struct{})
-	next := channelValue(1, 2, 3, fooErr)
+	next := channelValue(1, 2, 3, errFoo)
 
 	obs := FromChannel(next)
 	obs.ForEach(context.Background(), func(i interface{}) {
@@ -34,7 +32,7 @@ func Test_ForEach(t *testing.T) {
 	// We avoid using the assertion API on purpose
 	<-done
 	assert.Equal(t, 6, count)
-	assert.Equal(t, fooErr, gotErr)
+	assert.Equal(t, errFoo, gotErr)
 }
 
 func Test_Map_One(t *testing.T) {
@@ -43,7 +41,7 @@ func Test_Map_One(t *testing.T) {
 	obs := FromChannel(next).Map(context.Background(), func(i interface{}) (interface{}, error) {
 		return i.(int) + 1, nil
 	})
-	assertObservable(t, context.Background(), obs, hasItems(2, 3, 4), hasNotRaisedError())
+	AssertObservable(context.Background(), t, obs, HasItems(2, 3, 4), HasNotRaisedError())
 }
 
 func Test_Map_Multiple(t *testing.T) {
@@ -54,16 +52,16 @@ func Test_Map_Multiple(t *testing.T) {
 	}).Map(context.Background(), func(i interface{}) (interface{}, error) {
 		return i.(int) * 10, nil
 	})
-	assertObservable(t, context.Background(), obs, hasItems(20, 30, 40), hasNotRaisedError())
+	AssertObservable(context.Background(), t, obs, HasItems(20, 30, 40), HasNotRaisedError())
 }
 
 func Test_Map_Error(t *testing.T) {
-	next := channelValue(1, 2, 3, fooErr)
+	next := channelValue(1, 2, 3, errFoo)
 
 	obs := FromChannel(next).Map(context.Background(), func(i interface{}) (interface{}, error) {
 		return i.(int) + 1, nil
 	})
-	assertObservable(t, context.Background(), obs, hasItems(2, 3, 4), hasRaisedError(fooErr))
+	AssertObservable(context.Background(), t, obs, HasItems(2, 3, 4), HasRaisedError(errFoo))
 }
 
 func Test_Map_Cancel(t *testing.T) {
@@ -74,7 +72,7 @@ func Test_Map_Cancel(t *testing.T) {
 		return i.(int) + 1, nil
 	})
 	cancel()
-	assertObservable(t, context.Background(), obs, hasNoItems(), hasNotRaisedError())
+	AssertObservable(context.Background(), t, obs, HasNoItems(), HasNotRaisedError())
 }
 
 func Test_SkipWhile(t *testing.T) {
@@ -89,5 +87,5 @@ func Test_SkipWhile(t *testing.T) {
 		}
 	})
 
-	assertObservable(t, context.Background(), obs, hasItems(3, 4, 5), hasNotRaisedError())
+	AssertObservable(context.Background(), t, obs, HasItems(3, 4, 5), HasNotRaisedError())
 }

types.go

@@ -3,29 +3,40 @@ package rxgo
 import "context"
 
 type (
-	Func      func(interface{}) (interface{}, error)
-	Handler   func(ctx context.Context, src <-chan Item, dst chan<- Item)
-	Operator  func(item Item, dst chan<- Item, stop func())
+	// Func defines a function that computes a value from an input value.
+	Func func(interface{}) (interface{}, error)
+	// Handler defines a function implementing the handler logic for a stream.
+	Handler func(ctx context.Context, src <-chan Item, dst chan<- Item)
+	// Operator defines an operator function.
+	Operator func(item Item, dst chan<- Item, stop func())
+	// Predicate defines a func that returns a bool from an input value.
 	Predicate func(interface{}) bool
 
+	// NextFunc handles a next item in a stream.
 	NextFunc func(interface{})
-	ErrFunc  func(error)
+	// ErrFunc handles an error in a stream.
+	ErrFunc func(error)
+	// DoneFunc handles the end of a stream.
 	DoneFunc func()
 
+	// Item is a wrapper having either a value or an error.
 	Item struct {
 		Value interface{}
 		Err   error
 	}
 )
 
+// IsError checks if an item is an error.
 func (i Item) IsError() bool {
 	return i.Err != nil
 }
 
+// FromValue creates an item from a value.
 func FromValue(i interface{}) Item {
 	return Item{Value: i}
 }
 
+// FromError creates an item from an error.
 func FromError(err error) Item {
 	return Item{Err: err}
 }

util_test.go

@@ -4,8 +4,10 @@ import (
 	"errors"
 )
 
-var fooErr = errors.New("foo")
-var closeCmd = &struct{}{}
+var (
+	errFoo   = errors.New("foo")
+	closeCmd = &struct{}{}
+)
 
 func channelValue(items ...interface{}) chan Item {
 	next := make(chan Item)