fix: fx/fn.Head func will forever block when n is less than 1 (zeromi…

…cro#128)

* fix fx/Stream Head func will forever block when n is less than 1

* update test case

* update test case

core/fx/fn.go

@@ -159,6 +159,9 @@ func (p Stream) Group(fn KeyFunc) Stream {
 }
 
 func (p Stream) Head(n int64) Stream {
+	if n < 1 {
+		panic("n must be greater than 0")
+	}
 	source := make(chan interface{})
 
 	go func() {

core/fx/fn_test.go

@@ -169,6 +169,14 @@ func TestHead(t *testing.T) {
 	assert.Equal(t, 3, result)
 }
 
+func TestHeadZero(t *testing.T) {
+	assert.Panics(t, func() {
+		Just(1, 2, 3, 4).Head(0).Reduce(func(pipe <-chan interface{}) (interface{}, error) {
+			return nil, nil
+		})
+	})
+}
+
 func TestHeadMore(t *testing.T) {
 	var result int
 	Just(1, 2, 3, 4).Head(6).Reduce(func(pipe <-chan interface{}) (interface{}, error) {