Initial

Signed-off-by: Alex Ellis (OpenFaaS Ltd) <alexellis2@gmail.com>

vendor/github.com/docker/go-events/broadcast.go

@@ -0,0 +1,178 @@
+package events
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/sirupsen/logrus"
+)
+
+// Broadcaster sends events to multiple, reliable Sinks. The goal of this
+// component is to dispatch events to configured endpoints. Reliability can be
+// provided by wrapping incoming sinks.
+type Broadcaster struct {
+	sinks   []Sink
+	events  chan Event
+	adds    chan configureRequest
+	removes chan configureRequest
+
+	shutdown chan struct{}
+	closed   chan struct{}
+	once     sync.Once
+}
+
+// NewBroadcaster appends one or more sinks to the list of sinks. The
+// broadcaster behavior will be affected by the properties of the sink.
+// Generally, the sink should accept all messages and deal with reliability on
+// its own. Use of EventQueue and RetryingSink should be used here.
+func NewBroadcaster(sinks ...Sink) *Broadcaster {
+	b := Broadcaster{
+		sinks:    sinks,
+		events:   make(chan Event),
+		adds:     make(chan configureRequest),
+		removes:  make(chan configureRequest),
+		shutdown: make(chan struct{}),
+		closed:   make(chan struct{}),
+	}
+
+	// Start the broadcaster
+	go b.run()
+
+	return &b
+}
+
+// Write accepts an event to be dispatched to all sinks. This method will never
+// fail and should never block (hopefully!). The caller cedes the memory to the
+// broadcaster and should not modify it after calling write.
+func (b *Broadcaster) Write(event Event) error {
+	select {
+	case b.events <- event:
+	case <-b.closed:
+		return ErrSinkClosed
+	}
+	return nil
+}
+
+// Add the sink to the broadcaster.
+//
+// The provided sink must be comparable with equality. Typically, this just
+// works with a regular pointer type.
+func (b *Broadcaster) Add(sink Sink) error {
+	return b.configure(b.adds, sink)
+}
+
+// Remove the provided sink.
+func (b *Broadcaster) Remove(sink Sink) error {
+	return b.configure(b.removes, sink)
+}
+
+type configureRequest struct {
+	sink     Sink
+	response chan error
+}
+
+func (b *Broadcaster) configure(ch chan configureRequest, sink Sink) error {
+	response := make(chan error, 1)
+
+	for {
+		select {
+		case ch <- configureRequest{
+			sink:     sink,
+			response: response}:
+			ch = nil
+		case err := <-response:
+			return err
+		case <-b.closed:
+			return ErrSinkClosed
+		}
+	}
+}
+
+// Close the broadcaster, ensuring that all messages are flushed to the
+// underlying sink before returning.
+func (b *Broadcaster) Close() error {
+	b.once.Do(func() {
+		close(b.shutdown)
+	})
+
+	<-b.closed
+	return nil
+}
+
+// run is the main broadcast loop, started when the broadcaster is created.
+// Under normal conditions, it waits for events on the event channel. After
+// Close is called, this goroutine will exit.
+func (b *Broadcaster) run() {
+	defer close(b.closed)
+	remove := func(target Sink) {
+		for i, sink := range b.sinks {
+			if sink == target {
+				b.sinks = append(b.sinks[:i], b.sinks[i+1:]...)
+				break
+			}
+		}
+	}
+
+	for {
+		select {
+		case event := <-b.events:
+			for _, sink := range b.sinks {
+				if err := sink.Write(event); err != nil {
+					if err == ErrSinkClosed {
+						// remove closed sinks
+						remove(sink)
+						continue
+					}
+					logrus.WithField("event", event).WithField("events.sink", sink).WithError(err).
+						Errorf("broadcaster: dropping event")
+				}
+			}
+		case request := <-b.adds:
+			// while we have to iterate for add/remove, common iteration for
+			// send is faster against slice.
+
+			var found bool
+			for _, sink := range b.sinks {
+				if request.sink == sink {
+					found = true
+					break
+				}
+			}
+
+			if !found {
+				b.sinks = append(b.sinks, request.sink)
+			}
+			// b.sinks[request.sink] = struct{}{}
+			request.response <- nil
+		case request := <-b.removes:
+			remove(request.sink)
+			request.response <- nil
+		case <-b.shutdown:
+			// close all the underlying sinks
+			for _, sink := range b.sinks {
+				if err := sink.Close(); err != nil && err != ErrSinkClosed {
+					logrus.WithField("events.sink", sink).WithError(err).
+						Errorf("broadcaster: closing sink failed")
+				}
+			}
+			return
+		}
+	}
+}
+
+func (b *Broadcaster) String() string {
+	// Serialize copy of this broadcaster without the sync.Once, to avoid
+	// a data race.
+
+	b2 := map[string]interface{}{
+		"sinks":   b.sinks,
+		"events":  b.events,
+		"adds":    b.adds,
+		"removes": b.removes,
+
+		"shutdown": b.shutdown,
+		"closed":   b.closed,
+	}
+
+	return fmt.Sprint(b2)
+}