CastGroup, something that lets listeners subscribe to broadcast events. Built for the garage door; will be used to support hanging GETs on some event.

broadcast.go

package main

import (
  "log"
	"time"
)

func main() {
	c := NewCastGroup()

	go func() {
		log.Printf("golang 1")
		log.Printf("1.1 go: %v", c.Wait())
		log.Printf("1.2 go: %v", c.Wait())
	}()

	go func() {
		log.Printf("golang 2")
		log.Printf("2.1 go: %v", c.Wait())
	}()

	go func() {
		log.Printf("golang 3")
		log.Printf("3.1 go: %v", c.Wait())
		log.Printf("3.2 go: %v", c.Wait())
		log.Printf("3.3 go: %v", c.Wait())
		log.Printf("3.4 go: %v", c.Wait())
		log.Printf("3.5 go: %v", c.Wait())
		log.Printf("3.6 go: %v", c.Wait())
	}()

	count := 0
	for {
		time.Sleep(1 * time.Second)
		count++
		log.Printf("done event %d", count)
		c.Done(count)
	}
}

type CastGroup struct {
	result chan interface{}
	ch     chan *event
}

type event struct {
	wait   chan bool
	result interface{}
}

// Done signals all holders of the current event that the event is done; providing the
// given value in the event.
func (c *CastGroup) Done(v interface{}) {
	// the passed event is overloaded with the result of the previous event...
	c.result <- v
}

// Wait blocks until the current event is done, providing its result.
func (c *CastGroup) Wait() interface{} {
	current := <-c.ch
	_, ok := <-current.wait
	if ok {
		panic("chan should only ever be closed")
	}
	return current.result
}

func NewCastGroup() *CastGroup {
	c := &CastGroup{make(chan interface{}), make(chan *event)}
	go func() {
		var current *event
		for {
			select {
			case result := <-c.result:
				if current != nil {
					current.result = result
					close(current.wait)
				} else if result != nil {
					panic("first update must be nil")
				}
				current = &event{make(chan bool), nil}
			case c.ch <- current:
				// Wait() is being invoked; pass the current event.
			}
		}
	}()
	c.Done(nil)
	return c
}