blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-12-13 16:47:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

v2 transfer

Browse Source
This commit is contained in:
pluto 2023-08-23 17:01:14 +08:00
parent da1ae7157b
commit 5507d8f012
1 changed files with 14 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
tools/data-conversion/kafka-conversation.go
View File

@ -18,6 +18,7 @@ import (
"context"
"fmt"
"github.com/OpenIMSDK/Open-IM-Server/pkg/rpcclient"
pbMsg "github.com/OpenIMSDK/protocol/msg"
"github.com/OpenIMSDK/protocol/sdkws"
openKeeper "github.com/OpenIMSDK/tools/discoveryregistry/zookeeper"
"github.com/OpenIMSDK/tools/errs"
@ -103,12 +104,22 @@ func GetMessage() {
defer wg.Done()
for msg := range pc.Messages() {
//Transfer([]*sarama.ConsumerMessage{msg})
msgFromMQ := &sdkws.MsgData{}
err := proto.Unmarshal(msg.Value, msgFromMQ)
//V2
msgFromMQV2 := pbMsg.MsgDataToMQ{}
err := proto.Unmarshal(msg.Value, &msgFromMQV2)
if err != nil {
fmt.Printf("err:%s \n", err)
}
fmt.Printf("msg:%s \n", msgFromMQ)
fmt.Printf("msg:%s \n", &msgFromMQV2)
//V3
//msgFromMQ := &sdkws.MsgData{}
//err = proto.Unmarshal(msg.Value, msgFromMQ)
//if err != nil {
// fmt.Printf("err:%s \n", err)
//}
//fmt.Printf("msg:%s \n", &msgFromMQ)
//fmt.Printf("Partition:%d, Offset:%d, Key:%s, Value:%s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
}