pull/131/head
Gordon 3 years ago
parent aaf834a33b
commit d6ec8f4930

@ -49,7 +49,7 @@ func newUserSendMsgReq(token string, params *paramsUserSendMsg) *pbChat.UserSend
RecvID: params.Data.RecvID,
ForceList: params.Data.ForceList,
Content: params.Data.Content,
Options: params.Data.Options,
Options: utils.MapIntToJsonString(params.Data.Options),
ClientMsgID: params.Data.ClientMsgID,
OffLineInfo: utils.MapToJsonString(params.Data.OffLineInfo),
Ex: utils.MapToJsonString(params.Data.Ex),

@ -72,7 +72,7 @@ func newUserSendMsgReq(params *paramsManagementSendMsg) *pbChat.UserSendMsgReq {
ForceList: params.ForceList,
Content: newContent,
ClientMsgID: utils.GetMsgID(params.SendID),
Options: options,
Options: utils.MapIntToJsonString(options),
}
return &pbData
}

@ -214,7 +214,7 @@ func (ws *WServer) sendMsgReq(conn *UserConn, m *Req, sendTime int64) {
SenderNickName: data.SenderNickName,
SenderFaceURL: data.SenderFaceURL,
Content: data.Content,
Options: data.Options,
Options: utils.MapIntToJsonString(data.Options),
ClientMsgID: data.ClientMsgID,
SendTime: sendTime,
}

@ -54,10 +54,11 @@ func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string)
pbSaveData.OperationID = pbData.OperationID
pbSaveData.RecvID = pbData.RecvID
pbSaveData.PlatformID = pbData.PlatformID
options := utils.JsonStringToMap(pbData.Options)
//Control whether to store offline messages (mongo)
isHistory := utils.GetSwitchFromOptions(pbData.Options, "history")
isHistory := utils.GetSwitchFromOptions(options, "history")
//Control whether to store history messages (mysql)
isPersist := utils.GetSwitchFromOptions(pbData.Options, "persistent")
isPersist := utils.GetSwitchFromOptions(options, "persistent")
switch pbData.SessionType {
case constant.SingleChatType:
log.NewDebug(pbSaveData.OperationID, "msg_transfer chat type = SingleChatType", isHistory, isPersist)

@ -40,8 +40,9 @@ func (pc *PersistentConsumerHandler) handleChatWs2Mysql(msg []byte, msgKey strin
log.ErrorByKv("msg_transfer Unmarshal chat err", "", "chat", string(msg), "err", err.Error())
return
}
options := utils.JsonStringToMap(pbData.Options)
//Control whether to store history messages (mysql)
isPersist := utils.GetSwitchFromOptions(pbData.Options, "persistent")
isPersist := utils.GetSwitchFromOptions(options, "persistent")
//Only process receiver data
if isPersist {
if msgKey == pbData.RecvID && pbData.SessionType == constant.SingleChatType {

@ -12,6 +12,7 @@ import (
"Open_IM/pkg/common/log"
pbChat "Open_IM/pkg/proto/chat"
pbRelay "Open_IM/pkg/proto/relay"
"Open_IM/pkg/utils"
"github.com/Shopify/sarama"
"github.com/golang/protobuf/proto"
)
@ -53,7 +54,7 @@ func (ms *PushConsumerHandler) handleMs2PsChat(msg []byte) {
sendPbData.PlatformID = pbData.PlatformID
sendPbData.RecvSeq = pbData.RecvSeq
//Call push module to send message to the user
MsgToUser(&sendPbData, pbData.OfflineInfo, pbData.Options)
MsgToUser(&sendPbData, pbData.OfflineInfo, utils.JsonStringToMap(pbData.Options))
}
func (PushConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (PushConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

@ -71,7 +71,8 @@ func (rpc *rpcChat) UserSendMsg(_ context.Context, pb *pbChat.UserSendMsgReq) (*
} else {
pbData.SendTime = pb.SendTime
}
isHistory := utils.GetSwitchFromOptions(pbData.Options, "history")
options := utils.JsonStringToMap(pbData.Options)
isHistory := utils.GetSwitchFromOptions(options, "history")
mReq := MsgCallBackReq{
SendID: pb.SendID,
RecvID: pb.RecvID,
@ -237,10 +238,12 @@ func modifyMessageByUserMessageReceiveOpt(userID, sourceID string, sessionType i
case constant.NotReceiveMessage:
return false
case constant.ReceiveNotNotifyMessage:
if msg.Options == nil {
msg.Options = make(map[string]int32, 2)
options := utils.JsonStringToMap(msg.Options)
if options == nil {
options = make(map[string]int32, 2)
}
utils.SetSwitchFromOptions(msg.Options, "offlinePush", 0)
utils.SetSwitchFromOptions(options, "offlinePush", 0)
msg.Options = utils.MapIntToJsonString(options)
return true
}

@ -198,14 +198,14 @@ func (d *DataBases) SaveUserChat(uid string, sendTime int64, m *pbMsg.MsgSvrToPu
return errors.New("session == nil")
}
defer session.Close()
log.NewInfo("", "get mgoSession cost time", getCurrentTimestampByMill()-newTime)
log.NewDebug("", "get mgoSession cost time", getCurrentTimestampByMill()-newTime)
c := session.DB(config.Config.Mongo.DBDatabase).C(cChat)
seqUid = getSeqUid(uid, m.RecvSeq)
n, err := c.Find(bson.M{"uid": seqUid}).Count()
if err != nil {
return err
}
log.NewInfo("", "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
log.NewDebug("", "find mgo uid cost time", getCurrentTimestampByMill()-newTime)
sMsg := MsgInfo{}
sMsg.SendTime = sendTime
if sMsg.Msg, err = proto.Marshal(m); err != nil {
@ -225,7 +225,7 @@ func (d *DataBases) SaveUserChat(uid string, sendTime int64, m *pbMsg.MsgSvrToPu
return err
}
}
log.NewInfo("", "insert mgo data cost time", getCurrentTimestampByMill()-newTime)
log.NewDebug("", "insert mgo data cost time", getCurrentTimestampByMill()-newTime)
return nil
}

@ -24,32 +24,32 @@ var _ = math.Inf
const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
type PushMsgReq struct {
SendID string `protobuf:"bytes,1,opt,name=SendID" json:"SendID,omitempty"`
RecvID string `protobuf:"bytes,2,opt,name=RecvID" json:"RecvID,omitempty"`
Content string `protobuf:"bytes,3,opt,name=Content" json:"Content,omitempty"`
RecvSeq int64 `protobuf:"varint,4,opt,name=RecvSeq" json:"RecvSeq,omitempty"`
SendTime int64 `protobuf:"varint,5,opt,name=SendTime" json:"SendTime,omitempty"`
MsgFrom int32 `protobuf:"varint,6,opt,name=MsgFrom" json:"MsgFrom,omitempty"`
ContentType int32 `protobuf:"varint,7,opt,name=ContentType" json:"ContentType,omitempty"`
SessionType int32 `protobuf:"varint,8,opt,name=SessionType" json:"SessionType,omitempty"`
OperationID string `protobuf:"bytes,9,opt,name=OperationID" json:"OperationID,omitempty"`
MsgID string `protobuf:"bytes,10,opt,name=MsgID" json:"MsgID,omitempty"`
OfflineInfo string `protobuf:"bytes,11,opt,name=OfflineInfo" json:"OfflineInfo,omitempty"`
Options map[string]int32 `protobuf:"bytes,12,rep,name=Options" json:"Options,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"`
PlatformID int32 `protobuf:"varint,13,opt,name=PlatformID" json:"PlatformID,omitempty"`
SenderNickName string `protobuf:"bytes,14,opt,name=SenderNickName" json:"SenderNickName,omitempty"`
SenderFaceURL string `protobuf:"bytes,15,opt,name=SenderFaceURL" json:"SenderFaceURL,omitempty"`
ClientMsgID string `protobuf:"bytes,16,opt,name=ClientMsgID" json:"ClientMsgID,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
SendID string `protobuf:"bytes,1,opt,name=SendID" json:"SendID,omitempty"`
RecvID string `protobuf:"bytes,2,opt,name=RecvID" json:"RecvID,omitempty"`
Content string `protobuf:"bytes,3,opt,name=Content" json:"Content,omitempty"`
RecvSeq int64 `protobuf:"varint,4,opt,name=RecvSeq" json:"RecvSeq,omitempty"`
SendTime int64 `protobuf:"varint,5,opt,name=SendTime" json:"SendTime,omitempty"`
MsgFrom int32 `protobuf:"varint,6,opt,name=MsgFrom" json:"MsgFrom,omitempty"`
ContentType int32 `protobuf:"varint,7,opt,name=ContentType" json:"ContentType,omitempty"`
SessionType int32 `protobuf:"varint,8,opt,name=SessionType" json:"SessionType,omitempty"`
OperationID string `protobuf:"bytes,9,opt,name=OperationID" json:"OperationID,omitempty"`
MsgID string `protobuf:"bytes,10,opt,name=MsgID" json:"MsgID,omitempty"`
OfflineInfo string `protobuf:"bytes,11,opt,name=OfflineInfo" json:"OfflineInfo,omitempty"`
Options string `protobuf:"bytes,12,opt,name=Options" json:"Options,omitempty"`
PlatformID int32 `protobuf:"varint,13,opt,name=PlatformID" json:"PlatformID,omitempty"`
SenderNickName string `protobuf:"bytes,14,opt,name=SenderNickName" json:"SenderNickName,omitempty"`
SenderFaceURL string `protobuf:"bytes,15,opt,name=SenderFaceURL" json:"SenderFaceURL,omitempty"`
ClientMsgID string `protobuf:"bytes,16,opt,name=ClientMsgID" json:"ClientMsgID,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *PushMsgReq) Reset() { *m = PushMsgReq{} }
func (m *PushMsgReq) String() string { return proto.CompactTextString(m) }
func (*PushMsgReq) ProtoMessage() {}
func (*PushMsgReq) Descriptor() ([]byte, []int) {
return fileDescriptor_push_380e4afdaa5f6119, []int{0}
return fileDescriptor_push_e44270f7d93180b9, []int{0}
}
func (m *PushMsgReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_PushMsgReq.Unmarshal(m, b)
@ -146,11 +146,11 @@ func (m *PushMsgReq) GetOfflineInfo() string {
return ""
}
func (m *PushMsgReq) GetOptions() map[string]int32 {
func (m *PushMsgReq) GetOptions() string {
if m != nil {
return m.Options
}
return nil
return ""
}
func (m *PushMsgReq) GetPlatformID() int32 {
@ -192,7 +192,7 @@ func (m *PushMsgResp) Reset() { *m = PushMsgResp{} }
func (m *PushMsgResp) String() string { return proto.CompactTextString(m) }
func (*PushMsgResp) ProtoMessage() {}
func (*PushMsgResp) Descriptor() ([]byte, []int) {
return fileDescriptor_push_380e4afdaa5f6119, []int{1}
return fileDescriptor_push_e44270f7d93180b9, []int{1}
}
func (m *PushMsgResp) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_PushMsgResp.Unmarshal(m, b)
@ -221,7 +221,6 @@ func (m *PushMsgResp) GetResultCode() int32 {
func init() {
proto.RegisterType((*PushMsgReq)(nil), "push.PushMsgReq")
proto.RegisterMapType((map[string]int32)(nil), "push.PushMsgReq.OptionsEntry")
proto.RegisterType((*PushMsgResp)(nil), "push.PushMsgResp")
}
@ -297,35 +296,32 @@ var _PushMsgService_serviceDesc = grpc.ServiceDesc{
Metadata: "push/push.proto",
}
func init() { proto.RegisterFile("push/push.proto", fileDescriptor_push_380e4afdaa5f6119) }
var fileDescriptor_push_380e4afdaa5f6119 = []byte{
// 426 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x5c, 0x92, 0x5f, 0x6b, 0xdb, 0x30,
0x14, 0xc5, 0x71, 0x13, 0x27, 0xed, 0x4d, 0xf3, 0x67, 0x62, 0x0c, 0x11, 0xd8, 0x30, 0x65, 0x8c,
0xbc, 0xcc, 0x83, 0xee, 0x61, 0xa3, 0x7b, 0x19, 0xab, 0x57, 0x30, 0x2c, 0x6d, 0x51, 0xba, 0x97,
0xbd, 0xb9, 0xe9, 0x4d, 0x6a, 0xea, 0x48, 0xaa, 0x25, 0x07, 0xf2, 0x11, 0xf7, 0xad, 0x86, 0xae,
0xec, 0xc6, 0xcb, 0x8b, 0xd1, 0xf9, 0xdd, 0xa3, 0x6b, 0xdd, 0xc3, 0x85, 0xb1, 0xae, 0xcc, 0xe3,
0x27, 0xf7, 0x89, 0x75, 0xa9, 0xac, 0x62, 0x5d, 0x77, 0x3e, 0xfb, 0xdb, 0x05, 0xb8, 0xad, 0xcc,
0xe3, 0xdc, 0xac, 0x05, 0x3e, 0xb3, 0x37, 0xd0, 0x5b, 0xa0, 0x7c, 0x48, 0x13, 0x1e, 0x44, 0xc1,
0xec, 0x44, 0xd4, 0xca, 0x71, 0x81, 0xcb, 0x6d, 0x9a, 0xf0, 0x23, 0xcf, 0xbd, 0x62, 0x1c, 0xfa,
0x97, 0x4a, 0x5a, 0x94, 0x96, 0x77, 0xa8, 0xd0, 0x48, 0x57, 0x71, 0x9e, 0x05, 0x3e, 0xf3, 0x6e,
0x14, 0xcc, 0x3a, 0xa2, 0x91, 0x6c, 0x0a, 0xc7, 0xae, 0xeb, 0x5d, 0xbe, 0x41, 0x1e, 0x52, 0xe9,
0x45, 0xbb, 0x5b, 0x73, 0xb3, 0xbe, 0x2a, 0xd5, 0x86, 0xf7, 0xa2, 0x60, 0x16, 0x8a, 0x46, 0xb2,
0x08, 0x06, 0x75, 0xeb, 0xbb, 0x9d, 0x46, 0xde, 0xa7, 0x6a, 0x1b, 0x39, 0xc7, 0x02, 0x8d, 0xc9,
0x95, 0x24, 0xc7, 0xb1, 0x77, 0xb4, 0x90, 0x73, 0xdc, 0x68, 0x2c, 0x33, 0x9b, 0x2b, 0x99, 0x26,
0xfc, 0x84, 0x5e, 0xdc, 0x46, 0xec, 0x35, 0x84, 0x73, 0xb3, 0x4e, 0x13, 0x0e, 0x54, 0xf3, 0x82,
0xee, 0xad, 0x56, 0x45, 0x2e, 0x31, 0x95, 0x2b, 0xc5, 0x07, 0xf5, 0xbd, 0x3d, 0x62, 0x5f, 0xa0,
0x7f, 0xa3, 0x5d, 0x0f, 0xc3, 0x4f, 0xa3, 0xce, 0x6c, 0x70, 0xfe, 0x36, 0xa6, 0xa8, 0xf7, 0xd1,
0xc6, 0x75, 0xfd, 0xa7, 0xb4, 0xe5, 0x4e, 0x34, 0x6e, 0xf6, 0x0e, 0xe0, 0xb6, 0xc8, 0xec, 0x4a,
0x95, 0x9b, 0x34, 0xe1, 0x43, 0x7a, 0x73, 0x8b, 0xb0, 0x0f, 0x30, 0x72, 0xe1, 0x60, 0x79, 0x9d,
0x2f, 0x9f, 0xae, 0xb3, 0x0d, 0xf2, 0x11, 0xfd, 0xfd, 0x80, 0xb2, 0xf7, 0x30, 0xf4, 0xe4, 0x2a,
0x5b, 0xe2, 0x6f, 0xf1, 0x8b, 0x8f, 0xc9, 0xf6, 0x3f, 0xa4, 0x10, 0x8b, 0x1c, 0xa5, 0xf5, 0x43,
0x4e, 0xfc, 0x20, 0x2d, 0x34, 0xbd, 0x80, 0xd3, 0xf6, 0x43, 0xd9, 0x04, 0x3a, 0x4f, 0xb8, 0xab,
0xb7, 0xc1, 0x1d, 0x5d, 0x44, 0xdb, 0xac, 0xa8, 0x90, 0x36, 0x21, 0x14, 0x5e, 0x5c, 0x1c, 0x7d,
0x0d, 0xce, 0x3e, 0xc2, 0xe0, 0x65, 0x5e, 0xa3, 0xdd, 0x68, 0x02, 0x4d, 0x55, 0xd8, 0x4b, 0xf5,
0x80, 0xd4, 0x21, 0x14, 0x2d, 0x72, 0xfe, 0x1d, 0x46, 0xb5, 0x7d, 0x81, 0xe5, 0x36, 0x5f, 0x22,
0x8b, 0xa1, 0x5f, 0x13, 0x36, 0x39, 0xcc, 0x6f, 0xfa, 0xea, 0x80, 0x18, 0xfd, 0x63, 0xfc, 0x67,
0x18, 0xd3, 0x4a, 0x7f, 0xd3, 0xf7, 0x8e, 0xdf, 0xf7, 0x68, 0xb5, 0x3f, 0xff, 0x0b, 0x00, 0x00,
0xff, 0xff, 0xf0, 0x9a, 0xaa, 0x81, 0xed, 0x02, 0x00, 0x00,
func init() { proto.RegisterFile("push/push.proto", fileDescriptor_push_e44270f7d93180b9) }
var fileDescriptor_push_e44270f7d93180b9 = []byte{
// 378 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x5c, 0x92, 0x5d, 0xeb, 0xda, 0x30,
0x14, 0xc6, 0xe9, 0xb4, 0xbe, 0x1c, 0xe7, 0xcb, 0xc2, 0x18, 0xc1, 0x8b, 0x51, 0x64, 0x0c, 0x6f,
0xd6, 0xc1, 0x76, 0xb9, 0x9b, 0x31, 0x8b, 0x50, 0x98, 0x2f, 0xa4, 0xee, 0x66, 0x77, 0xb5, 0x9e,
0x6a, 0x59, 0x9b, 0xc4, 0xa6, 0x0a, 0xfb, 0xd2, 0xfb, 0x0c, 0x23, 0x49, 0xd5, 0xfe, 0xbd, 0x29,
0x7d, 0x7e, 0xe7, 0x39, 0x87, 0x27, 0xc9, 0x81, 0xb1, 0xbc, 0xa8, 0xd3, 0x67, 0xfd, 0xf1, 0x65,
0x29, 0x2a, 0x41, 0xda, 0xfa, 0x7f, 0xf6, 0xaf, 0x05, 0xb0, 0xbd, 0xa8, 0xd3, 0x4a, 0x1d, 0x19,
0x9e, 0xc9, 0x3b, 0xe8, 0x44, 0xc8, 0x0f, 0x61, 0x40, 0x1d, 0xcf, 0x99, 0xf7, 0x59, 0xad, 0x34,
0x67, 0x98, 0x5c, 0xc3, 0x80, 0xbe, 0xb2, 0xdc, 0x2a, 0x42, 0xa1, 0xbb, 0x10, 0xbc, 0x42, 0x5e,
0xd1, 0x96, 0x29, 0xdc, 0xa4, 0xae, 0x68, 0x4f, 0x84, 0x67, 0xda, 0xf6, 0x9c, 0x79, 0x8b, 0xdd,
0x24, 0x99, 0x42, 0x4f, 0x4f, 0xdd, 0x65, 0x05, 0x52, 0xd7, 0x94, 0xee, 0x5a, 0x77, 0xad, 0xd4,
0x71, 0x59, 0x8a, 0x82, 0x76, 0x3c, 0x67, 0xee, 0xb2, 0x9b, 0x24, 0x1e, 0x0c, 0xea, 0xd1, 0xbb,
0xbf, 0x12, 0x69, 0xd7, 0x54, 0x9b, 0x48, 0x3b, 0x22, 0x54, 0x2a, 0x13, 0xdc, 0x38, 0x7a, 0xd6,
0xd1, 0x40, 0xda, 0xb1, 0x91, 0x58, 0xc6, 0x55, 0x26, 0x78, 0x18, 0xd0, 0xbe, 0x49, 0xdc, 0x44,
0xe4, 0x2d, 0xb8, 0x2b, 0x75, 0x0c, 0x03, 0x0a, 0xa6, 0x66, 0x85, 0xe9, 0x4b, 0xd3, 0x3c, 0xe3,
0x18, 0xf2, 0x54, 0xd0, 0x41, 0xdd, 0xf7, 0x40, 0x3a, 0xf7, 0x46, 0xea, 0x19, 0x8a, 0xbe, 0xb6,
0xf7, 0x50, 0x4b, 0xf2, 0x1e, 0x60, 0x9b, 0xc7, 0x55, 0x2a, 0xca, 0x22, 0x0c, 0xe8, 0xd0, 0x84,
0x6a, 0x10, 0xf2, 0x11, 0x46, 0xfa, 0xf4, 0x58, 0xae, 0xb3, 0xe4, 0xcf, 0x3a, 0x2e, 0x90, 0x8e,
0xcc, 0x80, 0x27, 0x4a, 0x3e, 0xc0, 0xd0, 0x92, 0x65, 0x9c, 0xe0, 0x2f, 0xf6, 0x93, 0x8e, 0x8d,
0xed, 0x25, 0x34, 0xb7, 0x94, 0x67, 0xc8, 0x2b, 0x7b, 0x8a, 0x89, 0x4d, 0xda, 0x40, 0xb3, 0x4f,
0x30, 0xb8, 0xbf, 0xb7, 0x92, 0x3a, 0x1e, 0x43, 0x75, 0xc9, 0xab, 0x85, 0x38, 0xa0, 0x79, 0x74,
0x97, 0x35, 0xc8, 0x97, 0xef, 0x30, 0xaa, 0xed, 0x11, 0x96, 0xd7, 0x2c, 0x41, 0xe2, 0x43, 0xb7,
0x26, 0x64, 0xe2, 0x9b, 0x7d, 0x7a, 0xec, 0xcf, 0xf4, 0xcd, 0x13, 0x51, 0xf2, 0xc7, 0xf8, 0xf7,
0xd0, 0x37, 0x7b, 0xf7, 0x4d, 0xee, 0x35, 0xdf, 0x77, 0xcc, 0xfe, 0x7d, 0xfd, 0x1f, 0x00, 0x00,
0xff, 0xff, 0x6b, 0x53, 0xf4, 0xd4, 0x92, 0x02, 0x00, 0x00,
}

@ -111,7 +111,7 @@ func MapIntToJsonString(param map[string]int32) string {
dataString := string(dataType)
return dataString
}
func JsonStringToMap(str string) (tempMap map[string]interface{}) {
func JsonStringToMap(str string) (tempMap map[string]int32) {
_ = json.Unmarshal([]byte(str), &tempMap)
return tempMap
}

Loading…
Cancel
Save