Infinite channel buffer without "sync"

infinite.go

// assume the following exist and are set to unbuffered channels:
// var (
//    next chan Type
//    in   chan Type
// )

        go func() {
                defer close(next)

                // pending events (this is the "infinite" part)
                pending := []Type{}

        recv:
                for {
                        // Ensure that pending always has values so the select can
                        // multiplex between the receiver and sender properly
                        if len(pending) == 0 {
                                v, ok := <-in
                                if !ok {
                                        // in is closed, flush values
                                        break
                                }

                                pending = append(pending, v)
                        }

                        select {
                        // Queue incoming values
                        case v, ok := <-in:
                                if !ok {
                                        // in is closed, flush values
                                        break recv
                                }
                                pending = append(pending, v)
                        // Send values to the outer function
                        case next <- pending[0]:
                                pending = pending[1:]
                        }
                }

                // After in is closed, we may still have events to send
                for _, v := range pending {
                        next <- v
                }
        }()