func (m *T) run() {
	var (
		changes             = make(map[Worker]eventType)
		reassignDoneCh      = make(chan none.T, 1)
		lastChangeTimes     = make(map[Worker]clock.Time)
		deferredChanges     = make(map[Worker]clock.Time)
		nilOrDeferredCh     <-chan clock.Time
		nilOrReassignDoneCh <-chan none.T
		stop                = false
		nilOrStopCh         = m.stopCh
	)
	for {
		select {
		case ev := <-m.eventsCh:
			now := clock.Now().UTC()
			switch ev.t {
			case evWorkerSpawned:
				changes[ev.w] = ev.t
				lastChangeTimes[ev.w] = now
			case evReassignNeeded:
				if now.Sub(lastChangeTimes[ev.w]) < m.cfg.Consumer.RetryBackoff {
					if _, ok := deferredChanges[ev.w]; !ok {
						deferredChanges[ev.w] = now.Add(m.cfg.Consumer.RetryBackoff)
						if nilOrDeferredCh == nil {
							nilOrDeferredCh = clock.After(m.cfg.Consumer.RetryBackoff)
						}
					}
					break
				}
				changes[ev.w] = ev.t
				lastChangeTimes[ev.w] = now
				delete(deferredChanges, ev.w)
			case evWorkerStopped:
				changes[ev.w] = ev.t
				delete(lastChangeTimes, ev.w)
				delete(deferredChanges, ev.w)
			}
		case <-nilOrReassignDoneCh:
			nilOrReassignDoneCh = nil
		case <-nilOrDeferredCh:
			nilOrDeferredCh = nil
			now := clock.Now().UTC()
			nextRetryAt := now.Add(m.cfg.Consumer.RetryBackoff)
			for w, deadline := range deferredChanges {
				if deadline.After(now) {
					if deadline.Before(nextRetryAt) {
						nextRetryAt = deadline
					}
					continue
				}
				changes[w] = evReassignNeeded
				lastChangeTimes[w] = now
				delete(deferredChanges, w)
			}
			if len(deferredChanges) >= 0 {
				nilOrDeferredCh = clock.After(nextRetryAt.Sub(now))
			}
		case <-nilOrStopCh:
			nilOrStopCh = nil
			stop = true
		}
		if testMode {
			skip := atomic.AddInt32(&testSkipEvents, -1)
			if skip >= 0 {
				continue
			}
		}
		if len(changes) > 0 && nilOrReassignDoneCh == nil {
			m.actDesc.Log().Infof("Reassign initiated: changes=%s", prettyfmt.Val(changes))
			reassignActDesc := m.actDesc.NewChild("reassign")
			frozenChanges := changes
			actor.Spawn(reassignActDesc, nil, func() {
				m.reassign(reassignActDesc, frozenChanges, reassignDoneCh)
			})
			changes = make(map[Worker]eventType)
			nilOrReassignDoneCh = reassignDoneCh
		}
		if stop && nilOrReassignDoneCh == nil && len(m.assignments) == 0 {
			return
		}
	}
}
run listens for mapping events, batches them into a mappingChange object and
triggers reassignments, making sure to run only one at a time. When signaled
to stop it only quits when all children have closed.
package mapper
import (
	"fmt"
	"sync"
	"sync/atomic"
	"github.com/Shopify/sarama"
	"github.com/mailgun/holster/v4/clock"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/config"
	"github.com/mailgun/kafka-pixy/none"
	"github.com/mailgun/kafka-pixy/prettyfmt"
)
type T struct {
	actDesc     *actor.Descriptor
	cfg         *config.Proxy
	resolver    Resolver
	eventsCh    chan event
	assignments map[Worker]Executor
	references  map[Executor]int
	connections map[*sarama.Broker]Executor
	stopCh      chan none.T
	wg          sync.WaitGroup
}
type Resolver interface {
	ResolveBroker(worker Worker) (*sarama.Broker, error)
	SpawnExecutor(brokerConn *sarama.Broker) Executor
}
type Worker interface {
	Assignment() chan<- Executor
}
type Executor interface {
	BrokerConn() *sarama.Broker
	Stop()
}
type eventType int
type event struct {
	t eventType
	w Worker
}
const (
	evWorkerSpawned eventType = iota
	evWorkerStopped
	evReassignNeeded
)
var (
	eventNames = []string{"spawned", "stopped", "outdated"}
	testMode       bool
	testSkipEvents int32
)
func (et eventType) String() string {
	return eventNames[et]
}
func Spawn(parentActDesc *actor.Descriptor, cfg *config.Proxy, resolver Resolver) *T {
	m := &T{
		actDesc:     parentActDesc.NewChild("mapper"),
		cfg:         cfg,
		resolver:    resolver,
		eventsCh:    make(chan event, cfg.Consumer.ChannelBufferSize),
		assignments: make(map[Worker]Executor),
		references:  make(map[Executor]int),
		connections: make(map[*sarama.Broker]Executor),
		stopCh:      make(chan none.T),
	}
	actor.Spawn(m.actDesc, &m.wg, m.run)
	return m
}
func (m *T) OnWorkerSpawned(w Worker) {
	m.eventsCh <- event{evWorkerSpawned, w}
}
func (m *T) OnWorkerStopped(w Worker) {
	m.eventsCh <- event{evWorkerStopped, w}
}
func (m *T) TriggerReassign(w Worker) {
	m.eventsCh <- event{evReassignNeeded, w}
}
func (m *T) Stop() {
	close(m.stopCh)
	m.wg.Wait()
}