feat: mapper should return more than one message #29

Trendyol · Mar 24, 2023 · 078f15c · 078f15c
1 parent bdaee76
commit 078f15c
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 11 deletions.
diff --git a/connector.go b/connector.go
@@ -54,14 +54,16 @@ func (c *connector) listener(ctx *models.ListenerContext) {
 		return
 	}
 
-	if kafkaMessage := c.mapper(e); kafkaMessage != nil {
-		defer message.MessagePool.Put(kafkaMessage)
-		topic := c.config.Kafka.CollectionTopicMapping[e.CollectionName]
-		if topic == "" {
-			c.errorLogger.Printf("unexpected collection | %s", e.CollectionName)
-			return
+	for _, kafkaMessage := range c.mapper(e) {
+		if kafkaMessage != nil {
+			topic := c.config.Kafka.CollectionTopicMapping[e.CollectionName]
+			if topic == "" {
+				c.errorLogger.Printf("unexpected collection | %s", e.CollectionName)
+				return
+			}
+			c.producer.Produce(ctx, kafkaMessage.Value, kafkaMessage.Key, kafkaMessage.Headers, topic)
+			message.MessagePool.Put(kafkaMessage)
 		}
-		c.producer.Produce(ctx, kafkaMessage.Value, kafkaMessage.Key, kafkaMessage.Headers, topic)
 	}
 }
 

diff --git a/example/main.go b/example/main.go
@@ -6,9 +6,9 @@ import (
 	"github.com/Trendyol/go-kafka-connect-couchbase/kafka/message"
 )
 
-func mapper(event couchbase.Event) *message.KafkaMessage {
-	// return nil if you wish filter the event
-	return message.GetKafkaMessage(event.Key, event.Value, nil)
+func mapper(event couchbase.Event) []*message.KafkaMessage {
+	// return empty if you wish filter the event
+	return []*message.KafkaMessage{message.GetKafkaMessage(event.Key, event.Value, nil)}
 }
 
 func main() {

diff --git a/mapper.go b/mapper.go
@@ -5,4 +5,4 @@ import (
 	"github.com/Trendyol/go-kafka-connect-couchbase/kafka/message"
 )
 
-type Mapper func(event couchbase.Event) *message.KafkaMessage
+type Mapper func(event couchbase.Event) []*message.KafkaMessage