mirror of
https://github.com/openimsdk/open-im-server.git
synced 2025-11-05 21:02:11 +08:00
refactor: crontab update
This commit is contained in:
parent
9b3aedc795
commit
93b68f070c
@ -1,3 +1,4 @@
|
|||||||
chatRecordsClearTime: "0 2 * * 3"
|
chatRecordsClearTime: "0 2 * * 3"
|
||||||
msgDestructTime: "0 2 * * *"
|
msgDestructTime: "0 2 * * *"
|
||||||
retainChatRecords: 365
|
retainChatRecords: 365
|
||||||
|
enableCronLocker: false
|
||||||
|
|||||||
@ -16,22 +16,23 @@ package tools
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"github.com/openimsdk/open-im-server/v3/pkg/common/cmd"
|
||||||
"github.com/openimsdk/tools/db/redisutil"
|
"github.com/openimsdk/tools/db/redisutil"
|
||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
|
||||||
"github.com/openimsdk/tools/errs"
|
"github.com/openimsdk/tools/errs"
|
||||||
"github.com/openimsdk/tools/log"
|
"github.com/openimsdk/tools/log"
|
||||||
"github.com/redis/go-redis/v9"
|
"github.com/redis/go-redis/v9"
|
||||||
"github.com/robfig/cron/v3"
|
"github.com/robfig/cron/v3"
|
||||||
)
|
)
|
||||||
|
|
||||||
func StartTask(ctx context.Context, config *config.GlobalConfig) error {
|
func Start(ctx context.Context, config *cmd.CronTaskConfig) error {
|
||||||
|
|
||||||
log.CInfo(ctx, "CRON-TASK server is initializing", "chatRecordsClearTime", config.ChatRecordsClearTime, "msgDestructTime", config.MsgDestructTime)
|
log.CInfo(ctx, "CRON-TASK server is initializing", "chatRecordsClearTime",
|
||||||
|
config.CronTask.ChatRecordsClearTime, "msgDestructTime", config.CronTask.MsgDestructTime)
|
||||||
|
|
||||||
msgTool, err := InitMsgTool(ctx, config)
|
msgTool, err := InitMsgTool(ctx, config)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -40,7 +41,7 @@ func StartTask(ctx context.Context, config *config.GlobalConfig) error {
|
|||||||
|
|
||||||
msgTool.convertTools()
|
msgTool.convertTools()
|
||||||
|
|
||||||
rdb, err := redisutil.NewRedisClient(ctx, config.Redis.Build())
|
rdb, err := redisutil.NewRedisClient(ctx, config.RedisConfig.Build())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -48,12 +49,14 @@ func StartTask(ctx context.Context, config *config.GlobalConfig) error {
|
|||||||
// register cron tasks
|
// register cron tasks
|
||||||
var crontab = cron.New()
|
var crontab = cron.New()
|
||||||
|
|
||||||
_, err = crontab.AddFunc(config.ChatRecordsClearTime, cronWrapFunc(config, rdb, "cron_clear_msg_and_fix_seq", msgTool.AllConversationClearMsgAndFixSeq))
|
_, err = crontab.AddFunc(config.CronTask.ChatRecordsClearTime,
|
||||||
|
cronWrapFunc(config, rdb, "cron_clear_msg_and_fix_seq", msgTool.AllConversationClearMsgAndFixSeq))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return errs.Wrap(err)
|
return errs.Wrap(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
_, err = crontab.AddFunc(config.MsgDestructTime, cronWrapFunc(config, rdb, "cron_conversations_destruct_msgs", msgTool.ConversationsDestructMsgs))
|
_, err = crontab.AddFunc(config.CronTask.MsgDestructTime,
|
||||||
|
cronWrapFunc(config, rdb, "cron_conversations_destruct_msgs", msgTool.ConversationsDestructMsgs))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return errs.WrapMsg(err, "cron_conversations_destruct_msgs")
|
return errs.WrapMsg(err, "cron_conversations_destruct_msgs")
|
||||||
}
|
}
|
||||||
@ -91,8 +94,8 @@ func netlock(rdb redis.UniversalClient, key string, ttl time.Duration) bool {
|
|||||||
return ok
|
return ok
|
||||||
}
|
}
|
||||||
|
|
||||||
func cronWrapFunc(config *config.GlobalConfig, rdb redis.UniversalClient, key string, fn func()) func() {
|
func cronWrapFunc(config *cmd.CronTaskConfig, rdb redis.UniversalClient, key string, fn func()) func() {
|
||||||
enableCronLocker := config.EnableCronLocker
|
enableCronLocker := config.CronTask.EnableCronLocker
|
||||||
return func() {
|
return func() {
|
||||||
// if don't enable cron-locker, call fn directly.
|
// if don't enable cron-locker, call fn directly.
|
||||||
if !enableCronLocker {
|
if !enableCronLocker {
|
||||||
|
|||||||
@ -64,6 +64,7 @@ const (
|
|||||||
kafkaEnvPrefix = "openim-kafka"
|
kafkaEnvPrefix = "openim-kafka"
|
||||||
zoopkeeperEnvPrefix = "openim-zookeeper"
|
zoopkeeperEnvPrefix = "openim-zookeeper"
|
||||||
apiEnvPrefix = "openim-api"
|
apiEnvPrefix = "openim-api"
|
||||||
|
cornTaskEnvPrefix = "openim-crontask"
|
||||||
authEnvPrefix = "openim-auth"
|
authEnvPrefix = "openim-auth"
|
||||||
conversationEnvPrefix = "openim-conversation"
|
conversationEnvPrefix = "openim-conversation"
|
||||||
friendEnvPrefix = "openim-friend"
|
friendEnvPrefix = "openim-friend"
|
||||||
|
|||||||
@ -16,7 +16,6 @@ package cmd
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
|
||||||
"github.com/openimsdk/open-im-server/v3/internal/tools"
|
"github.com/openimsdk/open-im-server/v3/internal/tools"
|
||||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||||
"github.com/openimsdk/tools/system/program"
|
"github.com/openimsdk/tools/system/program"
|
||||||
@ -25,29 +24,34 @@ import (
|
|||||||
|
|
||||||
type CronTaskCmd struct {
|
type CronTaskCmd struct {
|
||||||
*RootCmd
|
*RootCmd
|
||||||
initFunc func(ctx context.Context, config *config.GlobalConfig) error
|
ctx context.Context
|
||||||
ctx context.Context
|
configMap map[string]StructEnvPrefix
|
||||||
|
cronTaskConfig CronTaskConfig
|
||||||
|
}
|
||||||
|
type CronTaskConfig struct {
|
||||||
|
CronTask config.CronTask
|
||||||
|
RedisConfig config.Redis
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewCronTaskCmd(name string) *CronTaskCmd {
|
func NewCronTaskCmd() *CronTaskCmd {
|
||||||
ret := &CronTaskCmd{RootCmd: NewRootCmd(program.GetProcessName(), name, WithCronTaskLogName()),
|
var cronTaskConfig CronTaskConfig
|
||||||
initFunc: tools.StartTask}
|
ret := &CronTaskCmd{cronTaskConfig: cronTaskConfig}
|
||||||
|
ret.configMap = map[string]StructEnvPrefix{
|
||||||
|
OpenIMCronTaskCfgFileName: {EnvPrefix: cornTaskEnvPrefix, ConfigStruct: &cronTaskConfig.CronTask},
|
||||||
|
RedisConfigFileName: {EnvPrefix: redisEnvPrefix, ConfigStruct: &cronTaskConfig.RedisConfig},
|
||||||
|
}
|
||||||
|
ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap))
|
||||||
ret.ctx = context.WithValue(context.Background(), "version", config.Version)
|
ret.ctx = context.WithValue(context.Background(), "version", config.Version)
|
||||||
ret.addRunE()
|
ret.Command.PreRunE = func(cmd *cobra.Command, args []string) error {
|
||||||
ret.SetRootCmdPt(ret)
|
return ret.preRunE()
|
||||||
|
}
|
||||||
return ret
|
return ret
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *CronTaskCmd) addRunE() {
|
func (a *CronTaskCmd) Exec() error {
|
||||||
c.Command.RunE = func(cmd *cobra.Command, args []string) error {
|
return a.Execute()
|
||||||
return c.initFunc(c.ctx, c.config)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *CronTaskCmd) Exec() error {
|
func (a *CronTaskCmd) preRunE() error {
|
||||||
return c.Execute()
|
return tools.Start(a.ctx, &a.cronTaskConfig)
|
||||||
}
|
|
||||||
|
|
||||||
func (c *CronTaskCmd) GetPortFromConfig(portType string) int {
|
|
||||||
return 0
|
|
||||||
}
|
}
|
||||||
|
|||||||
@ -95,6 +95,7 @@ type CronTask struct {
|
|||||||
ChatRecordsClearTime string `mapstructure:"chatRecordsClearTime"`
|
ChatRecordsClearTime string `mapstructure:"chatRecordsClearTime"`
|
||||||
MsgDestructTime string `mapstructure:"msgDestructTime"`
|
MsgDestructTime string `mapstructure:"msgDestructTime"`
|
||||||
RetainChatRecords int `mapstructure:"retainChatRecords"`
|
RetainChatRecords int `mapstructure:"retainChatRecords"`
|
||||||
|
EnableCronLocker bool `yaml:"enableCronLocker"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type OfflinePushConfig struct {
|
type OfflinePushConfig struct {
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user