feat(channels): add ToChannel + Generator + Batch + BatchWithTimeout …

…(WIP)

channel.go

@@ -149,3 +149,75 @@ func DispatchingStrategyMost[T any](msg T, index uint64, channels []<-chan T) in
 		return len(channels[item]) > len(channels[max]) && channelIsNotFull(channels[item])
 	})
 }
+
+// ToChannel returns a read-only channels of collection elements.
+func ToChannel[T any](bufferSize int, collection []T) <-chan T {
+	ch := make(chan T, bufferSize)
+
+	go func() {
+		for _, item := range collection {
+			ch <- item
+		}
+
+		close(ch)
+	}()
+
+	return ch
+}
+
+// Generator implements the generator design pattern.
+func Generator[T any](bufferSize int, generator func(yield func(T))) <-chan T {
+	ch := make(chan T, bufferSize)
+
+	go func() {
+		// WARNING: infinite loop
+		generator(func(t T) {
+			ch <- t
+		})
+
+		close(ch)
+	}()
+
+	return ch
+}
+
+// Batch creates a slice of n elements from a channel. Returns the slice and the slice length.
+func Batch[T any](ch <-chan T, size int) (collection []T, length int) {
+	buffer := make([]T, 0, size)
+	index := 0
+
+	for ; index < size; index++ {
+		item, ok := <-ch
+		if !ok {
+			return buffer, index
+		}
+
+		buffer = append(buffer, item)
+	}
+
+	return buffer, index
+}
+
+// BatchWithTimeout creates a slice of n elements from a channel, with timeout. Returns the slice and the slice length.
+func BatchWithTimeout[T any](ch <-chan T, size int, timeout time.Duration) (collection []T, length int) {
+	expire := time.After(timeout)
+
+	buffer := make([]T, 0, size)
+	index := 0
+
+	for ; index < size; index++ {
+		select {
+		case item, ok := <-ch:
+			if !ok {
+				return buffer, index
+			}
+
+			buffer = append(buffer, item)
+
+		case <-expire:
+			return buffer, index
+		}
+	}
+
+	return buffer, index
+}

channel_test.go

@@ -187,3 +187,99 @@ func TestDispatchingStrategyMost(t *testing.T) {
 	children[1] <- 1
 	is.Equal(0, DispatchingStrategyMost(42, 0, rochildren))
 }
+
+func TestToChannel(t *testing.T) {
+	t.Parallel()
+	testWithTimeout(t, 10*time.Millisecond)
+	is := assert.New(t)
+
+	ch := ToChannel[int](2, []int{1, 2, 3})
+
+	r1, ok1 := <-ch
+	r2, ok2 := <-ch
+	r3, ok3 := <-ch
+	is.True(ok1)
+	is.Equal(1, r1)
+	is.True(ok2)
+	is.Equal(2, r2)
+	is.True(ok3)
+	is.Equal(3, r3)
+
+	_, ok4 := <-ch
+	is.False(ok4)
+}
+
+func TestGenerate(t *testing.T) {
+	t.Parallel()
+	testWithTimeout(t, 10*time.Millisecond)
+	is := assert.New(t)
+
+	gen := func(yield func(int)) {
+		yield(0)
+		yield(1)
+		yield(2)
+		yield(3)
+	}
+
+	i := 0
+
+	for v := range Generator(2, gen) {
+		is.Equal(i, v)
+		i++
+	}
+
+	is.Equal(i, 4)
+}
+
+func TestBatch(t *testing.T) {
+	t.Parallel()
+	testWithTimeout(t, 10*time.Millisecond)
+	is := assert.New(t)
+
+	ch := ToChannel(2, []int{1, 2, 3})
+
+	items1, length1 := Batch(ch, 2)
+	items2, length2 := Batch(ch, 2)
+	items3, length3 := Batch(ch, 2)
+
+	is.Equal([]int{1, 2}, items1)
+	is.Equal(2, length1)
+	is.Equal([]int{3}, items2)
+	is.Equal(1, length2)
+	is.Equal([]int{}, items3)
+	is.Equal(0, length3)
+}
+
+func TestBatchWithTimeout(t *testing.T) {
+	t.Parallel()
+	testWithTimeout(t, 200*time.Millisecond)
+	is := assert.New(t)
+
+	ch := make(chan int)
+	go func() {
+		for i := 0; i < 5; i++ {
+			ch <- i
+			time.Sleep(10 * time.Millisecond)
+		}
+	}()
+
+	items1, length1 := BatchWithTimeout(ch, 20, 15*time.Millisecond)
+	is.Equal([]int{0, 1}, items1)
+	is.Equal(2, length1)
+
+	items2, length2 := BatchWithTimeout(ch, 20, 2*time.Millisecond)
+	is.Equal([]int{}, items2)
+	is.Equal(0, length2)
+
+	items3, length3 := BatchWithTimeout(ch, 1, 30*time.Millisecond)
+	is.Equal([]int{2}, items3)
+	is.Equal(1, length3)
+
+	items4, length4 := BatchWithTimeout(ch, 2, 25*time.Millisecond)
+	is.Equal([]int{3, 4}, items4)
+	is.Equal(2, length4)
+
+	items5, length5 := BatchWithTimeout(ch, 3, 25*time.Millisecond)
+	is.Equal([]int{}, items5)
+	is.Equal(0, length5)
+}