commit

internal/logic/api/logic_int.go

@@ -13,7 +13,8 @@ type LogicIntServer struct{}
 
 // SignIn 设备登录
 func (*LogicIntServer) ConnSignIn(ctx context.Context, req *pb.ConnSignInReq) (*pb.ConnSignInResp, error) {
-	return &pb.ConnSignInResp{}, service.AuthService.SignIn(ctx, req.UserId, req.DeviceId, req.Token, req.ConnAddr, req.ConnFd)
+	return &pb.ConnSignInResp{},
+		service.AuthService.SignIn(ctx, req.UserId, req.DeviceId, req.Token, req.ConnAddr, req.ConnFd, req.ClientAddr)
 }
 
 // Sync 设备同步消息
@@ -28,7 +29,7 @@ func (*LogicIntServer) MessageACK(ctx context.Context, req *pb.MessageACKReq) (*
 
 // Offline 设备离线
 func (*LogicIntServer) Offline(ctx context.Context, req *pb.OfflineReq) (*pb.OfflineResp, error) {
-	return &pb.OfflineResp{}, service.DeviceService.Offline(ctx, req.UserId, req.DeviceId)
+	return &pb.OfflineResp{}, service.DeviceService.Offline(ctx, req.UserId, req.DeviceId, req.ClientAddr)
 }
 
 // SendMessage 发送消息

internal/logic/dao/device.go

@@ -32,9 +32,9 @@ func (*deviceDao) Get(deviceId int64) (*model.Device, error) {
 		return nil, gerrors.WrapError(err)
 	}
 	if err == gorm.ErrRecordNotFound {
-		return nil, err
+		return nil, nil
 	}
-	return &device, err
+	return &device, nil
 }
 
 // ListUserOnline 查询用户所有的在线设备
@@ -48,9 +48,9 @@ func (*deviceDao) ListOnlineByUserId(userId int64) ([]model.Device, error) {
 }
 
 // UpdateUserIdAndStatus 更新设备绑定用户和设备在线状态
-func (*deviceDao) UpdateUserIdAndStatus(deviceId, userId int64, status int, connAddr string, connFd int64) error {
-	err := db.DB.Exec("update device  set user_id = ?,status = ?,conn_addr = ?,conn_fd = ? where id = ? ",
-		userId, status, connAddr, connFd, deviceId).Error
+func (*deviceDao) Update(deviceId, userId int64, status int, connAddr string, connFd int64, clientAddr string) error {
+	err := db.DB.Exec("update device set user_id = ?,status = ?,conn_addr = ?,conn_fd = ?,client_addr = ? where id = ? ",
+		userId, status, connAddr, connFd, clientAddr, deviceId).Error
 	if err != nil {
 		return gerrors.WrapError(err)
 	}

internal/logic/model/device.go

@@ -19,6 +19,7 @@ type Device struct {
 	Status        int32     // 在线状态，0：不在线；1：在线
 	ConnAddr      string    // 连接层服务层地址
 	ConnFd        int64     // TCP连接对应的文件描述符
+	ClientAddr    string    // 客户端地址
 	CreateTime    time.Time // 创建时间
 	UpdateTime    time.Time // 更新时间
 }

internal/logic/service/auth.go

@@ -11,14 +11,14 @@ type authService struct{}
 var AuthService = new(authService)
 
 // SignIn 长连接登录
-func (*authService) SignIn(ctx context.Context, userId, deviceId int64, token string, connAddr string, connFd int64) error {
+func (*authService) SignIn(ctx context.Context, userId, deviceId int64, token string, connAddr string, connFd int64, clientAddr string) error {
 	_, err := rpc.UserIntClient.Auth(ctx, &pb.AuthReq{UserId: userId, DeviceId: deviceId, Token: token})
 	if err != nil {
 		return err
 	}
 
 	// 标记用户在设备上登录
-	err = DeviceService.Online(ctx, deviceId, userId, connAddr, connFd)
+	err = DeviceService.Online(ctx, deviceId, userId, connAddr, connFd, clientAddr)
 	if err != nil {
 		return err
 	}

internal/logic/service/device.go

@@ -5,6 +5,9 @@ import (
 	"gim/internal/logic/cache"
 	"gim/internal/logic/dao"
 	"gim/internal/logic/model"
+	"gim/pkg/logger"
+
+	"go.uber.org/zap"
 )
 
 const (
@@ -75,8 +78,8 @@ func (*deviceService) ListOnlineByUserId(ctx context.Context, userId int64) ([]m
 }
 
 // Online 设备上线
-func (*deviceService) Online(ctx context.Context, deviceId, userId int64, connAddr string, connFd int64) error {
-	err := dao.DeviceDao.UpdateUserIdAndStatus(deviceId, userId, DeviceOnline, connAddr, connFd)
+func (*deviceService) Online(ctx context.Context, deviceId, userId int64, connAddr string, connFd int64, clientAddr string) error {
+	err := dao.DeviceDao.Update(deviceId, userId, DeviceOnline, connAddr, connFd, clientAddr)
 	if err != nil {
 		return err
 	}
@@ -89,8 +92,20 @@ func (*deviceService) Online(ctx context.Context, deviceId, userId int64, connAd
 }
 
 // Offline 设备离线
-func (*deviceService) Offline(ctx context.Context, userId, deviceId int64) error {
-	err := dao.DeviceDao.UpdateStatus(deviceId, DeviceOffline)
+func (*deviceService) Offline(ctx context.Context, userId, deviceId int64, clientAddr string) error {
+	device, err := dao.DeviceDao.Get(deviceId)
+	if err != nil {
+		return err
+	}
+	if device == nil {
+		logger.Logger.Warn("device is nil", zap.Int64("device_id", deviceId))
+		return nil
+	}
+	if device.ClientAddr != clientAddr {
+		return nil
+	}
+
+	err = dao.DeviceDao.UpdateStatus(deviceId, DeviceOffline)
 	if err != nil {
 		return err
 	}

internal/tcp_conn/handler.go

@@ -58,8 +58,9 @@ func (*handler) OnClose(c *gn.Conn, err error) {
 	logger.Logger.Debug("close", zap.String("addr", c.GetAddr()), zap.Int32("fd", c.GetFd()), zap.Any("data", c.GetData()), zap.Error(err))
 	if data, ok := c.GetData().(ConnData); ok {
 		_, _ = rpc.LogicIntClient.Offline(context.TODO(), &pb.OfflineReq{
-			UserId:   data.UserId,
-			DeviceId: data.DeviceId,
+			UserId:     data.UserId,
+			DeviceId:   data.DeviceId,
+			ClientAddr: c.GetAddr(),
 		})
 	}
 }
@@ -108,11 +109,12 @@ func (h *handler) SignIn(c *gn.Conn, input pb.Input) {
 	}
 
 	_, err = rpc.LogicIntClient.ConnSignIn(grpclib.ContextWithRequstId(context.TODO(), input.RequestId), &pb.ConnSignInReq{
-		UserId:   signIn.UserId,
-		DeviceId: signIn.DeviceId,
-		Token:    signIn.Token,
-		ConnAddr: config.TCPConn.LocalAddr,
-		ConnFd:   int64(c.GetFd()),
+		UserId:     signIn.UserId,
+		DeviceId:   signIn.DeviceId,
+		Token:      signIn.Token,
+		ConnAddr:   config.TCPConn.LocalAddr,
+		ConnFd:     int64(c.GetFd()),
+		ClientAddr: c.GetAddr(),
 	})
 
 	h.Send(c, pb.PackageType_PT_SIGN_IN, input.RequestId, err, nil)

internal/ws_conn/ws_conn_context.go

@@ -192,8 +192,9 @@ func (c *WSConnContext) Release() {
 	// 通知业务服务器设备下线
 	if c.DeviceId != PreConn {
 		_, _ = rpc.LogicIntClient.Offline(context.TODO(), &pb.OfflineReq{
-			UserId:   c.UserId,
-			DeviceId: c.DeviceId,
+			UserId:     c.UserId,
+			DeviceId:   c.DeviceId,
+			ClientAddr: c.Conn.RemoteAddr().String(),
 		})
 	}
 }

internal/ws_conn/ws_server.go

@@ -43,10 +43,11 @@ func wsHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	_, err := rpc.LogicIntClient.ConnSignIn(grpclib.ContextWithRequstId(context.TODO(), requestId), &pb.ConnSignInReq{
-		UserId:   userId,
-		DeviceId: deviceId,
-		Token:    token,
-		ConnAddr: config.WSConn.LocalAddr,
+		UserId:     userId,
+		DeviceId:   deviceId,
+		Token:      token,
+		ConnAddr:   config.WSConn.LocalAddr,
+		ClientAddr: r.Host,
 	})
 
 	s, _ := status.FromError(err)

pkg/proto/logic.int.proto

@@ -10,6 +10,7 @@ message ConnSignInReq {
     string token = 3; // 秘钥
     string conn_addr = 4; // 服务器地址
     int64 conn_fd = 5; // TCP对应的文件描述符
+    string client_addr = 6; // 客户端地址
 }
 message ConnSignInResp {
 }
@@ -37,6 +38,7 @@ message MessageACKResp {
 message OfflineReq {
     int64 user_id = 1; // 用户id
     int64 device_id = 2; // 设备id
+    string client_addr = 3; // 客户端地址
 }
 message OfflineResp {
 

sql/create_table.sql

@@ -14,6 +14,7 @@ CREATE TABLE `device`
     `status`         tinyint(3)          NOT NULL DEFAULT '0' COMMENT '在线状态，0：离线；1：在线',
     `conn_addr`      varchar(25)         NOT NULL COMMENT '连接层服务器地址',
     `conn_fd`        bigint(20)          NOT NULL COMMENT 'TCP连接对应的文件描述符',
+    `client_addr`    varchar(25)         NOT NULL COMMENT '客户端地址',
     `create_time`    datetime            NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
     `update_time`    datetime            NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
     PRIMARY KEY (`id`),