Merge pull request #57 from q2wdqea/feat-mq-kpush

feat(mq): add key field
zeromicro · Jul 24, 2024 · 5af862f · 5af862f
2 parents 59d5072 + 2c75b16
commit 5af862f
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 1 deletion.
diff --git a/example/kq/producer/produce.go b/example/kq/producer/produce.go
@@ -41,10 +41,13 @@ func main() {
 			log.Fatal(err)
 		}
 
-		fmt.Println(string(body))
 		if err := pusher.Push(context.Background(), string(body)); err != nil {
 			log.Fatal(err)
 		}
+
+		if err := pusher.KPush(context.Background(), "test", string(body)); err != nil {
+			log.Fatal(err)
+		}
 	}
 
 	cmdline.EnterToContinue()

diff --git a/kq/pusher.go b/kq/pusher.go
@@ -97,6 +97,19 @@ func (p *Pusher) Name() string {
 	return p.topic
 }
 
+// KPush sends a message to the Kafka topic.
+func (p *Pusher) KPush(ctx context.Context, k, v string) error {
+	msg := kafka.Message{
+		Key:   []byte(k), // current timestamp
+		Value: []byte(v),
+	}
+	if p.executor != nil {
+		return p.executor.Add(msg, len(v))
+	} else {
+		return p.producer.WriteMessages(ctx, msg)
+	}
+}
+
 // Push sends a message to the Kafka topic.
 func (p *Pusher) Push(ctx context.Context, v string) error {
 	msg := kafka.Message{