func (p Producer) Project(msg *confluent.Message) {
	p.collectBefore(msg)
	p.producer.ProduceChannel() <- msg
}
Project implements the Projector interface
package kafka
import (
	"log"
	"time"
	confluent "github.com/confluentinc/confluent-kafka-go/kafka"
	_instrument "github.com/etf1/kafka-transformer/internal/instrument"
	"github.com/etf1/kafka-transformer/pkg/instrument"
	"github.com/etf1/kafka-transformer/pkg/logger"
)
type Producer struct {
	producer  *confluent.Producer
	log       logger.Log
	collector instrument.Collector
}
func NewProducer(l logger.Log, config *confluent.ConfigMap, collector instrument.Collector) (Producer, error) {
	kafkaProducer, err := confluent.NewProducer(config)
	if err != nil {
		return Producer{}, err
	}
	p := Producer{
		producer:  kafkaProducer,
		log:       l,
		collector: collector,
	}
	go func() {
		defer log.Println("kafka producer stopped")
		for e := range kafkaProducer.Events() {
			switch ev := e.(type) {
			case *confluent.Message:
				msg := ev
				if msg.TopicPartition.Error != nil {
					p.collectAfter(msg, msg.TopicPartition.Error)
					l.Errorf("delivery failed: %v", msg.TopicPartition.Error)
				} else {
					p.collectAfter(msg, nil)
					l.Debugf("producer: delivered message to topic %s [%d] at offset %v",
						*msg.TopicPartition.Topic, msg.TopicPartition.Partition, msg.TopicPartition.Offset)
				}
			case confluent.Error:
				p.collectAfter(nil, ev)
				l.Errorf("producer: received an error with code %v: %v", ev.Code(), ev)
			default:
				l.Debugf("producer: ignoring event: %s", ev)
			}
		}
	}()
	return p, nil
}