func (mf *msgFetcher) run() {
	defer close(mf.messagesCh)
	if mf.errorsCh != nil {
		defer close(mf.errorsCh)
	}
	mf.f.onMsgIStreamSpawned(mf)
	defer mf.f.onMsgIStreamStopped(mf)
	var (
		fetchResultCh       = make(chan fetchRs, 1)
		nilOrFetchResultsCh <-chan fetchRs
		nilOrMessagesCh     chan<- consumer.Message
		fetchedMessages     []consumer.Message
		err                 error
		currMessage         consumer.Message
		currMessageIdx      int
	)
	for {
		select {
		case bw := <-mf.assignmentCh:
			mf.actDesc.Log().Infof("Assigned executor: %s", bw)
			be := bw.(*brokerExecutor)
			mf.brokerRequestCh = be.requestsCh
			if nilOrFetchResultsCh == nil && nilOrMessagesCh == nil {
				mf.nilOrBrokerRequestsCh = mf.brokerRequestCh
			}
		case mf.nilOrBrokerRequestsCh <- fetchRq{mf.id.topic, mf.id.partition, mf.offset, fetchResultCh}:
			mf.nilOrBrokerRequestsCh = nil
			nilOrFetchResultsCh = fetchResultCh
		case fetchRs := <-nilOrFetchResultsCh:
			nilOrFetchResultsCh = nil
			if fetchedMessages, err = mf.parseFetchResponse(fetchRs); err != nil {
				mf.reportError(err)
				if err == sarama.ErrOffsetOutOfRange {
					mf.actDesc.Log().WithError(err).Error("Fatal request failure")
					return
				}
				mf.actDesc.Log().WithError(err).Error("Request failed")
				mf.brokerRequestCh = nil
				mf.f.mapper.TriggerReassign(mf)
				continue
			}
			if len(fetchedMessages) == 0 {
				mf.nilOrBrokerRequestsCh = mf.brokerRequestCh
				continue
			}
			currMessageIdx = 0
			currMessage = fetchedMessages[currMessageIdx]
			nilOrMessagesCh = mf.messagesCh
		case nilOrMessagesCh <- currMessage:
			mf.offset = currMessage.Offset + 1
			currMessageIdx++
			if currMessageIdx < len(fetchedMessages) {
				currMessage = fetchedMessages[currMessageIdx]
				continue
			}
			nilOrMessagesCh = nil
			mf.nilOrBrokerRequestsCh = mf.brokerRequestCh
		case <-mf.stopCh:
			return
		}
	}
}
pullMessages sends fetched requests to the broker executor assigned by the
redispatch goroutine; parses broker fetch responses and pushes parsed
`ConsumerMessages` to the message channel. It tries to keep the message
channel buffer full making fetch requests to the assigned broker as needed.
package msgfetcher
import (
	"sync"
	"time"
	"github.com/Shopify/sarama"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/config"
	"github.com/mailgun/kafka-pixy/consumer"
	"github.com/mailgun/kafka-pixy/mapper"
	"github.com/mailgun/kafka-pixy/none"
	"github.com/pkg/errors"
)
type Factory interface {
	Spawn(parentActDesc *actor.Descriptor, topic string, partition int32, offset int64) (T, int64, error)
	Stop()
}
type T interface {
	Messages() <-chan consumer.Message
	Stop()
}
var (
	testReportErrors bool
	errMessageTooLarge    = errors.New("message is larger than consumer.fetch_max_bytes")
	errIncompleteResponse = errors.New("response did not contain the expected topic/partition block")
)
type factory struct {
	actDesc  *actor.Descriptor
	cfg      *config.Proxy
	kafkaClt sarama.Client
	mapper   *mapper.T
	childrenMu sync.Mutex
	children   map[instanceID]*msgFetcher
}
type instanceID struct {
	topic     string
	partition int32
}
func SpawnFactory(parentActDesc *actor.Descriptor, cfg *config.Proxy, kafkaClt sarama.Client) Factory {
	f := &factory{
		actDesc:  parentActDesc.NewChild("msg_fetcher_f"),
		cfg:      cfg,
		kafkaClt: kafkaClt,
		children: make(map[instanceID]*msgFetcher),
	}
	f.mapper = mapper.Spawn(f.actDesc, cfg, f)
	return f
}
func (f *factory) Spawn(parentActDesc *actor.Descriptor, topic string, partition int32, offset int64) (T, int64, error) {
	realOffset, err := f.chooseStartingOffset(topic, partition, offset)
	if err != nil {
		return nil, sarama.OffsetNewest, err
	}
	f.childrenMu.Lock()
	defer f.childrenMu.Unlock()
	id := instanceID{topic, partition}
	if _, ok := f.children[id]; ok {
		return nil, sarama.OffsetNewest, sarama.ConfigurationError("That topic/partition is already being consumed")
	}
	actDesc := parentActDesc.NewChild("msg_fetcher")
	actDesc.AddLogField("kafka.topic", topic)
	actDesc.AddLogField("kafka.partition", partition)
	mf := &msgFetcher{
		actDesc:      actDesc,
		f:            f,
		id:           id,
		assignmentCh: make(chan mapper.Executor, 1),
		messagesCh:   make(chan consumer.Message, f.cfg.Consumer.ChannelBufferSize),
		stopCh:       make(chan none.T, 1),
		offset:       realOffset,
	}
	if testReportErrors {
		mf.errorsCh = make(chan error, f.cfg.Consumer.ChannelBufferSize)
	}
	f.children[id] = mf
	actor.Spawn(mf.actDesc, &mf.wg, mf.run)
	return mf, realOffset, nil
}
func (f *factory) Stop() {
	f.mapper.Stop()
}
func (f *factory) ResolveBroker(worker mapper.Worker) (*sarama.Broker, error) {
	ms := worker.(*msgFetcher)
	if err := f.kafkaClt.RefreshMetadata(ms.id.topic); err != nil {
		return nil, err
	}
	return f.kafkaClt.Leader(ms.id.topic, ms.id.partition)
}
func (f *factory) SpawnExecutor(brokerConn *sarama.Broker) mapper.Executor {
	be := &brokerExecutor{
		aggrActDesc:      f.actDesc.NewChild("broker", brokerConn.ID(), "aggr"),
		execActDesc:      f.actDesc.NewChild("broker", brokerConn.ID(), "exec"),
		cfg:              f.cfg,
		conn:             brokerConn,
		requestsCh:       make(chan fetchRq),
		requestBatchesCh: make(chan []fetchRq),
	}
	actor.Spawn(be.aggrActDesc, &be.wg, be.runAggregator)
	actor.Spawn(be.execActDesc, &be.wg, be.runExecutor)
	return be
}
func (f *factory) chooseStartingOffset(topic string, partition int32, offset int64) (int64, error) {
	newestOffset, err := f.kafkaClt.GetOffset(topic, partition, sarama.OffsetNewest)
	if err != nil {
		return 0, err
	}
	oldestOffset, err := f.kafkaClt.GetOffset(topic, partition, sarama.OffsetOldest)
	if err != nil {
		return 0, err
	}
	switch {
	case offset == sarama.OffsetNewest || offset > newestOffset:
		return newestOffset, nil
	case offset == sarama.OffsetOldest || offset < oldestOffset:
		return oldestOffset, nil
	default:
		return offset, nil
	}
}
func (f *factory) onMsgIStreamSpawned(mf *msgFetcher) {
	f.mapper.OnWorkerSpawned(mf)
}
func (f *factory) onMsgIStreamStopped(mf *msgFetcher) {
	f.childrenMu.Lock()
	delete(f.children, mf.id)
	f.childrenMu.Unlock()
	f.mapper.OnWorkerStopped(mf)
}
type msgFetcher struct {
	actDesc               *actor.Descriptor
	f                     *factory
	id                    instanceID
	offset                int64
	assignmentCh          chan mapper.Executor
	messagesCh            chan consumer.Message
	errorsCh              chan error
	brokerRequestCh       chan<- fetchRq
	nilOrBrokerRequestsCh chan<- fetchRq
	stopCh                chan none.T
	wg                    sync.WaitGroup
}
func (mf *msgFetcher) Messages() <-chan consumer.Message {
	return mf.messagesCh
}
func (mf *msgFetcher) Stop() {
	close(mf.stopCh)
	mf.wg.Wait()
}
func (mf *msgFetcher) Assignment() chan<- mapper.Executor {
	return mf.assignmentCh
}