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 6.3 KiB

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