pig/pig-farm-controller
1
0
Fork 0
Code Issues 16 Pull Requests Actions Packages Projects Releases Wiki Activity

增加ping指令并获取带版本号的响应

Browse Source
This commit is contained in:
huang
2025-12-01 20:42:21 +08:00
parent d5999cee7e
commit 49555c0842
17 changed files with 557 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
internal/domain/device/device_service.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/proto"
)
// 设备行为
@@ -21,6 +22,26 @@ var (
MethodSwitch Method = "switch" // 启停指令
)
// SendOptions 包含了发送通用指令时的可选参数。
type SendOptions struct {
// NotTrackable 如果为 true，则指示本次发送无需被追踪。
// 这将阻止系统为本次发送创建 device_command_logs 记录。
// 默认为 false，即需要追踪。
NotTrackable bool
}
// SendOption 是一个函数类型，用于修改 SendOptions。
// 这是实现 "Functional Options Pattern" 的核心。
type SendOption func(*SendOptions)
// WithoutTracking 是一个公开的选项函数，用于明确指示本次发送无需追踪。
// 调用方在发送 Ping 等无需响应确认的指令时，应使用此选项。
func WithoutTracking() SendOption {
return func(opts *SendOptions) {
opts.NotTrackable = true
}
}
// Service 抽象了一组方法用于控制设备行为
type Service interface {
@@ -29,6 +50,10 @@ type Service interface {
// Collect 用于发起对指定区域主控下的多个设备的批量采集请求。
Collect(ctx context.Context, areaControllerID uint32, devicesToCollect []*models.Device) error
// Send 是一个通用的发送方法，用于将一个标准的指令载荷发送到指定的区域主控。
// 它负责将载荷包装成顶层指令、序列化、调用底层发送器，并默认记录下行命令日志。
Send(ctx context.Context, areaControllerID uint32, payload proto.InstructionPayload, opts ...SendOption) error
}
// 设备操作指令通用结构(最外层)

70
internal/domain/device/general_device_service.go
View File

@@ -20,6 +20,7 @@ import (
type GeneralDeviceService struct {
ctx context.Context
deviceRepo repository.DeviceRepository
areaControllerRepo repository.AreaControllerRepository
deviceCommandLogRepo repository.DeviceCommandLogRepository
pendingCollectionRepo repository.PendingCollectionRepository
comm transport.Communicator
@@ -29,6 +30,7 @@ type GeneralDeviceService struct {
func NewGeneralDeviceService(
ctx context.Context,
deviceRepo repository.DeviceRepository,
areaControllerRepo repository.AreaControllerRepository,
deviceCommandLogRepo repository.DeviceCommandLogRepository,
pendingCollectionRepo repository.PendingCollectionRepository,
comm transport.Communicator,
@@ -36,6 +38,7 @@ func NewGeneralDeviceService(
return &GeneralDeviceService{
ctx: ctx,
deviceRepo: deviceRepo,
areaControllerRepo: areaControllerRepo,
deviceCommandLogRepo: deviceCommandLogRepo,
pendingCollectionRepo: pendingCollectionRepo,
comm: comm,
@@ -249,3 +252,70 @@ func (g *GeneralDeviceService) Collect(ctx context.Context, areaControllerID uin
logger.Debugf("成功将采集请求 (CorrelationID: %s) 发送到设备 %s", correlationID, networkID)
return nil
}
// Send 实现了 Service 接口，用于发送一个通用的指令载荷。
// 它将载荷包装成顶层指令，然后执行查找网络地址、序列化、发送和记录日志的完整流程。
func (g *GeneralDeviceService) Send(ctx context.Context, areaControllerID uint32, payload proto.InstructionPayload, opts ...SendOption) error {
serviceCtx, logger := logs.Trace(ctx, g.ctx, "Send")
// 1. 应用选项
options := &SendOptions{}
for _, opt := range opts {
opt(options)
}
// 2. 查找区域主控以获取 NetworkID
areaController, err := g.areaControllerRepo.FindByID(serviceCtx, areaControllerID)
if err != nil {
return fmt.Errorf("发送通用指令失败：无法找到ID为 %d 的区域主控: %w", areaControllerID, err)
}
// 3. 将载荷包装进顶层 Instruction 结构体
instruction := &proto.Instruction{
Payload: payload,
}
// 4. 序列化指令
message, err := gproto.Marshal(instruction)
if err != nil {
return fmt.Errorf("序列化通用指令失败: %w", err)
}
// 5. 发送指令
networkID := areaController.NetworkID
sendResult, err := g.comm.Send(serviceCtx, networkID, message)
if err != nil {
return fmt.Errorf("发送通用指令到 %s 失败: %w", networkID, err)
}
// 6. 始终创建 DeviceCommandLog 记录，但根据选项设置其初始状态
logRecord := &models.DeviceCommandLog{
MessageID: sendResult.MessageID,
DeviceID: areaController.ID, // 将日志与区域主控关联
SentAt: time.Now(),
}
if options.NotTrackable {
// 对于无需追踪的指令，直接标记为已完成
now := time.Now()
logRecord.AcknowledgedAt = &now
logRecord.ReceivedSuccess = true
logger.Infow("成功发送一个无需追踪的通用指令，并记录为已完成日志", "networkID", networkID, "MessageID", sendResult.MessageID)
} else {
// 对于需要追踪的指令，记录其发送结果，等待异步确认
if sendResult.AcknowledgedAt != nil {
logRecord.AcknowledgedAt = sendResult.AcknowledgedAt
}
if sendResult.ReceivedSuccess != nil {
logRecord.ReceivedSuccess = *sendResult.ReceivedSuccess
}
logger.Infow("成功发送通用指令，并创建追踪日志", "networkID", networkID, "MessageID", sendResult.MessageID)
}
if err := g.deviceCommandLogRepo.Create(serviceCtx, logRecord); err != nil {
// 记录日志失败是一个需要关注的问题，但可能不应该中断主流程。
logger.Errorw("创建通用指令的日志失败", "MessageID", sendResult.MessageID, "error", err)
}
return nil
}

93
internal/domain/task/heartbeat_task.go Normal file
View File

@@ -0,0 +1,93 @@
package task
import (
"context"
"fmt"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/device"
"git.huangwc.com/pig/pig-farm-controller/internal/domain/plan"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/proto"
)
// HeartbeatTask 实现了 plan.Task 接口，用于执行一次区域主控心跳检测（发送Ping）
type HeartbeatTask struct {
ctx context.Context
log *models.TaskExecutionLog
areaControllerRepo repository.AreaControllerRepository
deviceService device.Service
}
// NewHeartbeatTask 创建一个心跳检测任务实例
func NewHeartbeatTask(
ctx context.Context,
log *models.TaskExecutionLog,
areaControllerRepo repository.AreaControllerRepository,
deviceService device.Service,
) plan.Task {
return &HeartbeatTask{
ctx: ctx,
log: log,
areaControllerRepo: areaControllerRepo,
deviceService: deviceService,
}
}
// Execute 是任务的核心执行逻辑
func (t *HeartbeatTask) Execute(ctx context.Context) error {
taskCtx, logger := logs.Trace(ctx, t.ctx, "Execute")
logger.Infow("开始执行区域主控心跳检测任务", "task_id", t.log.TaskID, "task_type", t.log.Task.Type, "log_id", t.log.ID)
controllers, err := t.areaControllerRepo.ListAll(taskCtx)
if err != nil {
return fmt.Errorf("心跳检测任务：获取所有区域主控失败: %w", err)
}
if len(controllers) == 0 {
logger.Infow("心跳检测任务：未发现任何区域主控，跳过本次检测")
return nil
}
// 构建 Ping 指令
pingInstruction := &proto.Instruction_Ping{
Ping: &proto.Ping{},
}
var firstError error
for _, controller := range controllers {
logger.Infow("向区域主控发送Ping指令", "controller_id", controller.ID)
err := t.deviceService.Send(taskCtx, controller.ID, pingInstruction, device.WithoutTracking())
if err != nil {
logger.Errorw("向区域主控发送Ping指令失败", "controller_id", controller.ID, "error", err)
if firstError == nil {
firstError = err // 保存第一个发生的错误
}
}
}
if firstError != nil {
return fmt.Errorf("心跳检测任务执行期间发生错误: %w", firstError)
}
logger.Infow("区域主控心跳检测任务执行完成", "task_id", t.log.TaskID, "task_type", t.log.Task.Type, "log_id", t.log.ID)
return nil
}
// OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑
func (t *HeartbeatTask) OnFailure(ctx context.Context, executeErr error) {
logger := logs.TraceLogger(ctx, t.ctx, "OnFailure")
logger.Errorw("区域主控心跳检测任务执行失败",
"task_id", t.log.TaskID,
"task_type", t.log.Task.Type,
"log_id", t.log.ID,
"error", executeErr,
)
}
// ResolveDeviceIDs 获取当前任务需要使用的设备ID列表
func (t *HeartbeatTask) ResolveDeviceIDs(ctx context.Context) ([]uint32, error) {
// 心跳检测任务不和任何特定设备绑定
return []uint32{}, nil
}

17
internal/domain/task/task.go
View File

@@ -18,14 +18,16 @@ const (
CompNameReleaseFeedWeight = "ReleaseFeedWeightTask"
CompNameFullCollectionTask = "FullCollectionTask"
CompNameAlarmNotification = "AlarmNotificationTask"
CompNameHeartbeatTask = "HeartbeatTask"
)
type taskFactory struct {
ctx context.Context
sensorDataRepo repository.SensorDataRepository
deviceRepo repository.DeviceRepository
alarmRepo repository.AlarmRepository
sensorDataRepo repository.SensorDataRepository
deviceRepo repository.DeviceRepository
alarmRepo repository.AlarmRepository
areaControllerRepo repository.AreaControllerRepository
deviceService device.Service
notificationService notify.Service
@@ -37,6 +39,7 @@ func NewTaskFactory(
sensorDataRepo repository.SensorDataRepository,
deviceRepo repository.DeviceRepository,
alarmRepo repository.AlarmRepository,
areaControllerRepo repository.AreaControllerRepository,
deviceService device.Service,
notifyService notify.Service,
alarmService alarm.AlarmService,
@@ -46,6 +49,7 @@ func NewTaskFactory(
sensorDataRepo: sensorDataRepo,
deviceRepo: deviceRepo,
alarmRepo: alarmRepo,
areaControllerRepo: areaControllerRepo,
deviceService: deviceService,
notificationService: notifyService,
alarmService: alarmService,
@@ -62,6 +66,8 @@ func (t *taskFactory) Production(ctx context.Context, claimedLog *models.TaskExe
return NewReleaseFeedWeightTask(logs.AddCompName(baseCtx, CompNameReleaseFeedWeight), claimedLog, t.sensorDataRepo, t.deviceRepo, t.deviceService)
case models.TaskTypeFullCollection:
return NewFullCollectionTask(logs.AddCompName(baseCtx, CompNameFullCollectionTask), claimedLog, t.deviceRepo, t.deviceService)
case models.TaskTypeHeartbeat:
return NewHeartbeatTask(logs.AddCompName(baseCtx, CompNameHeartbeatTask), claimedLog, t.areaControllerRepo, t.deviceService)
case models.TaskTypeAlarmNotification:
return NewAlarmNotificationTask(logs.AddCompName(baseCtx, CompNameAlarmNotification), claimedLog, t.notificationService, t.alarmRepo)
case models.TaskTypeDeviceThresholdCheck:
@@ -71,7 +77,6 @@ func (t *taskFactory) Production(ctx context.Context, claimedLog *models.TaskExe
case models.TaskTypeNotificationRefresh:
return NewRefreshNotificationTask(logs.AddCompName(baseCtx, "NotificationRefreshTask"), claimedLog, t.alarmService)
default:
// TODO 这里直接panic合适吗? 不过这个场景确实不该出现任何异常的任务类型
logger.Panicf("不支持的任务类型: %s", claimedLog.Task.Type)
panic("不支持的任务类型") // 显式panic防编译器报错
}
@@ -79,8 +84,6 @@ func (t *taskFactory) Production(ctx context.Context, claimedLog *models.TaskExe
// CreateTaskFromModel 实现了 TaskFactory 接口，用于从模型创建任务实例。
func (t *taskFactory) CreateTaskFromModel(ctx context.Context, taskModel *models.Task) (plan.TaskDeviceIDResolver, error) {
// 这个方法不关心 claimedLog 的其他字段，所以可以构造一个临时的
// 它只用于访问那些不依赖于执行日志的方法，比如 ResolveDeviceIDs
tempLog := &models.TaskExecutionLog{Task: *taskModel}
baseCtx := context.Background()
@@ -97,6 +100,8 @@ func (t *taskFactory) CreateTaskFromModel(ctx context.Context, taskModel *models
), nil
case models.TaskTypeFullCollection:
return NewFullCollectionTask(logs.AddCompName(baseCtx, CompNameFullCollectionTask), tempLog, t.deviceRepo, t.deviceService), nil
case models.TaskTypeHeartbeat:
return NewHeartbeatTask(logs.AddCompName(baseCtx, CompNameHeartbeatTask), tempLog, t.areaControllerRepo, t.deviceService), nil
case models.TaskTypeAlarmNotification:
return NewAlarmNotificationTask(logs.AddCompName(baseCtx, CompNameAlarmNotification), tempLog, t.notificationService, t.alarmRepo), nil
case models.TaskTypeDeviceThresholdCheck: