94 lines
3.1 KiB
Go
94 lines
3.1 KiB
Go
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
|
||
}
|