go/go-micro
1
0
Fork 0
mirror of https://github.com/go-micro/go-micro.git synced 2025-09-16 08:36:30 +02:00
Code Issues Releases Activity

Add header suppor for Kafka broker plugin; (#2470)

Browse Source
This commit is contained in:
Wang
2022-04-04 11:54:19 -07:00
committed by GitHub
parent d1806e2883
commit 6dedee5d8c
1 changed files with 15 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
plugins/broker/kafka/options.go
View File

@@ -80,6 +80,21 @@ func (h *consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, cl
continue
}
if p.m.Body == nil {
p.m.Body = msg.Value
}
// if we don't have headers, create empty map
if m.Header == nil {
m.Header = make(map[string]string)
}
for _, header := range msg.Headers {
m.Header[string(header.Key)] = string(header.Value)
}
m.Header["Micro-Topic"] = msg.Topic // only for RPC server, it somehow inspect Header for topic
if _, ok := m.Header["Content-Type"]; !ok {
m.Header["Content-Type"] = "application/json" // default to json codec
}
err := h.handler(p)
if err == nil && h.subopts.AutoAck {
sess.MarkMessage(msg, "")