golang-im聊天
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

message_service.go 7.3 KiB

2 years ago
1 year ago
1 year ago
1 year ago
2 years ago
1 year ago
2 years ago
1 year ago
2 years ago
1 year ago
2 years ago
1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
2 years ago
2 years ago
1 year ago
2 years ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244
  1. package service
  2. import (
  3. "context"
  4. "gim/internal/business/comm/db"
  5. svc "gim/internal/business/comm/service"
  6. "gim/internal/business/comm/utils"
  7. repo2 "gim/internal/business/domain/user/repo"
  8. "gim/internal/logic/domain/message/md"
  9. "gim/internal/logic/domain/message/model"
  10. "gim/internal/logic/domain/message/repo"
  11. "gim/internal/logic/proxy"
  12. "gim/pkg/grpclib"
  13. "gim/pkg/grpclib/picker"
  14. "gim/pkg/logger"
  15. "gim/pkg/pb"
  16. "gim/pkg/rpc"
  17. "gim/pkg/util"
  18. "go.uber.org/zap"
  19. "google.golang.org/protobuf/proto"
  20. "strconv"
  21. "time"
  22. )
  23. const MessageLimit = 50 // 最大消息同步数量
  24. const MaxSyncBufLen = 65536 // 最大字节数组长度
  25. type messageService struct{}
  26. var MessageService = new(messageService)
  27. // Sync 消息同步
  28. func (*messageService) Sync(ctx context.Context, userId, seq int64) (*pb.SyncResp, error) {
  29. messages, hasMore, err := MessageService.ListByUserIdAndSeq(ctx, userId, seq)
  30. if err != nil {
  31. return nil, err
  32. }
  33. pbMessages := model.MessagesToPB(messages)
  34. length := len(pbMessages)
  35. resp := &pb.SyncResp{Messages: pbMessages, HasMore: hasMore}
  36. bytes, err := proto.Marshal(resp)
  37. if err != nil {
  38. return nil, err
  39. }
  40. // 如果字节数组大于一个包的长度,需要减少字节数组
  41. for len(bytes) > MaxSyncBufLen {
  42. length = length * 2 / 3
  43. resp = &pb.SyncResp{Messages: pbMessages[0:length], HasMore: true}
  44. bytes, err = proto.Marshal(resp)
  45. if err != nil {
  46. return nil, err
  47. }
  48. }
  49. var userIds = make(map[int64]int32, len(resp.Messages))
  50. for i := range resp.Messages {
  51. if resp.Messages[i].Sender.SenderType == pb.SenderType_ST_USER {
  52. userIds[resp.Messages[i].Sender.SenderId] = 0
  53. }
  54. }
  55. usersResp, err := rpc.GetBusinessIntClient().GetUsers(ctx, &pb.GetUsersReq{UserIds: userIds})
  56. if err != nil {
  57. return nil, err
  58. }
  59. for i := range resp.Messages {
  60. if resp.Messages[i].Sender.SenderType == pb.SenderType_ST_USER {
  61. user, ok := usersResp.Users[resp.Messages[i].Sender.SenderId]
  62. if ok {
  63. resp.Messages[i].Sender.Nickname = user.Nickname
  64. resp.Messages[i].Sender.AvatarUrl = user.AvatarUrl
  65. resp.Messages[i].Sender.Extra = user.Extra
  66. } else {
  67. logger.Logger.Warn("get user failed", zap.Int64("user_id", resp.Messages[i].Sender.SenderId))
  68. }
  69. }
  70. }
  71. return resp, nil
  72. }
  73. // ListByUserIdAndSeq 查询消息
  74. func (*messageService) ListByUserIdAndSeq(ctx context.Context, userId, seq int64) ([]model.Message, bool, error) {
  75. var err error
  76. if seq == 0 {
  77. seq, err = DeviceAckService.GetMaxByUserId(ctx, userId)
  78. if err != nil {
  79. return nil, false, err
  80. }
  81. }
  82. return repo.MessageRepo.ListBySeq(userId, seq, MessageLimit)
  83. }
  84. // SendToUser 将消息发送给用户
  85. func (*messageService) SendToUser(ctx context.Context, sender *pb.Sender, toUserId int64, req *pb.SendMessageReq) (int64, error) {
  86. masterId, _ := grpclib.GetCtxMasterId(ctx)
  87. logger.Logger.Debug("SendToUser",
  88. zap.String("master_id", masterId),
  89. zap.Int64("request_id", grpclib.GetCtxRequestId(ctx)),
  90. zap.Int64("to_user_id", toUserId))
  91. var (
  92. seq int64 = 0
  93. err error
  94. )
  95. if req.IsPersist {
  96. seq, err = SeqService.GetUserNext(ctx, toUserId)
  97. if err != nil {
  98. return 0, err
  99. }
  100. selfMessage := model.Message{
  101. UserId: toUserId,
  102. RequestId: grpclib.GetCtxRequestId(ctx),
  103. SenderType: int32(sender.SenderType),
  104. SenderId: sender.SenderId,
  105. ReceiverType: int32(req.ReceiverType),
  106. ReceiverId: req.ReceiverId,
  107. ToUserIds: model.FormatUserIds(req.ToUserIds),
  108. Type: int(req.MessageType),
  109. Content: req.MessageContent,
  110. Seq: seq,
  111. SendTime: util.UnunixMilliTime(req.SendTime),
  112. Status: int32(pb.MessageStatus_MS_NORMAL),
  113. }
  114. err = repo.MessageRepo.Save(selfMessage)
  115. if err != nil {
  116. logger.Sugar.Error(err)
  117. return 0, err
  118. }
  119. if sender.SenderType == pb.SenderType_ST_USER && sender.SenderId == toUserId {
  120. // 用户需要增加自己的已经同步的序列号
  121. err = repo.DeviceACKRepo.Set(sender.SenderId, sender.DeviceId, seq)
  122. if err != nil {
  123. return 0, err
  124. }
  125. }
  126. }
  127. message := pb.Message{
  128. Sender: sender,
  129. ReceiverType: req.ReceiverType,
  130. ReceiverId: req.ReceiverId,
  131. ToUserIds: req.ToUserIds,
  132. MessageType: req.MessageType,
  133. MessageContent: req.MessageContent,
  134. Seq: seq,
  135. SendTime: req.SendTime,
  136. Status: pb.MessageStatus_MS_NORMAL,
  137. }
  138. // 查询用户在线设备
  139. devices, err := proxy.DeviceProxy.ListOnlineByUserId(ctx, toUserId)
  140. if err != nil {
  141. logger.Sugar.Error(err)
  142. return 0, err
  143. }
  144. isOpenAppPush := svc.SysCfgGet(masterId, "is_open_app_push")
  145. utils.FilePutContents("SendToUser", utils.SerializeStr(map[string]interface{}{
  146. "devices_len": len(devices),
  147. "sender": sender,
  148. "toUserId": toUserId,
  149. "isOpenAppPush": isOpenAppPush,
  150. "req": req,
  151. }))
  152. if len(devices) <= 0 {
  153. //isOpenAppPush := svc.SysCfgGet(masterId, "is_open_app_push")
  154. if sender.SenderType == pb.SenderType_ST_USER && req.ReceiverType == pb.ReceiverType_RT_USER && isOpenAppPush == "1" && req.ReceiverId != toUserId {
  155. uid := strconv.FormatInt(req.ReceiverId, 10)
  156. alia := db.DbUserPushForJg.UserPushForJgGetWithDb(masterId, uid)
  157. if alia != "" {
  158. if sender.Nickname == "" {
  159. sendUser, err := repo2.UserRepo.Get(sender.SenderId)
  160. if err != nil {
  161. logger.Sugar.Error(err)
  162. return 0, err
  163. }
  164. sender.Nickname = sendUser.Nickname
  165. sender.AvatarUrl = sendUser.AvatarUrl
  166. }
  167. //TODO::接收者类型为`user`, 进行极光推送
  168. CommAddPush(md.PushParams{
  169. MasterId: masterId,
  170. Uid: uid,
  171. PushAlia: alia,
  172. Title: "新消息提醒",
  173. Content: "您收到一条新消息,来自会员\"[消息发送者-会员昵称]\",发送时间\"[时间]\"",
  174. //Content: "您收到一条新消息,来自会员\"[消息发送者-会员昵称]\",发送时间\"[时间]\",消息类型\"[消息类型]\",备注\"[备注]\"",
  175. PushType: "zhi_ying_gim",
  176. MessageType: req.MessageType.String(),
  177. SendUserNickname: sender.Nickname,
  178. SendUserAvatarUrl: sender.AvatarUrl,
  179. Memo: sender.SenderType.String(),
  180. Times: time.Now().Format("2006-01-02 15:04:05"),
  181. })
  182. }
  183. }
  184. }
  185. for i := range devices {
  186. if sender.DeviceId == devices[i].DeviceId {
  187. // 消息不需要投递给发送消息的设备
  188. continue
  189. }
  190. err = MessageService.SendToDevice(ctx, devices[i], &message)
  191. if err != nil {
  192. logger.Sugar.Error(err, zap.Any("SendToUser error", devices[i]), zap.Error(err))
  193. }
  194. }
  195. return seq, nil
  196. }
  197. // SendToDevice 将消息发送给设备
  198. func (*messageService) SendToDevice(ctx context.Context, device *pb.Device, message *pb.Message) error {
  199. messageSend := pb.MessageSend{Message: message}
  200. _, err := rpc.GetConnectIntClient().DeliverMessage(picker.ContextWithAddr(ctx, device.ConnAddr), &pb.DeliverMessageReq{
  201. DeviceId: device.DeviceId,
  202. MessageSend: &messageSend,
  203. })
  204. if err != nil {
  205. logger.Logger.Error("SendToDevice error", zap.Error(err))
  206. return err
  207. }
  208. // todo 其他推送厂商
  209. return nil
  210. }
  211. func (*messageService) AddSenderInfo(sender *pb.Sender) {
  212. if sender.SenderType == pb.SenderType_ST_USER {
  213. user, err := rpc.GetBusinessIntClient().GetUser(context.TODO(), &pb.GetUserReq{UserId: sender.SenderId})
  214. if err == nil && user != nil {
  215. sender.AvatarUrl = user.User.AvatarUrl
  216. sender.Nickname = user.User.Nickname
  217. sender.Extra = user.User.Extra
  218. }
  219. }
  220. }