gistfile1.go

// batchChannel takes in a channel and outputs them grouped into batches. The batch will be sent when
// either there are enough to complete the batchSize or the flushInterval has elapsed.
func batchChannel(in chan interface{}, out chan interface{}, batchSize int, flushInterval time.Duration) {
	tickChan := time.Tick(flushInterval)
	buffer := make([]interface{}, 0)

	for {
		select {
		case el := <-in:
			buffer = append(buffer, el)
			if len(buffer) >= batchSize {
				out <- buffer
				buffer = make([]interface{}, 0)
			}

		case <-tickChan:
			if len(buffer) > 0 {
				out <- buffer
				buffer = make([]interface{}, 0)
			}
		}
	}
}