Channel with an unrestricted buffer in the middle

elastic.go

// It turns out this is almost identical to Kyle Lemons's solution at
// https://github.com/kylelemons/iq/blob/master/iq_slice.go
// If you're reading this, go use his. This one has a bug where it doesn't flush the pending
// buffer once the input is closed.

type Message struct {
	Value string
}

func MakeElastic() (<-chan Message, chan<- Message) {
	input := make(chan Message, 1)
	output := make(chan Message, 1)
	go func(input <-chan Message, output chan<- Message) {
		defer close(output)
		var buffer []Message
		for {
			// we need two code paths, one if we have any buffered items, and one if we don't
			if len(buffer) > 0 {
				select {
				case msg, ok := <-input:
					if !ok {
						return
					}
					// we have a new message, stick it on the buffer
					buffer = append(buffer, msg)
				case output <- buffer[0]:
					// we've successfully sent an item from the buffer
					buffer = buffer[1:]
				}
			} else {
				// we have no buffered items, so just wait for input
				msg, ok := <-input
				if !ok {
					return
				}
				// try to send immediately, and stick it in the buffer otherwise
				select {
				case output <- msg:
				default:
					buffer = []Message{msg}
				}
			}
		}
	}(input, output)
	return output, input
}