Support First

observable/observable.go

@@ -115,6 +115,19 @@ func (o Observable) Filter(apply fx.FilterableFunc) Observable {
 	return Observable(out)
 }
 
+// First returns new Observable which emit only first item.
+func (o Observable) First() Observable {
+	out := make(chan interface{})
+	go func() {
+		for item := range o {
+			out <- item
+			break
+		}
+		close(out)
+	}()
+	return Observable(out)
+}
+
 //Distinct supress duplicate items in the original Observable and returns
 // a new Observable.
 func (o Observable) Distinct(apply fx.KeySelectorFunc) Observable {

observable/observable_test.go

@@ -429,6 +429,48 @@ func TestObservableFilter(t *testing.T) {
 	assert.Exactly(t, []int{1, 2, 3, 7}, nums)
 }
 
+func TestObservableFirst(t *testing.T) {
+	items := []interface{}{0, 1, 3}
+	it, err := iterable.New(items)
+	if err != nil {
+		t.Fail()
+	}
+
+	stream1 := From(it)
+
+	stream2 := stream1.First()
+
+	nums := []int{}
+	onNext := handlers.NextFunc(func(item interface{}) {
+		if num, ok := item.(int); ok {
+			nums = append(nums, num)
+		}
+	})
+
+	sub := stream2.Subscribe(onNext)
+	<-sub
+
+	assert.Exactly(t, []int{0}, nums)
+}
+
+func TestObservableFirstWithEmpty(t *testing.T) {
+	stream1 := Empty()
+
+	stream2 := stream1.First()
+
+	nums := []int{}
+	onNext := handlers.NextFunc(func(item interface{}) {
+		if num, ok := item.(int); ok {
+			nums = append(nums, num)
+		}
+	})
+
+	sub := stream2.Subscribe(onNext)
+	<-sub
+
+	assert.Exactly(t, []int{}, nums)
+}
+
 func TestObservableDistinct(t *testing.T) {
 	items := []interface{}{1,2,2,1,3}
 	it, err := iterable.New(items)