@@ -3,11 +3,13 @@ package processor
33import (
44 "context"
55 "fmt"
6+ "time"
67
8+ "github.com/disgoorg/log"
9+ "github.com/go-co-op/gocron"
710 "github.com/nezuchan/scheduled-tasks/constants"
811 "github.com/rabbitmq/amqp091-go"
912 "github.com/redis/go-redis/v9"
10- "github.com/robfig/cron"
1113)
1214
1315func ProcessCronJob (client redis.UniversalClient , broker amqp091.Channel , name string , taskId string ) {
@@ -17,36 +19,40 @@ func ProcessCronJob(client redis.UniversalClient, broker amqp091.Channel, name s
1719 if CronValue == 1 && TaskKey == 1 {
1820 Cron := client .Get (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_CRON_VALUE , name )).Val ()
1921
20- c := cron .New ()
21-
22- c .AddFunc (Cron , func () {
23- Value := client .Get (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_KEY_ROUTE , taskId )).Val ()
24-
25- if Value == "" {
26- c .Stop ()
27- client .Unlink (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_KEY_ROUTE , taskId ))
28- client .Unlink (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_CRON_VALUE , name ))
29- client .SRem (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_CRON_SETS , fmt .Sprintf ("%s:%s" , taskId , name ))).Err ()
30- return
31- }
32-
33- Route := client .Get (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_KEY_ROUTE , taskId )).Val ()
34-
35- if Route != "" {
36- broker .PublishWithContext (context .Background (), constants .TASKER_EXCHANGE , Route , false , false , amqp091.Publishing {
37- ContentType : "text/plain" ,
38- Body : []byte (Value ),
39- })
40- } else {
41- broker .PublishWithContext (context .Background (), constants .TASKER_EXCHANGE , "*" , false , false , amqp091.Publishing {
42- ContentType : "text/plain" ,
43- Body : []byte (Value ),
44- })
45- }
46- })
47-
48- c .Start ()
49-
22+ c := gocron .NewScheduler (time .UTC )
23+
24+ c .Cron (Cron ).Do (
25+ func (){
26+ log .Infof ("Sending cron job %s to client" , name )
27+ Value := client .Get (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_KEY_VALUE , taskId )).Val ()
28+
29+ if Value == "" {
30+ c .Stop ()
31+ client .Unlink (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_KEY_VALUE , taskId ))
32+ client .Unlink (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_KEY_ROUTE , taskId ))
33+ client .Unlink (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_CRON_VALUE , name ))
34+ client .SRem (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_CRON_SETS , fmt .Sprintf ("%s:%s" , taskId , name ))).Err ()
35+ c .Clear ()
36+ return
37+ }
38+
39+ Route := client .Get (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_KEY_ROUTE , taskId )).Val ()
40+
41+ if Route != "" {
42+ broker .PublishWithContext (context .Background (), constants .TASKER_EXCHANGE , Route , false , false , amqp091.Publishing {
43+ ContentType : "text/plain" ,
44+ Body : []byte (Value ),
45+ })
46+ } else {
47+ broker .PublishWithContext (context .Background (), constants .TASKER_EXCHANGE , "*" , false , false , amqp091.Publishing {
48+ ContentType : "text/plain" ,
49+ Body : []byte (Value ),
50+ })
51+ }
52+ },
53+ )
54+
55+ c .StartAsync ()
5056 } else {
5157 client .Unlink (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_KEY_ROUTE , taskId ))
5258 client .Unlink (context .Background (), fmt .Sprintf ("%s:%s" , constants .TASK_REDIS_CRON_VALUE , name ))
0 commit comments