blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-06 04:15:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
withchao 2025-02-08 15:07:45 +08:00
parent 585e56783b
commit c48a082602
10 changed files with 49 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/api/init.go
View File

@ -27,19 +27,16 @@ import (
"time"
conf "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"
"github.com/openimsdk/tools/discovery"
"github.com/openimsdk/tools/discovery/etcd"
"github.com/openimsdk/tools/errs"
"github.com/openimsdk/tools/log"
"github.com/openimsdk/tools/mw"
"github.com/openimsdk/tools/system/program"
"github.com/openimsdk/tools/utils/datautil"
"github.com/openimsdk/tools/utils/network"
"github.com/openimsdk/tools/utils/runtimeenv"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
)
type Config struct {
@ -49,7 +46,7 @@ type Config struct {
Index conf.Index
}
func Start(ctx context.Context, config *Config, client discovery.Conn, server grpc.ServiceRegistrar) error {
func Start(ctx context.Context, config *Config, client discovery.Conn, _ grpc.ServiceRegistrar) error {
apiPort, err := datautil.GetElemByIndex(config.API.Api.Ports, int(config.Index))
if err != nil {
return err

7
internal/tools/cron_task.go
View File

@ -26,18 +26,15 @@ type CronTaskConfig struct {
CronTask config.CronTask
Share config.Share
Discovery config.Discovery
runTimeEnv string
}
func Start(ctx context.Context, conf *CronTaskConfig) error {
conf.runTimeEnv = runtimeenv.PrintRuntimeEnvironment()
log.CInfo(ctx, "CRON-TASK server is initializing", "runTimeEnv", conf.runTimeEnv, "chatRecordsClearTime", conf.CronTask.CronExecuteTime, "msgDestructTime", conf.CronTask.RetainChatRecords)
log.CInfo(ctx, "CRON-TASK server is initializing", "runTimeEnv", runtimeenv.RuntimeEnvironment(), "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(&conf.Discovery, conf.runTimeEnv, nil)
client, err := kdisc.NewDiscoveryRegister(&conf.Discovery, nil)
if err != nil {
return errs.WrapMsg(err, "failed to register discovery service")
}

21
pkg/common/cmd/api.go
View File

@ -19,6 +19,7 @@ import (
"github.com/openimsdk/open-im-server/v3/internal/api"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/startrpc"
"github.com/openimsdk/open-im-server/v3/version"
"github.com/openimsdk/tools/system/program"
"github.com/spf13/cobra"
@ -32,7 +33,7 @@ type ApiCmd struct {
}
func NewApiCmd() *ApiCmd {
apiConfig := api.Config{AllConfig: &config.AllConfig{}}
var apiConfig api.Config
ret := &ApiCmd{apiConfig: &apiConfig}
ret.configMap = map[string]any{
config.DiscoveryConfigFilename: &apiConfig.Discovery,
@ -61,7 +62,7 @@ func NewApiCmd() *ApiCmd {
ret.RootCmd = NewRootCmd(program.GetProcessName(), WithConfigMap(ret.configMap))
ret.ctx = context.WithValue(context.Background(), "version", version.Version)
ret.Command.RunE = func(cmd *cobra.Command, args []string) error {
apiConfig.ConfigPath = ret.configPath
apiConfig.ConfigPath = config.Path(ret.configPath)
return ret.runE()
}
return ret
@ -72,5 +73,19 @@ func (a *ApiCmd) Exec() error {
}
func (a *ApiCmd) runE() error {
return api.Start(a.ctx, a.Index(), a.apiConfig)
a.apiConfig.Index = config.Index(a.Index())
var prometheus config.Prometheus
return startrpc.Start(
a.ctx, &a.apiConfig.Discovery,
&prometheus,
a.apiConfig.API.Api.ListenIP, "",
false,
nil, int(a.apiConfig.Index),
a.apiConfig.Discovery.RpcService.MessageGateway,
&a.apiConfig.Notification,
a.apiConfig,
[]string{},
[]string{},
api.Start,
)
}

8
pkg/common/cmd/msg_gateway.go
View File

@ -57,10 +57,8 @@ func (m *MsgGatewayCmd) Exec() error {
func (m *MsgGatewayCmd) runE() error {
m.msgGatewayConfig.Index = config.Index(m.Index())
//return msggateway.Start(m.ctx, m.msgGatewayConfig, "", "")
//return msggateway.Start(m.ctx, m.msgGatewayConfig, "", "")
var prometheus config.Prometheus
rpc := m.msgGatewayConfig.MsgGateway.RPC
var prometheus config.Prometheus
return startrpc.Start(
m.ctx, &m.msgGatewayConfig.Discovery,
&prometheus,
@ -68,8 +66,8 @@ func (m *MsgGatewayCmd) runE() error {
rpc.AutoSetPorts,
rpc.Ports, int(m.msgGatewayConfig.Index),
m.msgGatewayConfig.Discovery.RpcService.MessageGateway,
"",
m.msgGatewayConfig.MsgGateway,
nil,
m.msgGatewayConfig,
[]string{},
[]string{},
msggateway.Start,

16
pkg/common/cmd/msg_transfer.go
View File

@ -19,6 +19,7 @@ import (
"github.com/openimsdk/open-im-server/v3/internal/msgtransfer"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/startrpc"
"github.com/openimsdk/open-im-server/v3/version"
"github.com/openimsdk/tools/system/program"
"github.com/spf13/cobra"
@ -57,5 +58,18 @@ func (m *MsgTransferCmd) Exec() error {
func (m *MsgTransferCmd) runE() error {
m.msgTransferConfig.Index = config.Index(m.Index())
return msgtransfer.Start(m.ctx, m.Index(), m.msgTransferConfig)
var prometheus config.Prometheus
return startrpc.Start(
m.ctx, &m.msgTransferConfig.Discovery,
&prometheus,
"", "",
false,
nil, int(m.msgTransferConfig.Index),
m.msgTransferConfig.Discovery.RpcService.MessageGateway,
nil,
m.msgTransferConfig,
[]string{},
[]string{},
msgtransfer.Start,
)
}

40
pkg/common/cmd/start.go
View File

@ -1,40 +0,0 @@
package cmd
//
//type StartFunc[C any] func(ctx context.Context, config *C, client discovery.Conn, server grpc.ServiceRegistrar) error
//
//func Start[C any](fn StartFunc[C]) {
// var _ RootCmd
// cmd := cobra.Command{
// Use: "Start openIM application",
// Long: fmt.Sprintf(`Start %s `, program.GetProcessName()),
// PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
// return rootCmd.persistentPreRun(cmd, opts...)
// },
// SilenceUsage: true,
// SilenceErrors: false,
// }
// cmd.Flags().StringP(config.FlagConf, "c", "", "path of config directory")
// cmd.Flags().IntP(config.FlagTransferIndex, "i", 0, "process startup sequence number")
//
//
//
//}
//
//func start[C any](fn StartFunc[C]) error {
//
//
// v := viper.New()
// v.SetConfigType("yaml")
// if err := v.ReadConfig(bytes.NewReader(confData)); err != nil {
// return err
// }
// fn := func(conf *mapstructure.DecoderConfig) {
// conf.TagName = config.StructTagName
// }
// if err := v.Unmarshal(val, fn); err != nil {
// return err
// }
//
// return nil
//}

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

@ -22,7 +22,6 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/startrpc"
"github.com/openimsdk/open-im-server/v3/version"
"github.com/openimsdk/tools/system/program"
"github.com/openimsdk/tools/utils/network"
"github.com/spf13/cobra"
)

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

@ -94,6 +94,9 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")),
)
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGTERM)
var gsrv grpcServiceRegistrar
err = rpcFn(ctx, config, client, &gsrv)
@ -161,9 +164,6 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf
}()
}
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGTERM)
select {
case val := <-sigs:
log.ZDebug(ctx, "recv exit", "signal", val.String())

14
tools/check-component/main.go
View File

@ -32,7 +32,6 @@ import (
"github.com/openimsdk/tools/mq/kafka"
"github.com/openimsdk/tools/s3/minio"
"github.com/openimsdk/tools/system/program"
"github.com/openimsdk/tools/utils/runtimeenv"
)
const maxRetry = 180
@ -84,37 +83,36 @@ func initConfig(configDir string) (*config.Mongo, *config.Redis, *config.Kafka,
discovery = &config.Discovery{}
thirdConfig = &config.Third{}
)
runtimeEnv := runtimeenv.PrintRuntimeEnvironment()
err := config.Load(configDir, config.MongodbConfigFileName, config.EnvPrefixMap[config.MongodbConfigFileName], runtimeEnv, mongoConfig)
err := config.Load(configDir, config.MongodbConfigFileName, config.EnvPrefixMap[config.MongodbConfigFileName], mongoConfig)
if err != nil {
return nil, nil, nil, nil, nil, err
}
err = config.Load(configDir, config.RedisConfigFileName, config.EnvPrefixMap[config.RedisConfigFileName], runtimeEnv, redisConfig)
err = config.Load(configDir, config.RedisConfigFileName, config.EnvPrefixMap[config.RedisConfigFileName], redisConfig)
if err != nil {
return nil, nil, nil, nil, nil, err
}
err = config.Load(configDir, config.KafkaConfigFileName, config.EnvPrefixMap[config.KafkaConfigFileName], runtimeEnv, kafkaConfig)
err = config.Load(configDir, config.KafkaConfigFileName, config.EnvPrefixMap[config.KafkaConfigFileName], kafkaConfig)
if err != nil {
return nil, nil, nil, nil, nil, err
}
err = config.Load(configDir, config.OpenIMRPCThirdCfgFileName, config.EnvPrefixMap[config.OpenIMRPCThirdCfgFileName], runtimeEnv, thirdConfig)
err = config.Load(configDir, config.OpenIMRPCThirdCfgFileName, config.EnvPrefixMap[config.OpenIMRPCThirdCfgFileName], thirdConfig)
if err != nil {
return nil, nil, nil, nil, nil, err
}
if thirdConfig.Object.Enable == "minio" {
err = config.Load(configDir, config.MinioConfigFileName, config.EnvPrefixMap[config.MinioConfigFileName], runtimeEnv, minioConfig)
err = config.Load(configDir, config.MinioConfigFileName, config.EnvPrefixMap[config.MinioConfigFileName], minioConfig)
if err != nil {
return nil, nil, nil, nil, nil, err
}
} else {
minioConfig = nil
}
err = config.Load(configDir, config.DiscoveryConfigFilename, config.EnvPrefixMap[config.DiscoveryConfigFilename], runtimeEnv, discovery)
err = config.Load(configDir, config.DiscoveryConfigFilename, config.EnvPrefixMap[config.DiscoveryConfigFilename], discovery)
if err != nil {
return nil, nil, nil, nil, nil, err
}

2
tools/seq/internal/seq.go
View File

@ -42,7 +42,7 @@ const (
)
func readConfig[T any](dir string, name string) (*T, error) {
if runtimeenv.PrintRuntimeEnvironment() == config.KUBERNETES {
if runtimeenv.RuntimeEnvironment() == config.KUBERNETES {
dir = os.Getenv(config.MountConfigFilePath)
}