feature(global): cron task

go.mod

@@ -3,6 +3,7 @@ module github.com/chenjianhao66/go-GB28181
 go 1.19
 
 require (
+	github.com/agiledragon/gomonkey v2.0.2+incompatible
 	github.com/beevik/etree v1.1.0
 	github.com/ghettovoice/gosip v0.0.0-20221216110459-a49cda0b8a0f
 	github.com/gin-gonic/gin v1.9.0
@@ -12,6 +13,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/redis/go-redis/v9 v9.0.2
 	github.com/sirupsen/logrus v1.9.0
+	github.com/smartystreets/goconvey v1.7.2
 	github.com/spf13/cast v1.5.0
 	github.com/spf13/viper v1.14.0
 	github.com/swaggo/files v1.0.1
@@ -47,11 +49,13 @@ require (
 	github.com/gobwas/pool v0.2.1 // indirect
 	github.com/gobwas/ws v1.1.0-rc.1 // indirect
 	github.com/goccy/go-json v0.10.0 // indirect
+	github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/jtolds/gls v4.20.0+incompatible // indirect
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
 	github.com/magiconair/properties v1.8.6 // indirect
@@ -67,7 +71,7 @@ require (
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.6 // indirect
 	github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b // indirect
-	github.com/smartystreets/goconvey v1.7.2 // indirect
+	github.com/smartystreets/assertions v1.2.0 // indirect
 	github.com/spf13/afero v1.9.2 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect

go.sum

@@ -42,6 +42,8 @@ github.com/BurntSushi/toml v1.2.1/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbi
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc=
 github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE=
+github.com/agiledragon/gomonkey v2.0.2+incompatible h1:eXKi9/piiC3cjJD1658mEE2o3NjkJ5vDLgYjCQu0Xlw=
+github.com/agiledragon/gomonkey v2.0.2+incompatible/go.mod h1:2NGfXu1a80LLr2cmWXGBDaHEjb1idR6+FVlX5T3D9hw=
 github.com/beevik/etree v1.1.0 h1:T0xke/WvNtMoCqgzPhkX2r4rjY3GDZFi+FjpRZY2Jbs=
 github.com/beevik/etree v1.1.0/go.mod h1:r8Aw8JqVegEf0w2fDnATrX9VpkMcyFeM0FhwO62wh+A=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=

internal/config/options.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"fmt"
 	"github.com/spf13/viper"
 )
 
@@ -37,6 +38,6 @@ func loadConfig() {
 	viper.SetConfigType("yml")
 	viper.AutomaticEnv()
 	if err := viper.ReadInConfig(); err != nil {
-		panic("load config fail,please check your config file whether in config/ in the directory")
+		fmt.Print("load config fail,please check your config file whether in config/ in the directory")
 	}
 }

internal/gb/notify.go

@@ -4,6 +4,7 @@ import (
 	"encoding/xml"
 	"github.com/chenjianhao66/go-GB28181/internal/log"
 	"github.com/chenjianhao66/go-GB28181/internal/parser"
+	"github.com/chenjianhao66/go-GB28181/internal/pkg/cron"
 	"github.com/ghettovoice/gosip/sip"
 	"net/http"
 )
@@ -38,5 +39,9 @@ func keepaliveHandler(req sip.Request, tx sip.ServerTransaction) {
 	if err := storage.deviceKeepalive(device.ID); err != nil {
 		log.Debugf("{%d,%s}更新心跳失败：%s", device.ID, device.DeviceId, err)
 	}
+	if err := cron.ResetTime(device.DeviceId, cron.TaskKeepLive); err != nil {
+		log.Errorf("{%d,%s}更新心跳失败：%s", device.ID, device.DeviceId, err)
+	}
+
 	_ = tx.Respond(sip.NewResponseFromRequest("", req, http.StatusOK, http.StatusText(http.StatusOK), ""))
 }

internal/gb/register.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/chenjianhao66/go-GB28181/internal/log"
 	"github.com/chenjianhao66/go-GB28181/internal/parser"
+	"github.com/chenjianhao66/go-GB28181/internal/pkg/cron"
 	"github.com/ghettovoice/gosip/sip"
 	"net/http"
 )
@@ -51,6 +52,9 @@ func RegisterHandler(req sip.Request, tx sip.ServerTransaction) {
 		if offlineFlag {
 			// 注销请求
 			_ = storage.deviceOffline(device)
+			if err := cron.StopTask(device.DeviceId, cron.TaskKeepLive); err != nil {
+				log.Errorf("停止心跳检测任务失败: %s", device.DeviceId)
+			}
 		} else {
 			// 注册请求
 			if err := storage.deviceOnline(device); err != nil {

internal/gb/storage.go

@@ -3,6 +3,7 @@ package gb
 import (
 	"github.com/chenjianhao66/go-GB28181/internal/log"
 	"github.com/chenjianhao66/go-GB28181/internal/model"
+	"github.com/chenjianhao66/go-GB28181/internal/pkg/cron"
 	st "github.com/chenjianhao66/go-GB28181/internal/storage"
 	"github.com/chenjianhao66/go-GB28181/internal/storage/mysql"
 	"time"
@@ -46,6 +47,12 @@ func (d *data) deviceOnline(device model.Device) error {
 		device.Offline = 1
 		err = d.s.Devices().Update(device)
 	}
+
+	err = cron.StartTask(device.DeviceId, cron.TaskKeepLive, 10*time.Second, func() {
+		device.Offline = 0
+		d.s.Devices().Save(device)
+	})
+
 	return err
 }
 

internal/pkg/cron/keeplive.go

@@ -0,0 +1,37 @@
+package cron
+
+import (
+	"github.com/chenjianhao66/go-GB28181/internal/log"
+	"time"
+)
+
+type keepLiveTask struct {
+	timer    *time.Ticker
+	deviceId string
+	duration time.Duration
+	runFunc  runFunc
+}
+
+func (k *keepLiveTask) start() error {
+	k.timer = time.NewTicker(k.duration)
+	go k.watch()
+
+	return nil
+}
+
+func (k *keepLiveTask) cancel() error {
+	k.timer.Stop()
+	return nil
+}
+
+func (k *keepLiveTask) refresh() {
+	k.timer.Reset(k.duration)
+}
+
+func (k *keepLiveTask) watch() {
+	<-k.timer.C
+	log.Warnf("设备离线！ 设备号: %s, 时间: %s", k.deviceId, time.Now().String())
+	k.runFunc()
+	taskList.deleteOneTask(k.deviceId, TaskKeepLive)
+	return
+}

internal/pkg/cron/task.go

@@ -0,0 +1,148 @@
+package cron
+
+import (
+	"fmt"
+	"github.com/chenjianhao66/go-GB28181/internal/log"
+	"github.com/pkg/errors"
+	"sync"
+	"time"
+)
+
+type TaskType string
+type runFunc func()
+
+type task interface {
+	cancel() error
+	start() error
+	refresh()
+	watch()
+}
+
+const (
+	TaskKeepLive TaskType = "KeepLive"
+)
+
+type taskSchedule map[string]map[TaskType]task
+
+// key: deviceId, value: {key: taskType, value: task Object}
+var taskList taskSchedule = make(map[string]map[TaskType]task)
+
+var once sync.Once
+
+func (t taskSchedule) deleteOneTask(deviceId string, taskType TaskType) {
+	delete(t[deviceId], taskType)
+}
+
+func (t taskSchedule) clearAllTasksForOneDevice(deviceId string) {
+	t[deviceId] = nil
+}
+
+func (t taskSchedule) deleteOneDeviceRecord(deviceId string) {
+	delete(t, deviceId)
+}
+
+func (t taskSchedule) getOneTask(deviceId string, taskType TaskType) task {
+	if t[deviceId] != nil {
+		return t[deviceId][taskType]
+	}
+	return nil
+}
+
+func (t taskSchedule) getAllTasksForOneDevice(deviceId string) map[TaskType]task {
+	return t[deviceId]
+}
+
+func (t taskSchedule) putOneTask(deviceId string, taskType TaskType, oneTask task) {
+	if t[deviceId] != nil && t[deviceId][taskType] == nil {
+		t[deviceId][taskType] = oneTask
+	}
+}
+
+func (t taskSchedule) initTaskList(deviceId string) {
+	if taskList.getAllTasksForOneDevice(deviceId) == nil {
+		taskList[deviceId] = make(map[TaskType]task)
+	}
+}
+
+func StopTask(deviceId string, taskType TaskType) error {
+	t, err := getTask(deviceId, taskType)
+
+	if err != nil {
+		return err
+	}
+
+	taskList.deleteOneTask(deviceId, taskType)
+
+	return t.cancel()
+}
+
+func StartTask(deviceId string, taskType TaskType, duration time.Duration, runFunc runFunc) error {
+	if taskList.getOneTask(deviceId, taskType) != nil {
+		log.Errorf("任务 %+v 已经存在！", taskType)
+		return errors.New(fmt.Sprintf("开启任务，任务类型: %+v, 设备ID: %+v", taskType, deviceId))
+	}
+
+	t, err := createTask(deviceId, taskType, duration, runFunc)
+
+	if err != nil {
+		return err
+	}
+
+	return t.start()
+}
+
+func ResetTime(deviceId string, taskType TaskType) error {
+	t, err := getTask(deviceId, taskType)
+
+	if err != nil {
+		return err
+	}
+
+	t.refresh()
+
+	return nil
+}
+
+func getTask(deviceId string, taskType TaskType) (task, error) {
+	if taskList.getAllTasksForOneDevice(deviceId) == nil {
+		log.Errorf("任务 %+v 设备ID: %+v 不存在!", taskType, deviceId)
+		return nil, errors.New(fmt.Sprintf("停止任务，任务类型: %+v, 设备ID: %+v", taskType, deviceId))
+	}
+
+	if taskList.getOneTask(deviceId, taskType) == nil {
+		log.Errorf("任务 %+v 设备ID: %+v 不存在!", taskType, deviceId)
+		return nil, errors.New(fmt.Sprintf("停止任务，任务类型: %+v, 设备ID: %+v", taskType, deviceId))
+	}
+
+	t := taskList.getOneTask(deviceId, taskType)
+
+	return t, nil
+}
+
+func createTask(deviceId string, taskType TaskType, duration time.Duration, runFunc runFunc) (task, error) {
+	if taskList.getOneTask(deviceId, taskType) != nil {
+		return nil, errors.New(fmt.Sprintf("获取任务对象失败, 任务类型: %+v", taskType))
+	}
+
+	taskList.initTaskList(deviceId)
+
+	var t task
+
+	switch taskType {
+
+	case TaskKeepLive:
+		t = &keepLiveTask{
+			deviceId: deviceId,
+			duration: duration,
+			runFunc:  runFunc,
+		}
+		taskList.putOneTask(deviceId, taskType, t)
+
+	default:
+		log.Errorf("不支持的任务类型：", taskType)
+		return nil, errors.New(fmt.Sprintf("不支持的任务类型: %+v, 设备ID: %+v",
+			taskType, deviceId))
+	}
+
+	return t, nil
+}