ring_buffer.go

// RingBuffer reads messages from in and writes them to out.
// Whenever out is full, it will remove the oldest message to make room.
// Adapted from https://blog.pivotal.io/labs/labs/a-concurrent-ring-buffer-for-go.
func RingBuffer(in <-chan Message, size int) <-chan Message {
	out := make(chan Message, size)
	go func() {
		defer close(out)
		for m := range in {
			select {
			case out <- m:
			default:
				// If out is full, read a message from the end to make room.
				select {
				case <-out:
				default:
					// Avoid a deadlock in case the buffer has since been drained.
				}
				out <- m
			}
		}
	}()
	return out
}