Infinitely Buffered Channel

chan.go

/*
This is an infinitely buffered channel.
*/
package main

import (
	"fmt"
	"sync"
	"time"
)

type Channel struct {
	In     chan interface{}
	Out    chan interface{}
	buffer []interface{}
}

func NewChannel() *Channel {
	ch := new(Channel)
	ch.In = make(chan interface{}, 0)
	ch.Out = make(chan interface{}, 0)
	ch.buffer = make([]interface{}, 0)
	go ch.buffering()
	return ch
}

func (ch *Channel) buffering() {
	for {
		if len(ch.buffer) > 0 {
			select {
			case in, open := <-ch.In:
				if !open {
					ch.drain()
					return
				}
				ch.buffer = append(ch.buffer, in)
			case ch.Out <- ch.buffer[0]:
				ch.buffer = ch.buffer[1:]
			}
		} else {
			in, open := <-ch.In
			if !open {
				ch.drain()
				return
			}
			ch.buffer = append(ch.buffer, in)
		}
	}
}

func (ch *Channel) drain() {
	for i := 0; i < len(ch.buffer); i++ {
		ch.Out <- ch.buffer[i]
	}
	ch.buffer = ch.buffer[:]
	close(ch.Out)
}

func main() {
	ch := NewChannel()
	ch.In <- "Hello"
	ch.In <- "World"
	ch.In <- "Fantastic"

	// simulate worker
	wg := new(sync.WaitGroup)
	wg.Add(1)
	go func() {
		defer wg.Done()
		for n := range ch.Out {
			fmt.Println(n)
		}
	}()
	time.Sleep(1 * time.Second)

	ch.In <- 1
	ch.In <- 2
	ch.In <- 3
	close(ch.In)
	wg.Wait()
}