Skip to content

Commit 1f4971e

Browse files
hezhuozhuohezhuozhuo
committed
refactor: 更新心跳判断逻辑
1 parent 3ae20ac commit 1f4971e

5 files changed

Lines changed: 63 additions & 45 deletions

File tree

‎internal/dal/devices.go‎

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,11 @@ func UpdateDeviceByMap(deviceID string, deviceMap map[string]interface{}) (*mode
6161
}
6262

6363
// 更新设备状态
64-
func UpdateDeviceStatus(deviceId string, status int16) error {
64+
// 返回值: (是否真的更新了状态, error)
65+
// 如果状态没有变化，返回 (false, nil)
66+
// 如果状态真的更新了，返回 (true, nil)
67+
// 如果发生错误，返回 (false, error)
68+
func UpdateDeviceStatus(deviceId string, status int16) (bool, error) {
6569
// 1. 先从 Redis 缓存读取设备信息，缓存未命中则从数据库加载
6670
var device *model.Device
6771
result, err := global.REDIS.Get(context.Background(), deviceId).Result()
@@ -78,15 +82,15 @@ func UpdateDeviceStatus(deviceId string, status int16) error {
7882
device, err = query.Device.Where(query.Device.ID.Eq(deviceId)).First()
7983
if err != nil {
8084
logrus.WithError(err).WithField("device_id", deviceId).Error("Failed to get device info")
81-
return err
85+
return false, err
8286
}
8387
}
8488

8589
// 2. 检查状态是否发生变化，如果没变化则不更新
8690
currentStatus := device.IsOnline
8791
if currentStatus == status {
8892
// 状态未变化，不需要更新
89-
return nil
93+
return false, nil
9094
}
9195

9296
// 3. 更新设备状态
@@ -100,20 +104,20 @@ func UpdateDeviceStatus(deviceId string, status int16) error {
100104
})
101105
if err != nil {
102106
logrus.Error(err)
103-
return err
107+
return false, err
104108
}
105109
if info.RowsAffected == 0 {
106-
return fmt.Errorf("update device status failed, no rows affected")
110+
return false, fmt.Errorf("update device status failed, no rows affected")
107111
}
108112
} else {
109113
// 设备上线时，只更新is_online字段
110114
info, err := query.Device.Where(query.Device.ID.Eq(deviceId)).Update(query.Device.IsOnline, status)
111115
if err != nil {
112116
logrus.Error(err)
113-
return err
117+
return false, err
114118
}
115119
if info.RowsAffected == 0 {
116-
return fmt.Errorf("update device status failed, no rows affected")
120+
return false, fmt.Errorf("update device status failed, no rows affected")
117121
}
118122
}
119123

@@ -133,7 +137,7 @@ func UpdateDeviceStatus(deviceId string, status int16) error {
133137
}
134138
}()
135139

136-
return nil
140+
return true, nil
137141
}
138142

139143
func DeleteDevice(id string, tenantID string) error {

‎internal/uplink/attribute.go‎

Lines changed: 16 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -352,25 +352,29 @@ func (f *AttributeUplink) refreshHeartbeat(device *model.Device) {
352352
// 检查是否需要自动上线
353353
if device.IsOnline != 1 {
354354
// 设备当前离线,收到消息后自动上线
355-
if err := dal.UpdateDeviceStatus(device.ID, 1); err != nil {
355+
statusChanged, err := dal.UpdateDeviceStatus(device.ID, 1)
356+
if err != nil {
356357
f.logger.WithError(err).WithField("device_id", device.ID).Error("Failed to auto online device")
357358
return
358359
}
359360

360-
f.logger.WithField("device_id", device.ID).Info("Device auto online by business message")
361+
// 只有当状态真的从离线变为在线时，才执行上线通知流程
362+
if statusChanged {
363+
f.logger.WithField("device_id", device.ID).Info("Device auto online by business message")
361364

362-
// 清理缓存
363-
initialize.DelDeviceCache(device.ID)
365+
// 清理缓存
366+
initialize.DelDeviceCache(device.ID)
364367

365-
// 获取最新设备信息
366-
updatedDevice, err := initialize.GetDeviceCacheById(device.ID)
367-
if err != nil {
368-
f.logger.WithError(err).Error("Failed to get updated device")
369-
return
370-
}
368+
// 获取最新设备信息
369+
updatedDevice, err := initialize.GetDeviceCacheById(device.ID)
370+
if err != nil {
371+
f.logger.WithError(err).Error("Failed to get updated device")
372+
return
373+
}
371374

372-
// SSE通知、自动化、预期数据(异步)
373-
go f.notifyDeviceOnline(updatedDevice)
375+
// SSE通知、自动化、预期数据(异步)
376+
go f.notifyDeviceOnline(updatedDevice)
377+
}
374378
}
375379

376380
// 刷新心跳 key (优先级: heartbeat > online_timeout)

‎internal/uplink/event.go‎

Lines changed: 16 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -348,25 +348,29 @@ func (f *EventUplink) refreshHeartbeat(device *model.Device) {
348348
// 检查是否需要自动上线
349349
if device.IsOnline != 1 {
350350
// 设备当前离线,收到消息后自动上线
351-
if err := dal.UpdateDeviceStatus(device.ID, 1); err != nil {
351+
statusChanged, err := dal.UpdateDeviceStatus(device.ID, 1)
352+
if err != nil {
352353
f.logger.WithError(err).WithField("device_id", device.ID).Error("Failed to auto online device")
353354
return
354355
}
355356

356-
f.logger.WithField("device_id", device.ID).Info("Device auto online by business message")
357+
// 只有当状态真的从离线变为在线时，才执行上线通知流程
358+
if statusChanged {
359+
f.logger.WithField("device_id", device.ID).Info("Device auto online by business message")
357360

358-
// 清理缓存
359-
initialize.DelDeviceCache(device.ID)
361+
// 清理缓存
362+
initialize.DelDeviceCache(device.ID)
360363

361-
// 获取最新设备信息
362-
updatedDevice, err := initialize.GetDeviceCacheById(device.ID)
363-
if err != nil {
364-
f.logger.WithError(err).Error("Failed to get updated device")
365-
return
366-
}
364+
// 获取最新设备信息
365+
updatedDevice, err := initialize.GetDeviceCacheById(device.ID)
366+
if err != nil {
367+
f.logger.WithError(err).Error("Failed to get updated device")
368+
return
369+
}
367370

368-
// SSE通知、自动化、预期数据(异步)
369-
go f.notifyDeviceOnline(updatedDevice)
371+
// SSE通知、自动化、预期数据(异步)
372+
go f.notifyDeviceOnline(updatedDevice)
373+
}
370374
}
371375

372376
// 刷新心跳 key (优先级: heartbeat > online_timeout)

‎internal/uplink/status.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -137,13 +137,15 @@ func (f *StatusUplink) processMessage(msg *DeviceMessage) {
137137
}
138138

139139
// 4. 更新数据库状态
140-
if err := dal.UpdateDeviceStatus(device.ID, status); err != nil {
140+
statusChanged, err := dal.UpdateDeviceStatus(device.ID, status)
141+
if err != nil {
141142
f.logger.WithError(err).WithFields(logrus.Fields{
142143
"device_id": device.ID,
143144
"status": status,
144145
}).Error("Failed to update device status")
145146
return
146147
}
148+
_ = statusChanged // 状态上报入口，不需要检查是否变化
147149

148150
f.logger.WithFields(logrus.Fields{
149151
"device_id": device.ID,

‎internal/uplink/telemetry.go‎

Lines changed: 16 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -405,25 +405,29 @@ func (f *TelemetryUplink) refreshHeartbeat(device *model.Device) {
405405
// 检查是否需要自动上线
406406
if device.IsOnline != 1 {
407407
// 设备当前离线,收到消息后自动上线
408-
if err := dal.UpdateDeviceStatus(device.ID, 1); err != nil {
408+
statusChanged, err := dal.UpdateDeviceStatus(device.ID, 1)
409+
if err != nil {
409410
f.logger.WithError(err).WithField("device_id", device.ID).Error("Failed to auto online device")
410411
return
411412
}
412413

413-
f.logger.WithField("device_id", device.ID).Info("Device auto online by business message")
414+
// 只有当状态真的从离线变为在线时，才执行上线通知流程
415+
if statusChanged {
416+
f.logger.WithField("device_id", device.ID).Info("Device auto online by business message")
414417

415-
// 清理缓存
416-
initialize.DelDeviceCache(device.ID)
418+
// 清理缓存
419+
initialize.DelDeviceCache(device.ID)
417420

418-
// 获取最新设备信息
419-
updatedDevice, err := initialize.GetDeviceCacheById(device.ID)
420-
if err != nil {
421-
f.logger.WithError(err).Error("Failed to get updated device")
422-
return
423-
}
421+
// 获取最新设备信息
422+
updatedDevice, err := initialize.GetDeviceCacheById(device.ID)
423+
if err != nil {
424+
f.logger.WithError(err).Error("Failed to get updated device")
425+
return
426+
}
424427

425-
// SSE通知、自动化、预期数据(异步)
426-
go f.notifyDeviceOnline(updatedDevice)
428+
// SSE通知、自动化、预期数据(异步)
429+
go f.notifyDeviceOnline(updatedDevice)
430+
}
427431
}
428432

429433
// 刷新心跳 key (优先级: heartbeat > online_timeout)

0 commit comments

Comments
 (0)