func Monitor(monitorCh chan<- *EndpointCrashReport) {
	DefaultBroker.Monitor(monitorCh)
}
Monitor calls DefaultBroker.Monitor
package broker
import (
	"fmt"
	"runtime"
	"sync"
	"time"
	"github.com/meeko/meekod/broker/log"
)
const Version = "0.0.1"
const (
	ErrorWindowSize             = time.Second
	NumAllowedConsecutiveErrors = 3
)
type EndpointFactory func() (Endpoint, error)
type EndpointCrashReport struct {
	FactoryId string
	Dropped   bool
	Error     error
}
type Broker struct {
	factories map[string]EndpointFactory
	monitorCh chan<- *EndpointCrashReport
	state     brokerState
	termCh    chan struct{}
	termAckCh chan struct{}
	mu        *sync.Mutex
}
type brokerState int
const (
	stateInitialised brokerState = iota
	stateServing
	stateTerminated
)
var stateStrings = [...]string{
	"INITIALISED",
	"SERVING",
	"TERMINATED",
}
func (s brokerState) String() string {
	return stateStrings[int(s)]
}
func New() *Broker {
	return &Broker{
		factories: make(map[string]EndpointFactory),
		termCh:    make(chan struct{}),
		termAckCh: make(chan struct{}),
		mu:        new(sync.Mutex),
	}
}
func (b *Broker) RegisterEndpointFactory(identifier string, factory EndpointFactory) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.ensureState(stateInitialised)
	log.Infof("[broker] Registered endpoint factory %v", identifier)
	b.factories[identifier] = factory
}
func (b *Broker) Monitor(monitorCh chan<- *EndpointCrashReport) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.ensureState(stateInitialised)
	log.Info("[broker] Monitoring enabled")
	b.monitorCh = monitorCh
}
func (b *Broker) ListenAndServe() {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.ensureState(stateInitialised)
	var wg sync.WaitGroup
	wg.Add(len(b.factories))
	go func() {
		wg.Wait()
		if b.monitorCh != nil {
			close(b.monitorCh)
		}
		close(b.termAckCh)
	}()
	for identifier, factoryMethod := range b.factories {
		go func(id string, factory EndpointFactory) {
			defer func() {
				if r := recover(); r != nil {
					st := make([]byte, 4096)
					n := runtime.Stack(st, false)
					log.Errorf("[broker] Endpoint %v panicked: %v\n%v", id, r, string(st[:n]))
				}
				log.Infof("[broker] Endpoint %v terminated", id)
				wg.Done()
			}()
			var (
				ep         Endpoint
				errCounter uint
				err        error
			)
			for {
				if errCounter == NumAllowedConsecutiveErrors {
					log.Warnf("[broker] Endpoint %v reached the error threshold, dropping ...", id)
					if b.monitorCh != nil {
						select {
						case b.monitorCh <- &EndpointCrashReport{
							FactoryId: id,
							Dropped:   true,
						}:
						case <-b.termCh:
						}
					}
					return
				}
				log.Infof("[broker] Instantiating endpoint %v", id)
				ep, err = factory()
				if err != nil {
					log.Errorf("[broker] Failed to instantiate %v: %v", id, err)
					errCounter++
					if b.monitorCh != nil {
						select {
						case b.monitorCh <- &EndpointCrashReport{
							FactoryId: id,
							Error:     err,
						}:
						case <-b.termCh:
							return
						}
					}
					continue
				}
				watcherReturnCh := make(chan struct{})
				go func() {
					select {
					case <-b.termCh:
						log.Infof("[broker] Terminating endpoint %v", id)
						if err := ep.Close(); err != nil {
							log.Errorf("[broker] Failed to close endpoint %v: %v", id, err)
						}
					case <-watcherReturnCh:
						return
					}
				}()
				listenTimestamp := time.Now()
				log.Debugf("[broker] Endpoint %v entering ListenAndServe", id)
				err = ep.ListenAndServe()
				log.Debugf("[broker] Endpoint %v leaving ListenAndServe with error=%v", id, err)
				close(watcherReturnCh)
				if err == nil {
					return
				}
				if err != nil {
					if err := ep.Close(); err != nil {
						log.Debugf("[broker] Failed to close endpoint %v: %v", id, err)
					}
					log.Debugf("[broker] Incrementing error counter for %v", id)
					errCounter++
				}
				if time.Now().Sub(listenTimestamp) > ErrorWindowSize {
					if errCounter > 0 {
						log.Debugf("[broker] Decrementing error counter for %v", id)
						errCounter--
					}
				}
			}
		}(identifier, factoryMethod)
	}
	b.state = stateServing
}
func (b *Broker) Terminate() {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.ensureState(stateServing)
	log.Info("[broker] Terminating...")
	close(b.termCh)
	b.state = stateTerminated
	b.Wait()
}
func (b *Broker) Terminated() <-chan struct{} {
	return b.termAckCh
}
func (b *Broker) Wait() {
	<-b.Terminated()
}
func (b *Broker) ensureState(state brokerState) {
	if b.state != state {
		panic(&ErrInvalidState{state, b.state})
	}
}
type ErrInvalidState struct {
	expected brokerState
	current  brokerState
}
func (err ErrInvalidState) Error() string {
	return fmt.Sprintf("Invalid broker state: expected %v, got %v", err.expected, err.current)
}
var DefaultBroker = New()
func RegisterEndpointFactory(factoryId string, factory EndpointFactory) {
	DefaultBroker.RegisterEndpointFactory(factoryId, factory)
}