sparrow/services/scene-service/internal/service/manager/timer.go

package manager

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/gogf/gf/container/gmap"
	"github.com/gogf/gf/v2/os/gcron"
	"sparrow/pkg/server"
)

type TimerSceneConfig struct {
	SceneId    string                `json:"scene_id"`
	Conditions []*TimerTaskCondition `json:"conditions"`
	Actions    []*Action             `json:"actions"`
}

type TimerTaskCondition struct {
	TaskId string `json:"task_id"`
	Times  int    `json:"times"` // 执行次数 -1 表示无限次
	Cron   string `json:"cron"`
}

// TimerSceneService 定时场景实现
type TimerSceneService struct {
	task *gmap.HashMap
	cron *gcron.Cron
}

// NewTimerSceneService 创建定时场景
func NewTimerSceneService() *TimerSceneService {
	return &TimerSceneService{
		task: gmap.NewHashMap(true),
		cron: gcron.New(),
	}
}

func (t *TimerSceneService) Add(config string) error {
	var c TimerSceneConfig
	err := json.Unmarshal([]byte(config), &c)
	if err != nil {
		server.Log.Errorf("config to timerConfig error :%s", err.Error())
		return err
	}
	//if len(c.Conditions) == 0 || len(c.Actions) == 0 {
	//	server.Log.Errorf("timer scene-manager config error")
	//	return errors.New("timer scene-manager config error")
	//}
	if len(c.Conditions) == 0 || len(c.Actions) == 0 {
		return errors.New("timer scene-manager config error:conditions or actions is empty")
	}
	var addedTaskIds []string
	defer func() {
		// 如果添加失败，回滚已添加的Cron任务
		if err != nil && len(addedTaskIds) > 0 {
			for _, taskId := range addedTaskIds {
				t.cron.Remove(taskId)
			}
			server.Log.Errorf("add scene %s failed, rollback cron tasks: %v", c.SceneId, addedTaskIds)
		}
	}()
	for _, v := range c.Conditions {
		err = t.addTask(c.SceneId, v, c.Actions)
		if err != nil {
			server.Log.Errorf("add timerTask error:sceneId:%s, taskId: %s, error: %v", c.SceneId, v.TaskId, err)
			return err
		}
		addedTaskIds = append(addedTaskIds, v.TaskId)
	}
	t.task.Set(c.SceneId, c)
	server.Log.Debugf("AddTimeScene :%s", c.SceneId)
	return nil
}
func (t *TimerSceneService) addTask(sceneId string, c *TimerTaskCondition, actions []*Action) error {
	_, err := t.cron.AddTimes(context.Background(), c.Cron, c.Times, func(ctx context.Context) {
		taskExecutor := NewTaskExecutor(actions)
		if err := taskExecutor.Do(sceneId); err != nil {
			server.Log.Errorf("do task :%s error:%s", c.TaskId, err.Error())
		}
		err := taskExecutor.saveHis(sceneId, []string{c.TaskId})
		if err != nil {
			server.Log.Errorf("save task history error:sceneId:%s", sceneId)
		}
	}, c.TaskId)
	return err
}

func (t *TimerSceneService) Update(config string) error {
	var c TimerSceneConfig
	err := json.Unmarshal([]byte(config), &c)
	if err != nil {
		server.Log.Errorf("config to timerConfig error :%s", err.Error())
		return errors.New("timer scene-manager config error")
	}
	if t.task.Contains(c.SceneId) {
		oldTask := t.task.Get(c.SceneId).(TimerSceneConfig)
		for _, v := range oldTask.Conditions {
			t.cron.Remove(v.TaskId)
		}
	}

	var addedTaskIds []string
	defer func() {
		if err != nil && len(addedTaskIds) > 0 {
			for _, taskId := range addedTaskIds {
				t.cron.Remove(taskId)
			}
			server.Log.Errorf("update scene %s failed, rollback new cron tasks: %v", c.SceneId, addedTaskIds)
		}
	}()

	for _, v := range c.Conditions {
		err = t.addTask(c.SceneId, v, c.Actions)
		if err != nil {
			server.Log.Errorf("add timerTask error:sceneId:%s, taskId: %s, error: %v", c.SceneId, v.TaskId, err)
			return err
		}
		addedTaskIds = append(addedTaskIds, v.TaskId)
	}
	t.task.Set(c.SceneId, c)
	server.Log.Debugf("UpdateTimeScene :%s", config)
	return nil
}

func (t *TimerSceneService) Remove(id string) error {
	if !t.task.Contains(id) {
		return errors.New("场景不存在\n")
	}
	scene := t.task.Get(id).(TimerSceneConfig)
	for _, v := range scene.Conditions {
		t.cron.Remove(v.TaskId)
		fmt.Printf("remove taskId:%s\n", v.TaskId)
	}
	server.Log.Debugf("RemoveTimeScene :%s\n", scene.SceneId)
	t.task.Remove(id)
	return nil
}

func (t *TimerSceneService) Start(id string) error {
	if !t.task.Contains(id) {
		return errors.New("场景不存在\n")
	}
	scene := t.task.Get(id).(TimerSceneConfig)
	for _, v := range scene.Conditions {
		t.cron.Start(v.TaskId)
	}
	server.Log.Debugf("StartTimeScene :%s", scene.SceneId)
	return nil
}

func (t *TimerSceneService) Stop(id string) error {
	if !t.task.Contains(id) {
		return errors.New("场景不存在\n")
	}
	scene := t.task.Get(id).(TimerSceneConfig)
	for _, v := range scene.Conditions {
		t.cron.Stop(v.TaskId)
	}
	server.Log.Debugf("StopTimeScene :%s", scene.SceneId)
	return nil
}