func (c *partitionConsumer) catchup(fromOffset int64) {
	if c.catchingUp || fromOffset == c.topic.nextOffset {
		return
	}
	c.catchingUp = true
	go func() {
		for {
			c.topic.mutex.Lock()
			nextOffset := c.topic.nextOffset
			if fromOffset == nextOffset {
				c.catchingUp = false
				c.topic.mutex.Unlock()
				return
			}
			c.topic.mutex.Unlock()
			if nextOffset > fromOffset+int64(cap(c.messages)) {
				nextOffset = fromOffset + int64(cap(c.messages))
			}
			msgs, err := c.topic.db.FetchMessages(c.topic.topicName, fromOffset, nextOffset)
			if err != nil {
				log.Print("Error reading messages: ", err)
				time.Sleep(10 * time.Second)
				continue
			}
			if len(msgs) == 0 {
				log.Fatalf("Corrupt database returned no messages between %d and %d", fromOffset, nextOffset)
			}
			for i := range msgs {
				c.messages <- msgs[i].ConsumerMessage(c.topic.topicName)
			}
			fromOffset = msgs[len(msgs)-1].Offset + 1
		}
	}()
}
catchup makes the consumer go into "catchup" mode, where messages are read
from the database instead of directly from producers.
Once the consumer is up to date, i.e. no new messages in the database, then
the consumer will go back into normal mode where new messages are written
directly to the channel.
Must be called with the c.topic.mutex lock
package naffka
import (
	"fmt"
	"log"
	"sync"
	"time"
	sarama "github.com/Shopify/sarama"
	"github.com/matrix-org/naffka/storage"
	"github.com/matrix-org/naffka/types"
)
type Naffka struct {
	db          storage.Database
	topicsMutex sync.Mutex
	topics      map[string]*topic
}
func New(db storage.Database) (*Naffka, error) {
	n := &Naffka{db: db, topics: map[string]*topic{}}
	maxOffsets, err := db.MaxOffsets()
	if err != nil {
		return nil, err
	}
	for topicName, offset := range maxOffsets {
		n.topics[topicName] = &topic{
			db:         db,
			topicName:  topicName,
			nextOffset: offset + 1,
		}
	}
	return n, nil
}
func (n *Naffka) SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) {
	err = n.SendMessages([]*sarama.ProducerMessage{msg})
	return msg.Partition, msg.Offset, err
}
func (n *Naffka) SendMessages(msgs []*sarama.ProducerMessage) error {
	byTopic := map[string][]*sarama.ProducerMessage{}
	for _, msg := range msgs {
		byTopic[msg.Topic] = append(byTopic[msg.Topic], msg)
	}
	var topicNames []string
	for topicName := range byTopic {
		topicNames = append(topicNames, topicName)
	}
	now := time.Now()
	topics := n.getTopics(topicNames)
	for topicName := range byTopic {
		if err := topics[topicName].send(now, byTopic[topicName]); err != nil {
			return err
		}
	}
	return nil
}
func (n *Naffka) getTopics(topicNames []string) map[string]*topic {
	n.topicsMutex.Lock()
	defer n.topicsMutex.Unlock()
	result := map[string]*topic{}
	for _, topicName := range topicNames {
		t := n.topics[topicName]
		if t == nil {
			t = &topic{db: n.db, topicName: topicName}
			n.topics[topicName] = t
		}
		result[topicName] = t
	}
	return result
}
func (n *Naffka) Topics() ([]string, error) {
	n.topicsMutex.Lock()
	defer n.topicsMutex.Unlock()
	var result []string
	for topic := range n.topics {
		result = append(result, topic)
	}
	return result, nil
}
func (n *Naffka) Partitions(topic string) ([]int32, error) {
	return []int32{0}, nil
}
func (n *Naffka) ConsumePartition(topic string, partition int32, offset int64) (sarama.PartitionConsumer, error) {
	if partition != 0 {
		return nil, fmt.Errorf("Unknown partition ID %d", partition)
	}
	topics := n.getTopics([]string{topic})
	return topics[topic].consume(offset), nil
}
func (n *Naffka) HighWaterMarks() map[string]map[int32]int64 {
	n.topicsMutex.Lock()
	defer n.topicsMutex.Unlock()
	result := map[string]map[int32]int64{}
	for topicName, topic := range n.topics {
		result[topicName] = map[int32]int64{
			0: topic.highwaterMark(),
		}
	}
	return result
}
func (n *Naffka) Close() error {
	return nil
}
const channelSize = 1024
type partitionConsumer struct {
	topic    *topic
	messages chan *sarama.ConsumerMessage
	catchingUp bool
}
func (c *partitionConsumer) AsyncClose() {
}
func (c *partitionConsumer) Close() error {
	return nil
}
func (c *partitionConsumer) Messages() <-chan *sarama.ConsumerMessage {
	return c.messages
}
func (c *partitionConsumer) Errors() <-chan *sarama.ConsumerError {
	return nil
}
func (c *partitionConsumer) HighWaterMarkOffset() int64 {
	return c.topic.highwaterMark()
}