Helper for buffering data from non-blocking channel for sending into blocking channel

chan_queue.go

// Usage example:
func process(in <-chan Msg, outBlocking chan<- Msg) {
        out := newQueueMsg(QueueSize, outBlocking)
        for {
                select {
                case msg := <-in:
                        out.append(msg)
                case out.C <- out.Elem:
                        out.del()
                }
        }
}

// Helper:
type queueMsg struct {
	C     chan<- Msg
	Elem  Msg
	Queue []Msg
	c     chan<- Msg
}

func newQueueMsg(size int, out chan<- Msg) *queueMsg {
	return &queueMsg{
		c:     out,
		Queue: make([]Msg, 0, size),
	}
}

func (q *queueMsg) add(elem Msg) bool {
	if len(q.Queue) == cap(q.Queue) {
		return false
	}
	q.append(elem)
	return true
}

func (q *queueMsg) append(elem Msg) {
	if len(q.Queue) == 0 {
		q.C = q.c
		q.Elem = elem
	}
	q.Queue = append(q.Queue, elem)
}

func (q *queueMsg) del() {
	copy(q.Queue, q.Queue[1:])
	q.Queue = q.Queue[:len(q.Queue)-1]
	if len(q.Queue) == 0 {
		q.C = nil
	} else {
		q.Elem = q.Queue[0]
	}
}