Rename watch.Mux -> watch.Broadcaster

A few reasons:
- Mux is already widely used in the codebase to refer to a http handler mux.
- Original meaning of Mux was something which sent a chose one of several inputs to
  and output.  This sends one output to all outputs.  Broadcast captures that idea
  better.
- Aligns with similar class config.Broadcaster (see #2747)

pkg/watch/mux.go

@@ -22,25 +22,25 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 )
 
-// Mux distributes event notifications among any number of watchers. Every event
+// Broadcaster distributes event notifications among any number of watchers. Every event
 // is delivered to every watcher.
-type Mux struct {
+type Broadcaster struct {
 	lock sync.Mutex
 
-	watchers    map[int64]*muxWatcher
+	watchers    map[int64]*broadcasterWatcher
 	nextWatcher int64
 
 	incoming chan Event
 }
 
-// NewMux creates a new Mux. queueLength is the maximum number of events to queue.
+// NewBroadcaster creates a new Broadcaster. queueLength is the maximum number of events to queue.
 // When queueLength is 0, Action will block until any prior event has been
 // completely distributed. It is guaranteed that events will be distibuted in the
 // order in which they ocurr, but the order in which a single event is distributed
 // among all of the watchers is unspecified.
-func NewMux(queueLength int) *Mux {
-	m := &Mux{
-		watchers: map[int64]*muxWatcher{},
+func NewBroadcaster(queueLength int) *Broadcaster {
+	m := &Broadcaster{
+		watchers: map[int64]*broadcasterWatcher{},
 		incoming: make(chan Event, queueLength),
 	}
 	go m.loop()
@@ -50,12 +50,12 @@ func NewMux(queueLength int) *Mux {
 // Watch adds a new watcher to the list and returns an Interface for it.
 // Note: new watchers will only receive new events. They won't get an entire history
 // of previous events.
-func (m *Mux) Watch() Interface {
+func (m *Broadcaster) Watch() Interface {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 	id := m.nextWatcher
 	m.nextWatcher++
-	w := &muxWatcher{
+	w := &broadcasterWatcher{
 		result:  make(chan Event),
 		stopped: make(chan struct{}),
 		id:      id,
@@ -66,7 +66,7 @@ func (m *Mux) Watch() Interface {
 }
 
 // stopWatching stops the given watcher and removes it from the list.
-func (m *Mux) stopWatching(id int64) {
+func (m *Broadcaster) stopWatching(id int64) {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 	w, ok := m.watchers[id]
@@ -79,7 +79,7 @@ func (m *Mux) stopWatching(id int64) {
 }
 
 // closeAll disconnects all watchers (presumably in response to a Shutdown call).
-func (m *Mux) closeAll() {
+func (m *Broadcaster) closeAll() {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 	for _, w := range m.watchers {
@@ -87,24 +87,24 @@ func (m *Mux) closeAll() {
 	}
 	// Delete everything from the map, since presence/absence in the map is used
 	// by stopWatching to avoid double-closing the channel.
-	m.watchers = map[int64]*muxWatcher{}
+	m.watchers = map[int64]*broadcasterWatcher{}
 }
 
 // Action distributes the given event among all watchers.
-func (m *Mux) Action(action EventType, obj runtime.Object) {
+func (m *Broadcaster) Action(action EventType, obj runtime.Object) {
 	m.incoming <- Event{action, obj}
 }
 
 // Shutdown disconnects all watchers (but any queued events will still be distributed).
 // You must not call Action after calling Shutdown.
-func (m *Mux) Shutdown() {
+func (m *Broadcaster) Shutdown() {
 	close(m.incoming)
 }
 
 // loop recieves from m.incoming and distributes to all watchers.
-func (m *Mux) loop() {
+func (m *Broadcaster) loop() {
 	// Deliberately not catching crashes here. Yes, bring down the process if there's a
-	// bug in watch.Mux.
+	// bug in watch.Broadcaster.
 	for {
 		event, ok := <-m.incoming
 		if !ok {
@@ -116,7 +116,7 @@ func (m *Mux) loop() {
 }
 
 // distribute sends event to all watchers. Blocking.
-func (m *Mux) distribute(event Event) {
+func (m *Broadcaster) distribute(event Event) {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 	for _, w := range m.watchers {
@@ -127,22 +127,22 @@ func (m *Mux) distribute(event Event) {
 	}
 }
 
-// muxWatcher handles a single watcher of a mux
-type muxWatcher struct {
+// broadcasterWatcher handles a single watcher of a broadcaster
+type broadcasterWatcher struct {
 	result  chan Event
 	stopped chan struct{}
 	stop    sync.Once
 	id      int64
-	m       *Mux
+	m       *Broadcaster
 }
 
 // ResultChan returns a channel to use for waiting on events.
-func (mw *muxWatcher) ResultChan() <-chan Event {
+func (mw *broadcasterWatcher) ResultChan() <-chan Event {
 	return mw.result
 }
 
 // Stop stops watching and removes mw from its list.
-func (mw *muxWatcher) Stop() {
+func (mw *broadcasterWatcher) Stop() {
 	mw.stop.Do(func() {
 		close(mw.stopped)
 		mw.m.stopWatching(mw.id)