func (rs *retryingSink) Write(events ...Event) error {
	rs.mu.Lock()
	defer rs.mu.Unlock()
retry:
	if rs.closed {
		return ErrSinkClosed
	}
	if !rs.proceed() {
		logrus.Warnf("%v encountered too many errors, backing off", rs.sink)
		rs.wait(rs.failures.backoff)
		goto retry
	}
	if err := rs.write(events...); err != nil {
		if err == ErrSinkClosed {
			return err
		}
		logrus.Errorf("retryingsink: error writing events: %v, retrying", err)
		goto retry
	}
	return nil
}
Write attempts to flush the events to the downstream sink until it succeeds
or the sink is closed.
package notifications
import (
	"container/list"
	"fmt"
	"sync"
	"time"
	"github.com/Sirupsen/logrus"
)
type Broadcaster struct {
	sinks  []Sink
	events chan []Event
	closed chan chan struct{}
}
func NewBroadcaster(sinks ...Sink) *Broadcaster {
	b := Broadcaster{
		sinks:  sinks,
		events: make(chan []Event),
		closed: make(chan chan struct{}),
	}
	go b.run()
	return &b
}
func (b *Broadcaster) Write(events ...Event) error {
	select {
	case b.events <- events:
	case <-b.closed:
		return ErrSinkClosed
	}
	return nil
}
func (b *Broadcaster) Close() error {
	logrus.Infof("broadcaster: closing")
	select {
	case <-b.closed:
		return fmt.Errorf("broadcaster: already closed")
	default:
		closed := make(chan struct{})
		b.closed <- closed
		close(b.closed)
		<-closed
		return nil
	}
}
func (b *Broadcaster) run() {
	for {
		select {
		case block := <-b.events:
			for _, sink := range b.sinks {
				if err := sink.Write(block...); err != nil {
					logrus.Errorf("broadcaster: error writing events to %v, these events will be lost: %v", sink, err)
				}
			}
		case closing := <-b.closed:
			for _, sink := range b.sinks {
				if err := sink.Close(); err != nil {
					logrus.Errorf("broadcaster: error closing sink %v: %v", sink, err)
				}
			}
			closing <- struct{}{}
			logrus.Debugf("broadcaster: closed")
			return
		}
	}
}
type eventQueue struct {
	sink      Sink
	events    *list.List
	listeners []eventQueueListener
	cond      *sync.Cond
	mu        sync.Mutex
	closed    bool
}
type eventQueueListener interface {
	ingress(events ...Event)
	egress(events ...Event)
}
func newEventQueue(sink Sink, listeners ...eventQueueListener) *eventQueue {
	eq := eventQueue{
		sink:      sink,
		events:    list.New(),
		listeners: listeners,
	}
	eq.cond = sync.NewCond(&eq.mu)
	go eq.run()
	return &eq
}
func (eq *eventQueue) Write(events ...Event) error {
	eq.mu.Lock()
	defer eq.mu.Unlock()
	if eq.closed {
		return ErrSinkClosed
	}
	for _, listener := range eq.listeners {
		listener.ingress(events...)
	}
	eq.events.PushBack(events)
	eq.cond.Signal()  
	return nil
}
func (eq *eventQueue) Close() error {
	eq.mu.Lock()
	defer eq.mu.Unlock()
	if eq.closed {
		return fmt.Errorf("eventqueue: already closed")
	}
	eq.closed = true
	eq.cond.Signal()  
	eq.cond.Wait()    
	return eq.sink.Close()
}
func (eq *eventQueue) run() {
	for {
		block := eq.next()
		if block == nil {
			return  
		}
		if err := eq.sink.Write(block...); err != nil {
			logrus.Warnf("eventqueue: error writing events to %v, these events will be lost: %v", eq.sink, err)
		}
		for _, listener := range eq.listeners {
			listener.egress(block...)
		}
	}
}
func (eq *eventQueue) next() []Event {
	eq.mu.Lock()
	defer eq.mu.Unlock()
	for eq.events.Len() < 1 {
		if eq.closed {
			eq.cond.Broadcast()
			return nil
		}
		eq.cond.Wait()
	}
	front := eq.events.Front()
	block := front.Value.([]Event)
	eq.events.Remove(front)
	return block
}
type ignoredMediaTypesSink struct {
	Sink
	ignored map[string]bool
}
func newIgnoredMediaTypesSink(sink Sink, ignored []string) Sink {
	if len(ignored) == 0 {
		return sink
	}
	ignoredMap := make(map[string]bool)
	for _, mediaType := range ignored {
		ignoredMap[mediaType] = true
	}
	return &ignoredMediaTypesSink{
		Sink:    sink,
		ignored: ignoredMap,
	}
}
func (imts *ignoredMediaTypesSink) Write(events ...Event) error {
	var kept []Event
	for _, e := range events {
		if !imts.ignored[e.Target.MediaType] {
			kept = append(kept, e)
		}
	}
	if len(kept) == 0 {
		return nil
	}
	return imts.Sink.Write(kept...)
}
type retryingSink struct {
	mu     sync.Mutex
	sink   Sink
	closed bool
	failures struct {
		threshold int
		recent    int
		last      time.Time
		backoff   time.Duration  
	}
}
type retryingSinkListener interface {
	active(events ...Event)
	retry(events ...Event)
}
func newRetryingSink(sink Sink, threshold int, backoff time.Duration) *retryingSink {
	rs := &retryingSink{
		sink: sink,
	}
	rs.failures.threshold = threshold
	rs.failures.backoff = backoff
	return rs
}