<commit_before>package kafka

import ( 
        "github.com/Shopify/sarama"
        "encoding/json" 
        "bytes"
        "log"
)

const (
  bufferSize     = 256
  initialOffset  = sarama.OffsetOldest // always start listening for the latest event. 
)

type Consumer struct {
  consumer           sarama.Consumer
  partitionConsumers []sarama.PartitionConsumer
  messages           chan *sarama.ConsumerMessage
  Registry           EventTypeRegistry
}

func NewConsumer(brokers []string, topic string) *Consumer {
  config := sarama.NewConfig()
  config.Consumer.Return.Errors = true

  consumer, err := sarama.NewConsumer(brokers, config) 
  if err != nil {
    log.Fatalln(err)
  }

  partitions, err := consumer.Partitions(topic) 
  if err != nil {
    log.Printf("Failed to get the list of partitions: %v", err)
  }

  log.Printf("%v partitions found for topic %v", len(partitions), topic)

  partitionConsumers := make([]sarama.PartitionConsumer, len(partitions))
  messages := make(chan *sarama.ConsumerMessage, bufferSize)

  for _, partition := range partitions {

    partitionConsumer, err := consumer.ConsumePartition(topic, partition, initialOffset)

    if err != nil {
      log.Fatalf("Failed to start consumer for partition %v: %v", partition, err)
    }

    go func(partitionConsumer sarama.PartitionConsumer) {
      for message := range partitionConsumer.Messages() {
        messages <- message
      }
    }(partitionConsumer)

  }

  return &Consumer{ 
    consumer           : consumer, 
    partitionConsumers : partitionConsumers,
    messages           : messages,
    Registry           : NewEventTypeRegistry(),
  }
}

// Consume messages and process them through the method pass in parameter
func (this *Consumer) Consume(processEvent func(interface{})) {
  go func() {
      log.Println("Start consuming messages ...")

      for message := range this.messages {
        log.Printf("Process message with offset %v", message.Offset)

        b := bytes.SplitAfterN(message.Value[:], []byte {44}, 1)

        eventType  := string(b[0][:])
        event, err := this.Registry.GetType(eventType)
        if err != nil {
          log.Println("Cannot find event type %v in the registry", eventType)
          continue
        }

        if err := json.Unmarshal(b[1], event) ; err != nil {
          log.Println("Cannot read event : ", err)
          continue
        }

        processEvent(event)
      }<commit_msg>Pass event type as a parameter of consumer<commit_after>package kafka

import ( 
        "github.com/Shopify/sarama"
        "encoding/json" 
        "reflect"
        "bytes"
        "log"
)

const (
  bufferSize     = 256
  initialOffset  = sarama.OffsetOldest // always start listening for the latest event. 
)

type Consumer struct {
  consumer           sarama.Consumer
  partitionConsumers []sarama.PartitionConsumer
  messages           chan *sarama.ConsumerMessage
}

func NewConsumer(brokers []string, topic string) *Consumer {
  config := sarama.NewConfig()
  config.Consumer.Return.Errors = true

  consumer, err := sarama.NewConsumer(brokers, config) 
  if err != nil {
    log.Fatalln(err)
  }

  partitions, err := consumer.Partitions(topic) 
  if err != nil {
    log.Printf("Failed to get the list of partitions: %v", err)
  }

  log.Printf("%v partitions found for topic %v", len(partitions), topic)

  partitionConsumers := make([]sarama.PartitionConsumer, len(partitions))
  messages := make(chan *sarama.ConsumerMessage, bufferSize)

  for _, partition := range partitions {

    partitionConsumer, err := consumer.ConsumePartition(topic, partition, initialOffset)

    if err != nil {
      log.Fatalf("Failed to start consumer for partition %v: %v", partition, err)
    }

    go func(partitionConsumer sarama.PartitionConsumer) {
      for message := range partitionConsumer.Messages() {
        messages <- message
      }
    }(partitionConsumer)

  }

  return &Consumer{ 
    consumer           : consumer, 
    partitionConsumers : partitionConsumers,
    messages           : messages,
  }
}

// Consume messages and process them through the method pass in parameter
func (this *Consumer) Consume(eventType reflect.Type, processEvent func(interface{})) {
  
  go func() {
      log.Println("Start consuming messages ...")

      for message := range this.messages {
        log.Printf("Process message with offset %v", message.Offset)

        b := bytes.SplitAfterN(message.Value[:], []byte {44}, 1)

        eventTypeFromMessage  := string(b[0][:])
        if eventType.Name() != eventTypeFromMessage {
          continue
        }

        event := reflect.New(eventType).Elem().Interface()
        if err := json.Unmarshal(b[1], event) ; err != nil {
          log.Println("Cannot read event : ", err)
          continue
        }

        processEvent(event)
      }
{'fraction_non_alphanumeric': 0.10975861357540746, 'fraction_numerical': 0.003301010934598721, 'mean_word_length': 2.755228505034857, 'pattern_counts': {'":': 0, '<': 5, '<?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': '703075', 'n_tokens_mistral': 1414, 'n_tokens_neox': 1326, 'n_words': 370}