Golint

assert.go

@@ -191,7 +191,6 @@ func AssertObservable(t *testing.T, observable Observable, assertions ...RxAsser
 	}, func(e error) {
 		err = e
 	}, nil).Block()
-
 	assertObservable(t, ass, got, err)
 }
 
@@ -213,6 +212,7 @@ func AssertObservableEventually(t *testing.T, observable Observable, timeout tim
 		chErr <- e
 	}, nil)
 	ctxTimeout, ctxTimeoutF := context.WithTimeout(context.Background(), timeout)
+	defer ctxTimeoutF()
 
 mainLoop:
 	for {
@@ -221,7 +221,6 @@ mainLoop:
 			if open {
 				got = append(got, item)
 			} else {
-				ctxTimeoutF()
 				break mainLoop
 			}
 		case e := <-chErr:

common_test.go

@@ -0,0 +1,7 @@
+package rxgo
+
+import (
+	"time"
+)
+
+const wait = 30 * time.Millisecond

flatmap.go

@@ -40,16 +40,19 @@ func flatObservedSequence(out chan interface{}, o Observable, apply func(interfa
 	for {
 		if item, err := it.Next(context.Background()); err == nil {
 			sequence := apply(item)
+			// TODO Error handling
 			sem.Acquire(ctx, 1)
 			go func() {
 				defer sem.Release(1)
+				// TODO Error handling
 				sequence.Subscribe(emissionObserver).Block()
 			}()
 		} else {
 			break
 		}
 	}
 
+	// TODO Error handling
 	sem.Acquire(ctx, int64(maxInParallel))
 }
 

flatmap_test.go

@@ -3,6 +3,8 @@ package rxgo
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/stretchr/testify/mock"
 )
 
@@ -17,7 +19,8 @@ func TestFlatMapCompletesWhenSequenceIsEmpty(t *testing.T) {
 	sequence = sequence.FlatMap(identity, 1)
 
 	// when subscribes to the sequence
-	sequence.Subscribe(emissionObserver.Capture()).Block()
+	err := sequence.Subscribe(emissionObserver.Capture()).Block()
+	assert.NoError(t, err)
 
 	// then completes without any emission
 	emissionObserver.AssertNotCalled(t, "OnNext", mock.Anything)
@@ -37,7 +40,8 @@ func TestFlatMapReturnsSameElementBecauseIdentifyApplied(t *testing.T) {
 	sequence = sequence.FlatMap(identity, 1)
 
 	// when subscribes to the sequence
-	sequence.Subscribe(emissionObserver.Capture()).Block()
+	err := sequence.Subscribe(emissionObserver.Capture()).Block()
+	assert.NoError(t, err)
 
 	// then completes with emission of the same element
 	emissionObserver.AssertNotCalled(t, "OnError", mock.Anything)
@@ -60,7 +64,8 @@ func TestFlatMapReturnsSliceElements(t *testing.T) {
 	sequence = sequence.FlatMap(flattenThreeElementSlice, 1)
 
 	// when subscribes to the sequence
-	sequence.Subscribe(emissionObserver.Capture()).Block()
+	err := sequence.Subscribe(emissionObserver.Capture()).Block()
+	assert.NoError(t, err)
 
 	// then completes with emission of flatten elements
 	emissionObserver.AssertNotCalled(t, "OnError", mock.Anything)
@@ -96,11 +101,6 @@ func TestFlatMapReturnsSliceElements(t *testing.T) {
 //	assert.Equal(t, uint(1), requestedMaxInParallel)
 //}
 
-var (
-	someElement  = "some element"
-	someSequence = Just(someElement)
-)
-
 func identity(el interface{}) Observable {
 	return Just(el)
 }

mock.go

@@ -3,11 +3,12 @@ package rxgo
 import (
 	"bufio"
 	"context"
-	"github.com/pkg/errors"
-	"github.com/stretchr/testify/mock"
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/mock"
 )
 
 const signalCh = byte(0)
@@ -40,20 +41,20 @@ type mockType struct {
 	index      int
 }
 
-func (m mockContext) Deadline() (deadline time.Time, ok bool) {
+func (m *mockContext) Deadline() (deadline time.Time, ok bool) {
 	panic("implement me")
 }
 
-func (m mockContext) Done() <-chan struct{} {
+func (m *mockContext) Done() <-chan struct{} {
 	outputs := m.Called()
 	return outputs.Get(0).(chan struct{})
 }
 
-func (m mockContext) Err() error {
+func (m *mockContext) Err() error {
 	panic("implement me")
 }
 
-func (m mockContext) Value(key interface{}) interface{} {
+func (m *mockContext) Value(key interface{}) interface{} {
 	panic("implement me")
 }
 
@@ -184,7 +185,7 @@ func causality(in string) ([]Observable, []context.Context) {
 		t := tasks[i]
 		index := t.index
 
-		if t.context == true {
+		if t.context {
 			ctx := contexts[index]
 			notif := make(chan struct{}, 1)
 			lastObservableType = -1

observable.go

@@ -447,7 +447,7 @@ func (o *observable) BufferWithCount(count, skip int) Observable {
 			return
 		}
 
-		buffer := make([]interface{}, count, count)
+		buffer := make([]interface{}, count)
 		iCount := 0
 		iSkip := 0
 		it := o.iterable.Iterator(context.Background())
@@ -472,7 +472,7 @@ func (o *observable) BufferWithCount(count, skip int) Observable {
 
 					if iSkip == skip { // Send current buffer
 						out <- buffer
-						buffer = make([]interface{}, count, count)
+						buffer = make([]interface{}, count)
 						iCount = 0
 						iSkip = 0
 					}