Go - coalescing messages

coalescing

http://play.golang.org/p/ORXhGHM5xV

package main

import "fmt"
import "time"

type msg string

func main() {
	c := make(chan msg)
	flush := make(chan chan int)
	go coalesce(c, flush)

	c <- "one"
	c <- "two"

	<-time.After(15 * time.Second)
	c <- "three"
	done := make(chan int)
	flush <- done
	<-done
}

func coalesce(in <-chan msg, flush <-chan chan int) {
	msgs := make([]msg, 0)
	t := time.NewTimer(0)
	var timerCh <-chan time.Time

	for {
		select {
		case m := <-in:
			fmt.Printf("BATCHING %s\n", m)
			msgs = append(msgs, m)
			if timerCh == nil {
				t.Reset(3 * time.Second)
				timerCh = t.C
			}
		case <-timerCh:
			fmt.Println("TIMER")
			go do_something_with(msgs)
			msgs = make([]msg, 0)
			timerCh = nil
		case done := <-flush:
			fmt.Println("FLUSHING")
			do_something_with(msgs)
			msgs = make([]msg, 0)
			timerCh = nil
			fmt.Println("FLUSHED!")
			done <- 1

		}
	}

}

func do_something_with(msgs []msg) {
	if len(msgs) == 0 {
		return
	}

	fmt.Println("BEGIN DOSOMETHING")
	<-time.After(1 * time.Second)
	for _, m := range msgs {
		fmt.Printf("  MESSAGE: %s\n", string(m))
	}
	fmt.Println("END DOSOMETHING")
}