blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-10-25 20:52:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: config

Browse Source
This commit is contained in:
icey-yu 2025-01-04 15:53:51 +08:00
parent a78a1d7de4
commit d498fd90ca
17 changed files with 52 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -13,7 +13,7 @@ require (
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/mitchellh/mapstructure v1.5.0
github.com/openimsdk/protocol v0.0.72-alpha.69
github.com/openimsdk/tools v0.0.50-alpha.62
github.com/openimsdk/tools v0.0.50-alpha.64
github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.18.0
github.com/stretchr/testify v1.9.0

4
go.sum
View File

@ -349,8 +349,8 @@ github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrk
github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
github.com/openimsdk/protocol v0.0.72-alpha.69 h1:b22oY2XTdBR/BePqA73KsrM3GDF3Vk8YcBEXZU4ArJc=
github.com/openimsdk/protocol v0.0.72-alpha.69/go.mod h1:Iet+piS/jaS+kWWyj6EEr36mk4ISzIRYjoMSVA4dq2M=
github.com/openimsdk/tools v0.0.50-alpha.62 h1:e/m1XL7+EXbkOoxr/En/612WcOPKOUHPBj0++gG6MuQ=
github.com/openimsdk/tools v0.0.50-alpha.62/go.mod h1:JowL2jYr8tu4vcQe+5hJh4v3BtSx1T0CIS3pgU/Mw+U=
github.com/openimsdk/tools v0.0.50-alpha.64 h1:KmtE8V2K8atQJJg1xq2ySSrPQyf8ldwk8fw6jRnsxCw=
github.com/openimsdk/tools v0.0.50-alpha.64/go.mod h1:B+oqV0zdewN7OiEHYJm+hW+8/Te7B8tHHgD8rK5ZLZk=
github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=

4
internal/api/init.go
View File

@ -58,7 +58,9 @@ func Start(ctx context.Context, index int, config *Config) error {
config.RuntimeEnv = runtimeenv.PrintRuntimeEnvironment()
client, err := kdisc.NewDiscoveryRegister(&config.Discovery, config.RuntimeEnv)
client, err := kdisc.NewDiscoveryRegister(&config.Discovery, config.RuntimeEnv, []string{
config.Discovery.RpcService.MessageGateway,
})
if err != nil {
return errs.WrapMsg(err, "failed to register discovery service")
}

6
internal/msggateway/hub_server.go
View File

@ -16,9 +16,10 @@ package msggateway
import (
"context"
"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
"sync/atomic"
"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
"github.com/openimsdk/open-im-server/v3/pkg/common/startrpc"
@ -64,6 +65,9 @@ func (s *Server) Start(ctx context.Context, index int, conf *Config) error {
conf.WebhooksConfig.GetConfigFileName(),
conf.RedisConfig.GetConfigFileName(),
},
[]string{
conf.Discovery.RpcService.MessageGateway,
},
s.InitServer,
)
}

2
internal/msgtransfer/init.go
View File

@ -87,7 +87,7 @@ func Start(ctx context.Context, index int, config *Config) error {
if err != nil {
return err
}
client, err := discRegister.NewDiscoveryRegister(&config.Discovery, runTimeEnv)
client, err := discRegister.NewDiscoveryRegister(&config.Discovery, runTimeEnv, nil)
if err != nil {
return err
}

34
internal/tools/cron_task.go
View File

@ -2,11 +2,14 @@ package tools
import (
"context"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discovery"
disetcd "github.com/openimsdk/open-im-server/v3/pkg/common/discovery/etcd"
pbconversation "github.com/openimsdk/protocol/conversation"
"github.com/openimsdk/protocol/msg"
"github.com/openimsdk/protocol/third"
"github.com/openimsdk/tools/discovery/etcd"
"github.com/openimsdk/tools/mcontext"
"github.com/openimsdk/tools/mw"
@ -27,38 +30,47 @@ type CronTaskConfig struct {
runTimeEnv string
}
func Start(ctx context.Context, config *CronTaskConfig) error {
config.runTimeEnv = runtimeenv.PrintRuntimeEnvironment()
func Start(ctx context.Context, conf *CronTaskConfig) error {
conf.runTimeEnv = runtimeenv.PrintRuntimeEnvironment()
log.CInfo(ctx, "CRON-TASK server is initializing", "runTimeEnv", config.runTimeEnv, "chatRecordsClearTime", config.CronTask.CronExecuteTime, "msgDestructTime", config.CronTask.RetainChatRecords)
if config.CronTask.RetainChatRecords < 1 {
log.CInfo(ctx, "CRON-TASK server is initializing", "runTimeEnv", conf.runTimeEnv, "chatRecordsClearTime", conf.CronTask.CronExecuteTime, "msgDestructTime", conf.CronTask.RetainChatRecords)
if conf.CronTask.RetainChatRecords < 1 {
return errs.New("msg destruct time must be greater than 1").Wrap()
}
client, err := kdisc.NewDiscoveryRegister(&config.Discovery, config.runTimeEnv)
client, err := kdisc.NewDiscoveryRegister(&conf.Discovery, conf.runTimeEnv, nil)
if err != nil {
return errs.WrapMsg(err, "failed to register discovery service")
}
client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
ctx = mcontext.SetOpUserID(ctx, config.Share.IMAdminUserID[0])
ctx = mcontext.SetOpUserID(ctx, conf.Share.IMAdminUserID[0])
msgConn, err := client.GetConn(ctx, config.Discovery.RpcService.Msg)
msgConn, err := client.GetConn(ctx, conf.Discovery.RpcService.Msg)
if err != nil {
return err
}
thirdConn, err := client.GetConn(ctx, config.Discovery.RpcService.Third)
thirdConn, err := client.GetConn(ctx, conf.Discovery.RpcService.Third)
if err != nil {
return err
}
conversationConn, err := client.GetConn(ctx, config.Discovery.RpcService.Conversation)
conversationConn, err := client.GetConn(ctx, conf.Discovery.RpcService.Conversation)
if err != nil {
return err
}
if conf.Discovery.Enable == config.ETCD {
cm := disetcd.NewConfigManager(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient(), []string{
conf.CronTask.GetConfigFileName(),
conf.Share.GetConfigFileName(),
conf.Discovery.GetConfigFileName(),
})
cm.Watch(ctx)
}
srv := &cronServer{
ctx: ctx,
config: config,
config: conf,
cron: cron.New(),
msgClient: msg.NewMsgClient(msgConn),
conversationClient: pbconversation.NewConversationClient(conversationConn),
@ -74,7 +86,7 @@ func Start(ctx context.Context, config *CronTaskConfig) error {
if err := srv.registerClearUserMsg(); err != nil {
return err
}
log.ZDebug(ctx, "start cron task", "CronExecuteTime", config.CronTask.CronExecuteTime)
log.ZDebug(ctx, "start cron task", "CronExecuteTime", conf.CronTask.CronExecuteTime)
srv.cron.Start()
<-ctx.Done()
return nil

3
pkg/common/cmd/auth.go
View File

@ -64,5 +64,8 @@ func (a *AuthRpcCmd) runE() error {
a.authConfig.RedisConfig.GetConfigFileName(),
a.authConfig.Discovery.GetConfigFileName(),
},
[]string{
a.authConfig.Discovery.RpcService.MessageGateway,
},
auth.Start)
}

2
pkg/common/cmd/conversation.go
View File

@ -68,6 +68,6 @@ func (a *ConversationRpcCmd) runE() error {
a.conversationConfig.Share.GetConfigFileName(),
a.conversationConfig.LocalCacheConfig.GetConfigFileName(),
a.conversationConfig.Discovery.GetConfigFileName(),
},
}, nil,
conversation.Start)
}

2
pkg/common/cmd/friend.go
View File

@ -70,6 +70,6 @@ func (a *FriendRpcCmd) runE() error {
a.relationConfig.WebhooksConfig.GetConfigFileName(),
a.relationConfig.LocalCacheConfig.GetConfigFileName(),
a.relationConfig.Discovery.GetConfigFileName(),
},
}, nil,
relation.Start)
}

2
pkg/common/cmd/group.go
View File

@ -71,6 +71,6 @@ func (a *GroupRpcCmd) runE() error {
a.groupConfig.WebhooksConfig.GetConfigFileName(),
a.groupConfig.LocalCacheConfig.GetConfigFileName(),
a.groupConfig.Discovery.GetConfigFileName(),
},
}, nil,
group.Start, versionctx.EnableVersionCtx())
}

2
pkg/common/cmd/msg.go
View File

@ -72,6 +72,6 @@ func (a *MsgRpcCmd) runE() error {
a.msgConfig.WebhooksConfig.GetConfigFileName(),
a.msgConfig.LocalCacheConfig.GetConfigFileName(),
a.msgConfig.Discovery.GetConfigFileName(),
},
}, nil,
msg.Start)
}

3
pkg/common/cmd/push.go
View File

@ -72,5 +72,8 @@ func (a *PushRpcCmd) runE() error {
a.pushConfig.LocalCacheConfig.GetConfigFileName(),
a.pushConfig.Discovery.GetConfigFileName(),
},
[]string{
a.pushConfig.Discovery.RpcService.MessageGateway,
},
push.Start)
}

2
pkg/common/cmd/root.go
View File

@ -93,7 +93,7 @@ func (r *RootCmd) initEtcd() error {
return err
}
if disConfig.Enable == config.ETCD {
discov, _ := kdisc.NewDiscoveryRegister(&disConfig, env)
discov, _ := kdisc.NewDiscoveryRegister(&disConfig, env, nil)
r.etcdClient = discov.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
}
return nil

2
pkg/common/cmd/third.go
View File

@ -70,6 +70,6 @@ func (a *ThirdRpcCmd) runE() error {
a.thirdConfig.MinioConfig.GetConfigFileName(),
a.thirdConfig.LocalCacheConfig.GetConfigFileName(),
a.thirdConfig.Discovery.GetConfigFileName(),
},
}, nil,
third.Start)
}

2
pkg/common/cmd/user.go
View File

@ -72,6 +72,6 @@ func (a *UserRpcCmd) runE() error {
a.userConfig.WebhooksConfig.GetConfigFileName(),
a.userConfig.LocalCacheConfig.GetConfigFileName(),
a.userConfig.Discovery.GetConfigFileName(),
},
}, nil,
user.Start)
}

3
pkg/common/discovery/discoveryregister.go
View File

@ -28,7 +28,7 @@ import (
)
// NewDiscoveryRegister creates a new service discovery and registry client based on the provided environment type.
func NewDiscoveryRegister(discovery *config.Discovery, runtimeEnv string) (discovery.SvcDiscoveryRegistry, error) {
func NewDiscoveryRegister(discovery *config.Discovery, runtimeEnv string, watchNames []string) (discovery.SvcDiscoveryRegistry, error) {
if runtimeEnv == config.KUBERNETES {
return kubernetes.NewKubernetesConnManager(discovery.Kubernetes.Namespace,
grpc.WithDefaultCallOptions(
@ -42,6 +42,7 @@ func NewDiscoveryRegister(discovery *config.Discovery, runtimeEnv string) (disco
return etcd.NewSvcDiscoveryRegistry(
discovery.Etcd.RootDirectory,
discovery.Etcd.Address,
watchNames,
etcd.WithDialTimeout(10*time.Second),
etcd.WithMaxCallSendMsgSize(20*1024*1024),
etcd.WithUsernameAndPassword(discovery.Etcd.Username, discovery.Etcd.Password))

4
pkg/common/startrpc/start.go
View File

@ -49,7 +49,7 @@ import (
// Start rpc server.
func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T,
watchConfigNames []string,
watchConfigNames []string, watchServiceNames []string,
rpcFn func(ctx context.Context, config T, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error,
options ...grpc.ServerOption) error {
@ -95,7 +95,7 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
if autoSetPorts && discovery.Enable != conf.ETCD {
return errs.New("only etcd support autoSetPorts", "rpcRegisterName", rpcRegisterName).Wrap()
}
client, err := kdisc.NewDiscoveryRegister(discovery, runTimeEnv)
client, err := kdisc.NewDiscoveryRegister(discovery, runTimeEnv, watchServiceNames)
if err != nil {
return err
}