4. 实现升级任务创建与下发

任务目标

本案例实现OTA升级后台的核心入口：创建一条升级任务，绑定目标设备组与固件版本，并将升级指令可靠地下发到组内设备。任务状态贯穿“待下发→下发中→已完成”，为后续的进度跟踪与失败重试提供基础。

核心实现

// upgrade_task.go
package main

import (
	"errors"
	"fmt"
	"sync"
	"time"
)

// Firmware 固件版本信息
type Firmware struct {
	Version string
	URL     string
	MD5     string
}

// Device 设备信息
type Device struct {
	ID       string
	GroupID  string
	Online   bool
	Firmware string // 当前固件版本
}

// UpgradeTask 升级任务
type UpgradeTask struct {
	ID         string
	GroupID    string
	Firmware   Firmware
	Status     string // pending / delivering / completed / failed
	TargetIDs  []string
	Delivered  map[string]bool
	mu         sync.Mutex
}

// UpgradeManager 管理升级任务
type UpgradeManager struct {
	devices map[string]*Device // deviceID -> Device
	tasks   map[string]*UpgradeTask
	mu      sync.Mutex
}

func NewUpgradeManager() *UpgradeManager {
	return &UpgradeManager{
		devices: make(map[string]*Device),
		tasks:   make(map[string]*UpgradeTask),
	}
}

// RegisterDevice 注册设备（模拟设备接入）
func (m *UpgradeManager) RegisterDevice(d *Device) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.devices[d.ID] = d
}

// CreateTask 创建升级任务：指定设备组与固件版本
func (m *UpgradeManager) CreateTask(taskID, groupID string, fw Firmware) (*UpgradeTask, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// 校验固件信息
	if fw.Version == "" || fw.URL == "" {
		return nil, errors.New("firmware version and URL are required")
	}

	// 收集该组在线设备
	var targets []string
	for _, dev := range m.devices {
		if dev.GroupID == groupID && dev.Online {
			targets = append(targets, dev.ID)
		}
	}
	if len(targets) == 0 {
		return nil, fmt.Errorf("no online devices in group %s", groupID)
	}

	task := &UpgradeTask{
		ID:        taskID,
		GroupID:   groupID,
		Firmware:  fw,
		Status:    "pending",
		TargetIDs: targets,
		Delivered: make(map[string]bool),
	}
	m.tasks[taskID] = task
	return task, nil
}

// DeliverTask 下发升级指令到目标设备（模拟异步下发）
func (m *UpgradeManager) DeliverTask(taskID string) error {
	m.mu.Lock()
	task, ok := m.tasks[taskID]
	m.mu.Unlock()
	if !ok {
		return errors.New("task not found")
	}

	task.mu.Lock()
	defer task.mu.Unlock()
	if task.Status != "pending" {
		return fmt.Errorf("task already in %s status", task.Status)
	}

	task.Status = "delivering"
	for _, devID := range task.TargetIDs {
		// 模拟下发：实际场景中此处应通过MQTT/CoAP推送指令
		// 并等待设备上报确认，此处简化为直接标记成功
		task.Delivered[devID] = true
	}
	task.Status = "completed"
	return nil
}

// TaskStatus 查询任务状态
func (m *UpgradeManager) TaskStatus(taskID string) (string, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	task, ok := m.tasks[taskID]
	if !ok {
		return "", errors.New("task not found")
	}
	return task.Status, nil
}

// ---- 最小可运行示例 ----
func main() {
	mgr := NewUpgradeManager()

	// 注册设备：2台在线，1台离线
	mgr.RegisterDevice(&Device{ID: "dev-001", GroupID: "grp-alpha", Online: true, Firmware: "v1.0"})
	mgr.RegisterDevice(&Device{ID: "dev-002", GroupID: "grp-alpha", Online: true, Firmware: "v1.0"})
	mgr.RegisterDevice(&Device{ID: "dev-003", GroupID: "grp-alpha", Online: false, Firmware: "v1.0"})

	// 创建任务：目标grp-alpha组，升级到v2.0
	fw := Firmware{Version: "v2.0", URL: "https://ota.example.com/fw/v2.0.bin", MD5: "abc123"}
	task, err := mgr.CreateTask("task-001", "grp-alpha", fw)
	if err != nil {
		fmt.Println("create failed:", err)
		return
	}
	fmt.Printf("task created, targets: %v\n", task.TargetIDs)

	// 下发升级指令
	if err := mgr.DeliverTask("task-001"); err != nil {
		fmt.Println("deliver failed:", err)
		return
	}

	// 验证结果
	status, _ := mgr.TaskStatus("task-001")
	fmt.Printf("task status: %s\n", status)
	fmt.Printf("delivered count: %d/%d\n", len(task.Delivered), len(task.TargetIDs))
}

运行与验证

1. 保存代码为 upgrade_task.go，执行 go run upgrade_task.go。
2. 预期输出（确定性结果）：

task created, targets: [dev-001 dev-002]
task status: completed
delivered count: 2/2

3. 判定通过条件：dev-003因离线未被纳入目标；任务状态从 pending 转为 completed，且下发数量与在线设备数一致。

关键说明

• 设备组过滤：创建任务时仅选择目标组中 Online=true 的设备，离线设备不参与本次升级。
• 下发机制：示例中 DeliverTask 直接标记成功，用于演示任务状态流转。真实工程中此处应替换为MQTT/CoAP指令推送，并等待设备上报升级结果后更新 Delivered 状态。
• 并发安全：任务与设备均使用 sync.Mutex 保护，避免并发访问冲突。
• 扩展点：如需支持失败重试，可在 Delivered 中记录失败原因，并增加重试计数逻辑。