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
}
makeSortedIns given a partition->input map returns a slice of all the inputs
from the map sorted in ascending order of partition ids.
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
		}
	}
}