<commit_before>package kafka

import (
	"github.com/Shopify/sarama"
	"github.com/gotgo/fw/logging"
)

type Consumer struct {
	Name            string
	StartingOffsets map[int]int64
	Log             logging.Logger `inject:""`
	Topic           string
	stopper         chan struct{}
	events          chan *ConsumerEvent
	client          *sarama.Client
}

func NewConsumer(client *sarama.Client, name, topic string, startingOffsets map[int]int64) *Consumer {
	consumer := &Consumer{
		Name:            name,
		StartingOffsets: startingOffsets,
		Topic:           topic,
		events:          make(chan *ConsumerEvent),
		client:          client,
	}
	return consumer
}

// Run - Nonblocking
func (c *Consumer) Run() {
	consumers := c.setupConsumers()
	c.consume(consumers)
}

func (c *Consumer) Events() <-chan *ConsumerEvent {
	return c.events
}

func (c *Consumer) Shutdown() {
	if c.stopper != nil {
		close(c.stopper)
	}
}

func (c *Consumer) setupConsumers() []*sarama.Consumer {
	offsets := c.StartingOffsets
	count := len(offsets)
	consumers := make([]*sarama.Consumer, count)
	i := 0
	for partition, offset := range offsets {
		if consumer, err := c.getConsumer(c.client, c.Topic, int32(partition), c.Name, offset); err != nil {
			panic(err)
		} else {
			consumers[i] = consumer
		}
		i++
	}
	return consumers
}

func (c *Consumer) consume(consumers []*sarama.Consumer) {
	c.stopper = make(chan struct{})
	for _, consumer := range consumers {
		defer consumer.Close()
	}

	for _, consumer := range consumers {
		go func() {
			select {
			case value := <-consumer.Events():
				evt := &ConsumerEvent{
					Error:     value.Err,
					Message:   value.Value,<commit_msg>cleanup<commit_after>package kafka

import (
	"github.com/Shopify/sarama"
	"github.com/gotgo/fw/logging"
)

type Consumer struct {
	ClientName      string
	StartingOffsets map[int]int64
	Log             logging.Logger `inject:""`
	Topic           string
	stopper         chan struct{}
	events          chan *ConsumerEvent
	client          *sarama.Client
}

func NewConsumer(client *sarama.Client, clientName, topic string, startingOffsets map[int]int64) *Consumer {
	consumer := &Consumer{
		ClientName:      clientName,
		StartingOffsets: startingOffsets,
		Topic:           topic,
		events:          make(chan *ConsumerEvent),
		client:          client,
	}
	return consumer
}

// Run - Nonblocking
func (c *Consumer) Run() {
	consumers := c.setupConsumers()
	c.consume(consumers)
}

func (c *Consumer) Events() <-chan *ConsumerEvent {
	return c.events
}

func (c *Consumer) Shutdown() {
	if c.stopper != nil {
		close(c.stopper)
	}
}

func (c *Consumer) setupConsumers() []*sarama.Consumer {
	offsets := c.StartingOffsets
	count := len(offsets)
	consumers := make([]*sarama.Consumer, count)
	i := 0
	for partition, offset := range offsets {
		if consumer, err := c.getConsumer(c.client, c.Topic, int32(partition), c.ClientName, offset); err != nil {
			panic(err)
		} else {
			consumers[i] = consumer
		}
		i++
	}
	return consumers
}

func (c *Consumer) consume(consumers []*sarama.Consumer) {
	c.stopper = make(chan struct{})
	for _, consumer := range consumers {
		defer consumer.Close()
	}

	for _, consumer := range consumers {
		go func() {
			select {
			case value := <-consumer.Events():
				evt := &ConsumerEvent{
					Error:     value.Err,
					Message:   value.Value,
{'fraction_non_alphanumeric': 0.13014925373134328, 'fraction_numerical': 0.00417910447761194, 'mean_word_length': 3.121771217712177, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3779875', 'n_tokens_mistral': 1204, 'n_tokens_neox': 1095, 'n_words': 279}