batching channel implementation

batchchan.go

package batchchan

import "github.com/cheekybits/genny/generic"

import (
	"fmt"
)

type T generic.Type

func batchTChan(
	inputc <-chan T,
	outputc chan []T,
	batchsize int,
) {
	if batchsize <= 1 {
		panic(fmt.Sprintf("invalid batch size: %d", batchsize))
	}

	var values []T
Main:
	for {
		v, ok := <-inputc
		if !ok {
			return
		}
		values = append(values, v)

	Consume:
		for len(values) < batchsize {
			select {
			case v, ok := <-inputc:
				if !ok {
					break Main
				}
				values = append(values, v)
				continue Consume
			default:
			}

			select {
			case v, ok := <-inputc:
				if !ok {
					break Main
				}
				values = append(values, v)
				continue Consume
			case outputc <- values:
				values = nil
				continue Main
			}
		}
		outputc <- values
		values = nil
	}
	outputc <- values
}