mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-05-10 12:05:58 +08:00
add cmd/open_im_push
This commit is contained in:
xmcy0011
@@ -0,0 +1,36 @@
|
||||
/*
|
||||
** description("").
|
||||
** copyright('open-im,www.open-im.io').
|
||||
** author("fg,Gordon@open-im.io").
|
||||
** time(2021/3/22 15:33).
|
||||
*/
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/src/common/config"
|
||||
"Open_IM/src/common/kafka"
|
||||
"Open_IM/src/common/log"
|
||||
"Open_IM/src/utils"
|
||||
)
|
||||
|
||||
var (
|
||||
rpcServer RPCServer
|
||||
pushCh PushConsumerHandler
|
||||
pushTerminal []int32
|
||||
producer *kafka.Producer
|
||||
)
|
||||
|
||||
func Init(rpcPort int) {
|
||||
log.NewPrivateLog(config.Config.ModuleName.PushName)
|
||||
rpcServer.Init(rpcPort)
|
||||
pushCh.Init()
|
||||
pushTerminal = []int32{utils.IOSPlatformID}
|
||||
}
|
||||
func init() {
|
||||
producer = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.Ws2mschat.Topic)
|
||||
}
|
||||
|
||||
func Run() {
|
||||
go rpcServer.run()
|
||||
go pushCh.pushConsumerGroup.RegisterHandleAndConsumer(&pushCh)
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
/*
|
||||
** description("").
|
||||
** copyright('Open_IM,www.Open_IM.io').
|
||||
** author("fg,Gordon@tuoyun.net").
|
||||
** time(2021/5/13 10:33).
|
||||
*/
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/src/common/config"
|
||||
kfk "Open_IM/src/common/kafka"
|
||||
"Open_IM/src/common/log"
|
||||
pbChat "Open_IM/src/proto/chat"
|
||||
pbRelay "Open_IM/src/proto/relay"
|
||||
"github.com/Shopify/sarama"
|
||||
"github.com/golang/protobuf/proto"
|
||||
)
|
||||
|
||||
type fcb func(msg []byte)
|
||||
|
||||
type PushConsumerHandler struct {
|
||||
msgHandle map[string]fcb
|
||||
pushConsumerGroup *kfk.MConsumerGroup
|
||||
}
|
||||
|
||||
func (ms *PushConsumerHandler) Init() {
|
||||
ms.msgHandle = make(map[string]fcb)
|
||||
ms.msgHandle[config.Config.Kafka.Ms2pschat.Topic] = ms.handleMs2PsChat
|
||||
ms.pushConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V0_10_2_0,
|
||||
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ms2pschat.Topic}, config.Config.Kafka.Ms2pschat.Addr,
|
||||
config.Config.Kafka.ConsumerGroupID.MsgToPush)
|
||||
}
|
||||
func (ms *PushConsumerHandler) handleMs2PsChat(msg []byte) {
|
||||
log.InfoByKv("msg come from kafka And push!!!", "", "msg", string(msg))
|
||||
pbData := pbChat.MsgSvrToPushSvrChatMsg{}
|
||||
if err := proto.Unmarshal(msg, &pbData); err != nil {
|
||||
log.ErrorByKv("push Unmarshal msg err", "", "msg", string(msg), "err", err.Error())
|
||||
return
|
||||
}
|
||||
sendPbData := pbRelay.MsgToUserReq{}
|
||||
sendPbData.SendTime = pbData.SendTime
|
||||
sendPbData.OperationID = pbData.OperationID
|
||||
sendPbData.ServerMsgID = pbData.MsgID
|
||||
sendPbData.MsgFrom = pbData.MsgFrom
|
||||
sendPbData.ContentType = pbData.ContentType
|
||||
sendPbData.SessionType = pbData.SessionType
|
||||
sendPbData.RecvID = pbData.RecvID
|
||||
sendPbData.Content = pbData.Content
|
||||
sendPbData.SendID = pbData.SendID
|
||||
sendPbData.SenderNickName = pbData.SenderNickName
|
||||
sendPbData.SenderFaceURL = pbData.SenderFaceURL
|
||||
sendPbData.ClientMsgID = pbData.ClientMsgID
|
||||
sendPbData.PlatformID = pbData.PlatformID
|
||||
sendPbData.RecvSeq = pbData.RecvSeq
|
||||
//Call push module to send message to the user
|
||||
MsgToUser(&sendPbData, pbData.OfflineInfo, pbData.Options)
|
||||
}
|
||||
func (PushConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (PushConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (ms *PushConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
claim sarama.ConsumerGroupClaim) error {
|
||||
for msg := range claim.Messages() {
|
||||
log.InfoByKv("kafka get info to mysql", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "msg", string(msg.Value))
|
||||
ms.msgHandle[msg.Topic](msg.Value)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/src/common/config"
|
||||
"Open_IM/src/common/log"
|
||||
"Open_IM/src/grpc-etcdv3/getcdv3"
|
||||
"Open_IM/src/proto/push"
|
||||
pbRelay "Open_IM/src/proto/relay"
|
||||
"Open_IM/src/utils"
|
||||
"context"
|
||||
"google.golang.org/grpc"
|
||||
"net"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type RPCServer struct {
|
||||
rpcPort int
|
||||
rpcRegisterName string
|
||||
etcdSchema string
|
||||
etcdAddr []string
|
||||
}
|
||||
|
||||
func (r *RPCServer) Init(rpcPort int) {
|
||||
r.rpcPort = rpcPort
|
||||
r.rpcRegisterName = config.Config.RpcRegisterName.OpenImPushName
|
||||
r.etcdSchema = config.Config.Etcd.EtcdSchema
|
||||
r.etcdAddr = config.Config.Etcd.EtcdAddr
|
||||
}
|
||||
func (r *RPCServer) run() {
|
||||
ip := utils.ServerIP
|
||||
registerAddress := ip + ":" + utils.IntToString(r.rpcPort)
|
||||
listener, err := net.Listen("tcp", registerAddress)
|
||||
if err != nil {
|
||||
log.ErrorByKv("push module rpc listening port err", "", "err", err.Error())
|
||||
return
|
||||
}
|
||||
defer listener.Close()
|
||||
srv := grpc.NewServer()
|
||||
defer srv.GracefulStop()
|
||||
pbPush.RegisterPushMsgServiceServer(srv, r)
|
||||
err = getcdv3.RegisterEtcd(r.etcdSchema, strings.Join(r.etcdAddr, ","), ip, r.rpcPort, r.rpcRegisterName, 10)
|
||||
if err != nil {
|
||||
log.ErrorByKv("register push module rpc to etcd err", "", "err", err.Error())
|
||||
}
|
||||
err = srv.Serve(listener)
|
||||
if err != nil {
|
||||
log.ErrorByKv("push module rpc start err", "", "err", err.Error())
|
||||
return
|
||||
}
|
||||
}
|
||||
func (r *RPCServer) PushMsg(_ context.Context, pbData *pbPush.PushMsgReq) (*pbPush.PushMsgResp, error) {
|
||||
sendPbData := pbRelay.MsgToUserReq{}
|
||||
sendPbData.SendTime = pbData.SendTime
|
||||
sendPbData.OperationID = pbData.OperationID
|
||||
sendPbData.ServerMsgID = pbData.MsgID
|
||||
sendPbData.MsgFrom = pbData.MsgFrom
|
||||
sendPbData.ContentType = pbData.ContentType
|
||||
sendPbData.SenderNickName = pbData.SenderNickName
|
||||
sendPbData.SenderFaceURL = pbData.SenderFaceURL
|
||||
sendPbData.ClientMsgID = pbData.ClientMsgID
|
||||
sendPbData.SessionType = pbData.SessionType
|
||||
sendPbData.RecvID = pbData.RecvID
|
||||
sendPbData.Content = pbData.Content
|
||||
sendPbData.SendID = pbData.SendID
|
||||
sendPbData.PlatformID = pbData.PlatformID
|
||||
sendPbData.RecvSeq = pbData.RecvSeq
|
||||
//Call push module to send message to the user
|
||||
MsgToUser(&sendPbData, pbData.OfflineInfo, pbData.Options)
|
||||
return &pbPush.PushMsgResp{
|
||||
ResultCode: 0,
|
||||
}, nil
|
||||
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
/*
|
||||
** description("").
|
||||
** copyright('open-im,www.open-im.io').
|
||||
** author("fg,Gordon@open-im.io").
|
||||
** time(2021/3/5 14:31).
|
||||
*/
|
||||
package logic
|
||||
|
||||
import (
|
||||
"Open_IM/src/common/config"
|
||||
"Open_IM/src/common/constant"
|
||||
"Open_IM/src/common/log"
|
||||
"Open_IM/src/grpc-etcdv3/getcdv3"
|
||||
pbChat "Open_IM/src/proto/chat"
|
||||
pbGroup "Open_IM/src/proto/group"
|
||||
pbRelay "Open_IM/src/proto/relay"
|
||||
pbGetInfo "Open_IM/src/proto/user"
|
||||
rpcChat "Open_IM/src/rpc/chat/chat"
|
||||
"Open_IM/src/rpc/user/internal_service"
|
||||
"Open_IM/src/utils"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type EChatContent struct {
|
||||
SessionType int `json:"chatType"`
|
||||
From string `json:"from"`
|
||||
To string `json:"to"`
|
||||
Seq int64 `json:"seq"`
|
||||
}
|
||||
|
||||
func MsgToUser(sendPbData *pbRelay.MsgToUserReq, OfflineInfo, Options string) {
|
||||
var wsResult []*pbRelay.SingleMsgToUser
|
||||
isShouldOfflinePush := true
|
||||
MOptions := utils.JsonStringToMap(Options) //Control whether to push message to sender's other terminal
|
||||
//isSenderSync := utils.GetSwitchFromOptions(MOptions, "senderSync")
|
||||
isOfflinePush := utils.GetSwitchFromOptions(MOptions, "offlinePush")
|
||||
log.InfoByKv("Get chat from msg_transfer And push chat", sendPbData.OperationID, "PushData", sendPbData)
|
||||
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName)
|
||||
//Online push message
|
||||
log.InfoByKv("test", sendPbData.OperationID, "len grpc", len(grpcCons), "data", sendPbData)
|
||||
for _, v := range grpcCons {
|
||||
msgClient := pbRelay.NewOnlineMessageRelayServiceClient(v)
|
||||
reply, err := msgClient.MsgToUser(context.Background(), sendPbData)
|
||||
if err != nil {
|
||||
log.InfoByKv("push data to client rpc err", sendPbData.OperationID, "err", err)
|
||||
}
|
||||
if reply != nil && reply.Resp != nil && err == nil {
|
||||
wsResult = append(wsResult, reply.Resp...)
|
||||
}
|
||||
}
|
||||
log.InfoByKv("push_result", sendPbData.OperationID, "result", wsResult)
|
||||
if isOfflinePush {
|
||||
|
||||
for _, t := range pushTerminal {
|
||||
for _, v := range wsResult {
|
||||
if v.RecvPlatFormID == t && v.ResultCode == 0 {
|
||||
isShouldOfflinePush = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if isShouldOfflinePush {
|
||||
//Use offline push messaging
|
||||
var UIDList []string
|
||||
UIDList = append(UIDList, sendPbData.RecvID)
|
||||
var sendUIDList []string
|
||||
sendUIDList = append(sendUIDList, sendPbData.SendID)
|
||||
userInfo, err := internal_service.GetUserInfoClient(&pbGetInfo.GetUserInfoReq{UserIDList: sendUIDList, OperationID: sendPbData.OperationID})
|
||||
if err != nil {
|
||||
log.ErrorByArgs(fmt.Sprintf("err=%v,call GetUserInfoClient rpc server failed", err))
|
||||
return
|
||||
}
|
||||
|
||||
customContent := EChatContent{
|
||||
SessionType: int(sendPbData.SessionType),
|
||||
From: sendPbData.SendID,
|
||||
To: sendPbData.RecvID,
|
||||
Seq: sendPbData.RecvSeq,
|
||||
}
|
||||
bCustomContent, _ := json.Marshal(customContent)
|
||||
|
||||
jsonCustomContent := string(bCustomContent)
|
||||
switch sendPbData.ContentType {
|
||||
case constant.Text:
|
||||
IOSAccountListPush(UIDList, userInfo.Data[0].Name, sendPbData.Content, jsonCustomContent)
|
||||
case constant.Picture:
|
||||
IOSAccountListPush(UIDList, userInfo.Data[0].Name, constant.ContentType2PushContent[constant.Picture], jsonCustomContent)
|
||||
case constant.Voice:
|
||||
IOSAccountListPush(UIDList, userInfo.Data[0].Name, constant.ContentType2PushContent[constant.Voice], jsonCustomContent)
|
||||
case constant.Video:
|
||||
IOSAccountListPush(UIDList, userInfo.Data[0].Name, constant.ContentType2PushContent[constant.Video], jsonCustomContent)
|
||||
case constant.File:
|
||||
IOSAccountListPush(UIDList, userInfo.Data[0].Name, constant.ContentType2PushContent[constant.File], jsonCustomContent)
|
||||
default:
|
||||
|
||||
}
|
||||
|
||||
} else {
|
||||
isShouldOfflinePush = true
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) {
|
||||
m.MsgID = rpcChat.GetMsgID(m.SendID)
|
||||
m.ClientMsgID = m.MsgID
|
||||
switch m.SessionType {
|
||||
case constant.SingleChatType:
|
||||
sendMsgToKafka(m, m.SendID, "msgKey--sendID")
|
||||
sendMsgToKafka(m, m.RecvID, "msgKey--recvID")
|
||||
case constant.GroupChatType:
|
||||
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
|
||||
client := pbGroup.NewGroupClient(etcdConn)
|
||||
req := &pbGroup.GetGroupAllMemberReq{
|
||||
GroupID: m.RecvID,
|
||||
Token: config.Config.Secret,
|
||||
OperationID: m.OperationID,
|
||||
}
|
||||
reply, err := client.GetGroupAllMember(context.Background(), req)
|
||||
if err != nil {
|
||||
log.Error(m.Token, m.OperationID, "rpc getGroupInfo failed, err = %s", err.Error())
|
||||
return
|
||||
}
|
||||
if reply.ErrorCode != 0 {
|
||||
log.Error(m.Token, m.OperationID, "rpc getGroupInfo failed, err = %s", reply.ErrorMsg)
|
||||
return
|
||||
}
|
||||
groupID := m.RecvID
|
||||
for i, v := range reply.MemberList {
|
||||
m.RecvID = v.UserId + " " + groupID
|
||||
sendMsgToKafka(m, utils.IntToString(i), "msgKey--recvID+\" \"+groupID")
|
||||
}
|
||||
default:
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
func sendMsgToKafka(m *pbChat.WSToMsgSvrChatMsg, key string, flag string) {
|
||||
pid, offset, err := producer.SendMessage(m, key)
|
||||
if err != nil {
|
||||
log.ErrorByKv("kafka send failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), flag, key)
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
tpns "Open_IM/internal/push/sdk/tpns-server-sdk-go/go"
|
||||
"Open_IM/internal/push/sdk/tpns-server-sdk-go/go/auth"
|
||||
"Open_IM/internal/push/sdk/tpns-server-sdk-go/go/common"
|
||||
"Open_IM/internal/push/sdk/tpns-server-sdk-go/go/req"
|
||||
"Open_IM/src/common/config"
|
||||
)
|
||||
|
||||
var badgeType = -2
|
||||
var iosAcceptId = auth.Auther{AccessID: config.Config.Push.Tpns.Ios.AccessID, SecretKey: config.Config.Push.Tpns.Ios.SecretKey}
|
||||
|
||||
func IOSAccountListPush(accounts []string, title, content, jsonCustomContent string) {
|
||||
var iosMessage = tpns.Message{
|
||||
Title: title,
|
||||
Content: content,
|
||||
IOS: &tpns.IOSParams{
|
||||
Aps: &tpns.Aps{
|
||||
BadgeType: &badgeType,
|
||||
Sound: "default",
|
||||
Category: "INVITE_CATEGORY",
|
||||
},
|
||||
CustomContent: jsonCustomContent,
|
||||
//CustomContent: `"{"key\":\"value\"}"`,
|
||||
},
|
||||
}
|
||||
pushReq, reqBody, err := req.NewListAccountPush(accounts, iosMessage)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
iosAcceptId.Auth(pushReq, auth.UseSignAuthored, iosAcceptId, reqBody)
|
||||
common.PushAndGetResult(pushReq)
|
||||
}
|
||||
Reference in New Issue
Block a user