Refine time operator

Author: teivah

README.md

@@ -277,7 +277,7 @@ In this example, we create a pool of 32 goroutines that consume items concurrent
 * [Range](doc/range.md) — create an Observable that emits a range of sequential integers
 * [Repeat](doc/repeat.md) — create an Observable that emits a particular item or sequence of items repeatedly
 * [Start](doc/start.md) — create an Observable that emits the return value of a function
-* [Timer](doc/timer.md) — create an Observable that emits a single item after a given delay
+* [Timer](doc/timer.md) — create an Observable that completes after a specified delay
 
 ### Transforming Observables
 * [Buffer](doc/buffer.md) — periodically gather items from an Observable into bundles and emit these bundles rather than emitting the items one at a time

doc/timer.md

@@ -2,7 +2,7 @@
 
 ## Overview
 
-Create an Observable that emits a single item after a given delay.
+Create an Observable that completes after a specified delay.
 
 ![](http://reactivex.io/documentation/operators/images/timer.png)
 

factory.go

@@ -336,7 +336,7 @@ func Thrown(err error) Observable {
 	}
 }
 
-// Timer returns an Observable that emits an empty structure after a specified delay, and then completes.
+// Timer returns an Observable that completes after a specified delay.
 func Timer(d Duration, opts ...Option) Observable {
 	option := parseOptions(opts...)
 	next := make(chan Item, 1)
@@ -348,7 +348,7 @@ func Timer(d Duration, opts ...Option) Observable {
 		case <-ctx.Done():
 			return
 		case <-time.After(d.duration()):
-			next <- Of(struct{}{})
+			return
 		}
 	}()
 	return &ObservableImpl{

factory_test.go

@@ -385,7 +385,11 @@ func Test_Thrown(t *testing.T) {
 
 func Test_Timer(t *testing.T) {
 	obs := Timer(WithDuration(time.Nanosecond))
-	Assert(context.Background(), t, obs, IsNotEmpty())
+	select {
+	case <-time.Tick(time.Second):
+		assert.FailNow(t, "observable not closed")
+	case <-obs.Observe():
+	}
 }
 
 func Test_Timer_Empty(t *testing.T) {
@@ -395,5 +399,9 @@ func Test_Timer_Empty(t *testing.T) {
 		time.Sleep(50 * time.Millisecond)
 		cancel()
 	}()
-	Assert(context.Background(), t, obs, IsEmpty())
+	select {
+	case <-time.Tick(time.Second):
+		assert.FailNow(t, "observable not closed")
+	case <-obs.Observe():
+	}
 }