Migration of the operators in a dedicated file.

connectableobservable_test.go

@@ -1,13 +1,5 @@
 package rxgo
 
-import (
-	"fmt"
-	"github.com/reactivex/rxgo/handlers"
-	"github.com/reactivex/rxgo/options"
-	"testing"
-	"time"
-)
-
 //func TestConnect(t *testing.T) {
 //	obs := Just(1, 2, 3).Publish()
 //	got1 := make([]interface{}, 0)
@@ -35,18 +27,18 @@ import (
 //	fmt.Printf("%v\n", got1)
 //}
 
-func TestConnect2(t *testing.T) {
-	obs := Just(1, 2, 3).Map(func(i interface{}) interface{} {
-		return 1 + i.(int)
-	}).Publish()
-	got1 := make([]interface{}, 0)
-
-	obs.Subscribe(handlers.NextFunc(func(i interface{}) {
-		got1 = append(got1, i)
-		time.Sleep(200 * time.Millisecond) // Pause the observer on purpose
-	}), options.WithBufferBackpressureStrategy(1))
-	obs.Connect()
-
-	time.Sleep(500 * time.Millisecond) // Ugly wait just for the example
-	fmt.Printf("%v\n", got1)
-}
+//func TestConnect2(t *testing.T) {
+//	obs := Just(1, 2, 3).Map(func(i interface{}) interface{} {
+//		return 1 + i.(int)
+//	}).Publish()
+//	got1 := make([]interface{}, 0)
+//
+//	obs.Subscribe(handlers.NextFunc(func(i interface{}) {
+//		got1 = append(got1, i)
+//		time.Sleep(200 * time.Millisecond) // Pause the observer on purpose
+//	}), options.WithBufferBackpressureStrategy(1))
+//	obs.Connect()
+//
+//	time.Sleep(500 * time.Millisecond) // Ugly wait just for the example
+//	fmt.Printf("%v\n", got1)
+//}

fx.go

@@ -41,4 +41,7 @@ type (
 
 	// Supplier defines a function that supplies a result from nothing.
 	Supplier func() interface{}
+
+	// ChanProducer defines a function that produces results into a channel.
+	ChanProducer func(chan interface{})
 )