IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-13 21:45:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

batch push

Browse Source
This commit is contained in:
skiffer-git
2022-06-09 13:12:46 +08:00
parent d1293dd047
commit 96f09c45f5
3 changed files with 11 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/common/kafka/producer.go
+7 -4
View File
@@ -33,19 +33,22 @@ func NewKafkaProducer(addr []string, topic string) *Producer {
return &p
}
func (p *Producer) SendMessage(m proto.Message, key string) (int32, int64, error) {
func (p *Producer) SendMessage(m proto.Message, key string, operationID string) (int32, int64, error) {
log2.Info(operationID, "SendMessage", "key ", key, m.String(), p.producer)
kMsg := &sarama.ProducerMessage{}
kMsg.Topic = p.topic
kMsg.Key = sarama.StringEncoder(key)
bMsg, err := proto.Marshal(m)
if err != nil {
log2.Error("", "", "proto marshal err = %s", err.Error())
log2.Error(operationID, "", "proto marshal err = %s", err.Error())
return -1, -1, err
}
if len(bMsg) == 0 {
return 0, 0, errors.New("msg content is nil")
}
kMsg.Value = sarama.ByteEncoder(bMsg)
return p.producer.SendMessage(kMsg)
log2.Info(operationID, "ByteEncoder SendMessage begin", "key ", kMsg, p.producer)
a, b, c := p.producer.SendMessage(kMsg)
log2.Info(operationID, "ByteEncoder SendMessage end", "key ", kMsg, p.producer)
return a, b, c
}