Method batchingBuffer

batching_channel.go:54–87

()

}

func (ch *BatchingChannel) batchingBuffer() {
	var input, output, nextInput chan interface{}
	nextInput = ch.input
	input = nextInput

	for input != nil || output != nil {
		select {
		case elem, open := <-input:
			if open {
				ch.buffer = append(ch.buffer, elem)
			} else {
				input = nil
				nextInput = nil
			}
		case output <- ch.buffer:
			ch.buffer = nil
		case ch.length <- len(ch.buffer):
		}

		if len(ch.buffer) == 0 {
			input = nextInput
			output = nil
		} else if ch.size != Infinity && len(ch.buffer) >= int(ch.size) {
			input = nil
			output = ch.output
		} else {
			input = nextInput
			output = ch.output
		}
	}

	close(ch.output)
	close(ch.length)
}