-
Notifications
You must be signed in to change notification settings - Fork 0
/
sub_nats.go
102 lines (99 loc) · 3.02 KB
/
sub_nats.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
package OrgTime
import (
"fmt"
BaseService "github.com/fotomxq/weeekj_core/v5/base/service"
BaseSystemMission "github.com/fotomxq/weeekj_core/v5/base/system_mission"
CoreFilter "github.com/fotomxq/weeekj_core/v5/core/filter"
CoreLog "github.com/fotomxq/weeekj_core/v5/core/log"
CoreSQL "github.com/fotomxq/weeekj_core/v5/core/sql"
Router2SystemConfig "github.com/fotomxq/weeekj_core/v5/router2/system_config"
)
func subNats() {
//任务调度,更新考勤情况
BaseSystemMission.ReginSub(&runUpdateSysM, subNatsRunUpdate)
//注册服务
_ = BaseService.SetService(&BaseService.ArgsSetService{
ExpireAt: CoreFilter.GetNowTimeCarbon().AddDay().Time,
Name: "组织考勤变更",
Description: "",
EventSubType: "all",
Code: "org_time_update",
EventType: "nats",
EventURL: "/org/time/update",
//TODO:待补充
EventParams: "",
})
}
func subNatsRunUpdate() {
//日志
logAppend := "org time run, "
//捕捉异常
defer func() {
//进度控制
runUpdateSysM.Bind.UpdateNextAtFutureSec(1)
//跳出处理
if r := recover(); r != nil {
runUpdateSysM.Update(fmt.Sprint("发生错误: ", r), "run.error", 0)
CoreLog.Error(logAppend, r)
}
}()
//确保数据全部在内存内
loadAllConfigToMem()
//锁定机制
runUpdateCacheLock.Lock()
defer runUpdateCacheLock.Unlock()
//跟踪器
runUpdateSysM.Start("开始更新", "start", int64(len(allConfigList)))
//遍历数据处理
for k, v := range allConfigList {
//跟踪器
runUpdateSysM.Update(fmt.Sprint("正在处理", v.Name), fmt.Sprint("config:", v.ID), int64(k))
//检查是否上班
checkIsWork := checkIsWorkByData(&v)
if checkIsWork == v.IsWork {
continue
}
//切换轮动机制
if len(v.RotConfig.WorkTime) > 0 && checkIsWork == false {
//更换轮动次序
v.RotConfig.NowKey += 1
if v.RotConfig.NowKey < 0 {
v.RotConfig.NowKey = 0
}
if v.RotConfig.NowKey >= len(v.RotConfig.WorkTime) {
v.RotConfig.NowKey = 0
}
//标记切换上下班状态
if _, err := CoreSQL.UpdateOne(Router2SystemConfig.MainDB.DB, "UPDATE org_work_time SET is_work = :is_work, rot_config = :rot_config WHERE id = :id", map[string]interface{}{
"id": v.ID,
"is_work": checkIsWork,
"rot_config": v.RotConfig,
}); err != nil {
CoreLog.Error(logAppend, "update failed, ", err)
continue
} else {
//修改内存数据
allConfigList[k].IsWork = checkIsWork
allConfigList[k].RotConfig = v.RotConfig
}
} else {
//标记切换上下班状态
if _, err := CoreSQL.UpdateOne(Router2SystemConfig.MainDB.DB, "UPDATE org_work_time SET is_work = :is_work WHERE id = :id", map[string]interface{}{
"id": v.ID,
"is_work": checkIsWork,
}); err != nil {
CoreLog.Error(logAppend, "update failed, ", err)
continue
} else {
//修改内存数据
allConfigList[k].IsWork = checkIsWork
}
}
//推送nats通知
pushNatsWork(v.OrgID, v.Binds, v.Groups, checkIsWork)
//删除缓冲
deleteConfigCache(v.ID)
}
//收尾处理
runUpdateSysM.Finish()
}