func selectInput(prevSelectedIdx int, sortedIns []*input) int {
	maxLag := int64(-1)
	selectedIdx := -1
	for i, input := range sortedIns {
		if !input.msgOk {
			continue
		}
		lag := input.msg.HighWaterMark - input.msg.Offset
		if lag > maxLag {
			maxLag = lag
			selectedIdx = i
			continue
		}
		if lag < maxLag {
			continue
		}
		if selectedIdx > prevSelectedIdx {
			continue
		}
		if i > prevSelectedIdx {
			selectedIdx = i
		}
	}
	return selectedIdx
}
selectInput picks an input that should be multiplexed next. It prefers the
inputs with the largest lag. If there is more then one input with the same
largest lag, then it picks the one that has index following prevSelectedIdx.
package multiplexer
import (
	"reflect"
	"sort"
	"sync"
	"time"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/consumer"
	"github.com/mailgun/kafka-pixy/none"
)
type T struct {
	actDesc   *actor.Descriptor
	spawnInFn SpawnInFn
	inputs    map[int32]*input
	output    Out
	isRunning bool
	stopCh    chan none.T
	wg        sync.WaitGroup
	sortedInsMu sync.Mutex
	sortedIns   []*input
}
type In interface {
	Messages() <-chan consumer.Message
	IsSafe2Stop() bool
	Stop()
}
type Out interface {
	Messages() chan<- consumer.Message
}
type SpawnInFn func(partition int32) In
func New(parentActDesc *actor.Descriptor, spawnInFn SpawnInFn) *T {
	return &T{
		actDesc:   parentActDesc.NewChild("mux"),
		inputs:    make(map[int32]*input),
		spawnInFn: spawnInFn,
		stopCh:    make(chan none.T),
	}
}
type input struct {
	In
	partition int32
	msg       consumer.Message
	msgOk     bool
}
func (m *T) IsRunning() bool {
	return m.isRunning
}
func (m *T) IsSafe2Stop() bool {
	m.sortedInsMu.Lock()
	sortedIns := m.sortedIns
	m.sortedInsMu.Unlock()
	for _, in := range sortedIns {
		if !in.IsSafe2Stop() {
			return false
		}
	}
	return true
}
func (m *T) WireUp(output Out, assigned []int32) {
	var wg sync.WaitGroup
	wiringBegin := time.Now()
	m.stopIfRunning()
	m.output = output
	if output == nil {
		for p, in := range m.inputs {
			wg.Add(1)
			go func(in *input) {
				defer wg.Done()
				in.Stop()
			}(in)
			delete(m.inputs, p)
		}
		wg.Wait()
		return
	}
	for p, in := range m.inputs {
		if !hasPartition(p, assigned) {
			wg.Add(1)
			go func(in *input) {
				defer wg.Done()
				in.Stop()
			}(in)
			delete(m.inputs, p)
		}
	}
	for _, p := range assigned {
		if _, ok := m.inputs[p]; !ok {
			m.inputs[p] = &input{In: m.spawnInFn(p), partition: p}
		}
	}
	m.refreshSortedIns()
	if len(m.inputs) > 0 {
		m.start()
	}
	waitBegin := time.Now()
	wiringTook := waitBegin.Sub(wiringBegin)
	wg.Wait()
	waitedFor := time.Since(waitBegin)
	m.actDesc.Log().Infof("Wiring completed: took=%v, waited=%v", wiringTook, waitedFor)
}
func (m *T) Stop() {
	m.WireUp(nil, nil)
}
func (m *T) start() {
	actor.Spawn(m.actDesc, &m.wg, m.run)
	m.isRunning = true
}
func (m *T) stopIfRunning() {
	if m.isRunning {
		m.stopCh <- none.V
		m.wg.Wait()
		m.isRunning = false
	}
}
func (m *T) refreshSortedIns() {
	sortedIns := makeSortedIns(m.inputs)
	m.sortedInsMu.Lock()
	m.sortedIns = sortedIns
	m.sortedInsMu.Unlock()
}
func (m *T) run() {
reset:
	inputCount := len(m.inputs)
	if inputCount == 0 {
		return
	}
	selectCases := make([]reflect.SelectCase, inputCount+1)
	for i, in := range m.sortedIns {
		selectCases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(in.Messages())}
	}
	selectCases[inputCount] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(m.stopCh)}
	inputIdx := -1
	for {
		isAtLeastOneAvailable := false
		for _, in := range m.sortedIns {
			if in.msgOk {
				isAtLeastOneAvailable = true
				continue
			}
			select {
			case msg, ok := <-in.Messages():
				if !ok {
					m.actDesc.Log().Infof("input channel closed: partition=%d", in.partition)
					delete(m.inputs, in.partition)
					m.refreshSortedIns()
					goto reset
				}
				in.msg = msg
				in.msgOk = true
				isAtLeastOneAvailable = true
			default:
			}
		}
		if !isAtLeastOneAvailable {
			idx, value, _ := reflect.Select(selectCases)
			if idx == inputCount {
				return
			}
			m.sortedIns[idx].msg = value.Interface().(consumer.Message)
			m.sortedIns[idx].msgOk = true
		}
		inputIdx = selectInput(inputIdx, m.sortedIns)
		select {
		case <-m.stopCh:
			return
		case m.output.Messages() <- m.sortedIns[inputIdx].msg:
			m.sortedIns[inputIdx].msgOk = false
		}
	}
}
func makeSortedIns(inputs map[int32]*input) []*input {
	partitions := make([]int32, 0, len(inputs))
	for p := range inputs {
		partitions = append(partitions, p)
	}
	sort.Slice(partitions, func(i, j int) bool { return partitions[i] < partitions[j] })
	sortedIns := make([]*input, len(inputs))
	for i, p := range partitions {
		sortedIns[i] = inputs[p]
	}
	return sortedIns
}
func hasPartition(partition int32, partitions []int32) bool {
	count := len(partitions)
	if count == 0 {
		return false
	}
	return partitions[0] <= partition && partition <= partitions[count-1]
}