conflict

internal/api/config_manager.go

@@ -259,6 +259,10 @@ func (cm *ConfigManager) restart(c *gin.Context) {
 }
 
 func (cm *ConfigManager) SetEnableConfigManager(c *gin.Context) {
+	if cm.config.Discovery.Enable != config.ETCD {
+		apiresp.GinError(c, errs.New("only etcd support config manager").Wrap())
+		return
+	}
 	var req apistruct.SetEnableConfigManagerReq
 	if err := c.BindJSON(&req); err != nil {
 		apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())

internal/api/init.go

@@ -1,3 +1,17 @@
+// Copyright © 2023 OpenIM. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package api
 
 import (
@@ -12,36 +26,37 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/tools/mw"
-	"github.com/openimsdk/tools/utils/datautil"
-	"github.com/openimsdk/tools/utils/network"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-
-	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
+	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/open-im-server/v3/pkg/common/prommetrics"
-	"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/jsonutil"
+	"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 {
-	API       config.API
-	Share     config.Share
-	Discovery config.Discovery
+	*conf.AllConfig
+
+	RuntimeEnv string
+	ConfigPath string
 }
 
-func Start(ctx context.Context, index int, cfg *Config) error {
-	apiPort, err := datautil.GetElemByIndex(cfg.API.Api.Ports, index)
+func Start(ctx context.Context, index int, config *Config) error {
+	apiPort, err := datautil.GetElemByIndex(config.API.Api.Ports, index)
 	if err != nil {
 		return err
 	}
 
-	var client discovery.SvcDiscoveryRegistry
+	config.RuntimeEnv = runtimeenv.PrintRuntimeEnvironment()
 
 	client, err := kdisc.NewDiscoveryRegister(&config.Discovery, config.RuntimeEnv, []string{
 		config.Discovery.RpcService.MessageGateway,
@@ -49,8 +64,7 @@ func Start(ctx context.Context, index int, cfg *Config) error {
 	if err != nil {
 		return errs.WrapMsg(err, "failed to register discovery service")
 	}
-	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
 
 	var (
 		netDone        = make(chan struct{}, 1)
@@ -58,10 +72,6 @@ func Start(ctx context.Context, index int, cfg *Config) error {
 		prometheusPort int
 	)
 
-	router, err := newGinRouter(ctx, client, cfg)
-	if err != nil {
-		return err
-	}
 	registerIP, err := network.GetRpcRegisterIP("")
 	if err != nil {
 		return err
@@ -78,16 +88,20 @@ func Start(ctx context.Context, index int, cfg *Config) error {
 		return listener, port, nil
 	}
 
-	if cfg.API.Prometheus.AutoSetPorts && cfg.Discovery.Enable != config.ETCD {
+	if config.API.Prometheus.AutoSetPorts && config.Discovery.Enable != conf.ETCD {
 		return errs.New("only etcd support autoSetPorts", "RegisterName", "api").Wrap()
 	}
 
-	if cfg.API.Prometheus.Enable {
+	router, err := newGinRouter(ctx, client, config)
+	if err != nil {
+		return err
+	}
+	if config.API.Prometheus.Enable {
 		var (
 			listener net.Listener
 		)
 
-		if cfg.API.Prometheus.AutoSetPorts {
+		if config.API.Prometheus.AutoSetPorts {
 			listener, prometheusPort, err = getAutoPort()
 			if err != nil {
 				return err
@@ -100,7 +114,7 @@ func Start(ctx context.Context, index int, cfg *Config) error {
 				return errs.WrapMsg(err, "etcd put err")
 			}
 		} else {
-			prometheusPort, err = datautil.GetElemByIndex(cfg.API.Prometheus.Ports, index)
+			prometheusPort, err = datautil.GetElemByIndex(config.API.Prometheus.Ports, index)
 			if err != nil {
 				return err
 			}
@@ -118,31 +132,42 @@ func Start(ctx context.Context, index int, cfg *Config) error {
 		}()
 
 	}
-	address := net.JoinHostPort(network.GetListenIP(cfg.API.Api.ListenIP), strconv.Itoa(apiPort))
+	address := net.JoinHostPort(network.GetListenIP(config.API.Api.ListenIP), strconv.Itoa(apiPort))
 
 	server := http.Server{Addr: address, Handler: router}
-	log.CInfo(ctx, "API server is initializing", "address", address, "apiPort", apiPort, "prometheusPort", prometheusPort)
+	log.CInfo(ctx, "API server is initializing", "runtimeEnv", config.RuntimeEnv, "address", address, "apiPort", apiPort, "prometheusPort", prometheusPort)
 	go func() {
 		err = server.ListenAndServe()
 		if err != nil && !errors.Is(err, http.ErrServerClosed) {
 			netErr = errs.WrapMsg(err, fmt.Sprintf("api start err: %s", server.Addr))
 			netDone <- struct{}{}
-
 		}
 	}()
 
+	if config.Discovery.Enable == conf.ETCD {
+		cm := disetcd.NewConfigManager(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient(), config.GetConfigNames())
+		cm.Watch(ctx)
+	}
+
 	sigs := make(chan os.Signal, 1)
 	signal.Notify(sigs, syscall.SIGTERM)
 
-	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
-	defer cancel()
-	select {
-	case <-sigs:
-		program.SIGTERMExit()
+	shutdown := func() error {
+		ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+		defer cancel()
 		err := server.Shutdown(ctx)
 		if err != nil {
 			return errs.WrapMsg(err, "shutdown err")
 		}
+		return nil
+	}
+	disetcd.RegisterShutDown(shutdown)
+	select {
+	case <-sigs:
+		program.SIGTERMExit()
+		if err := shutdown(); err != nil {
+			return err
+		}
 	case <-netDone:
 		close(netDone)
 		return netErr

internal/api/jssdk/tools.go

@@ -2,6 +2,9 @@ package jssdk
 
 import (
 	"context"
+	"io"
+	"strings"
+
 	"github.com/gin-gonic/gin"
 	"github.com/openimsdk/tools/a2r"
 	"github.com/openimsdk/tools/apiresp"
@@ -9,8 +12,6 @@ import (
 	"github.com/openimsdk/tools/errs"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"
-	"io"
-	"strings"
 )
 
 func field[A, B, C any](ctx context.Context, fn func(ctx context.Context, req *A, opts ...grpc.CallOption) (*B, error), req *A, get func(*B) C) (C, error) {

internal/api/prometheus_discovery.go

@@ -5,7 +5,7 @@ import (
 	"net/http"
 
 	"github.com/gin-gonic/gin"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/tools/apiresp"
 	"github.com/openimsdk/tools/discovery"
@@ -20,18 +20,18 @@ type PrometheusDiscoveryApi struct {
 	client *clientv3.Client
 }
 
-func NewPrometheusDiscoveryApi(cfg *Config, client discovery.SvcDiscoveryRegistry) *PrometheusDiscoveryApi {
+func NewPrometheusDiscoveryApi(config *Config, client discovery.SvcDiscoveryRegistry) *PrometheusDiscoveryApi {
 	api := &PrometheusDiscoveryApi{
-		config: cfg,
+		config: config,
 	}
-	if cfg.Discovery.Enable == config.ETCD {
+	if config.Discovery.Enable == conf.ETCD {
 		api.client = client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
 	}
 	return api
 }
 
 func (p *PrometheusDiscoveryApi) Enable(c *gin.Context) {
-	if p.config.Discovery.Enable != config.ETCD {
+	if p.config.Discovery.Enable != conf.ETCD {
 		c.JSON(http.StatusOK, []struct{}{})
 		c.Abort()
 	}
@@ -74,39 +74,39 @@ func (p *PrometheusDiscoveryApi) Api(c *gin.Context) {
 }
 
 func (p *PrometheusDiscoveryApi) User(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.User)
+	p.discovery(c, p.config.Discovery.RpcService.User)
 }
 
 func (p *PrometheusDiscoveryApi) Group(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.Group)
+	p.discovery(c, p.config.Discovery.RpcService.Group)
 }
 
 func (p *PrometheusDiscoveryApi) Msg(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.Msg)
+	p.discovery(c, p.config.Discovery.RpcService.Msg)
 }
 
 func (p *PrometheusDiscoveryApi) Friend(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.Friend)
+	p.discovery(c, p.config.Discovery.RpcService.Friend)
 }
 
 func (p *PrometheusDiscoveryApi) Conversation(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.Conversation)
+	p.discovery(c, p.config.Discovery.RpcService.Conversation)
 }
 
 func (p *PrometheusDiscoveryApi) Third(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.Third)
+	p.discovery(c, p.config.Discovery.RpcService.Third)
 }
 
 func (p *PrometheusDiscoveryApi) Auth(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.Auth)
+	p.discovery(c, p.config.Discovery.RpcService.Auth)
 }
 
 func (p *PrometheusDiscoveryApi) Push(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.Push)
+	p.discovery(c, p.config.Discovery.RpcService.Push)
 }
 
 func (p *PrometheusDiscoveryApi) MessageGateway(c *gin.Context) {
-	p.discovery(c, p.config.Share.RpcRegisterName.MessageGateway)
+	p.discovery(c, p.config.Discovery.RpcService.MessageGateway)
 }
 
 func (p *PrometheusDiscoveryApi) MessageTransfer(c *gin.Context) {

internal/api/router.go

@@ -5,29 +5,29 @@ import (
 	"net/http"
 	"strings"
 
+	"github.com/gin-contrib/gzip"
+	"github.com/gin-gonic/gin"
+	"github.com/gin-gonic/gin/binding"
+	"github.com/go-playground/validator/v10"
+	"github.com/openimsdk/open-im-server/v3/internal/api/jssdk"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
 	pbAuth "github.com/openimsdk/protocol/auth"
+	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/protocol/conversation"
 	"github.com/openimsdk/protocol/group"
 	"github.com/openimsdk/protocol/msg"
 	"github.com/openimsdk/protocol/relation"
 	"github.com/openimsdk/protocol/third"
 	"github.com/openimsdk/protocol/user"
-
-	"github.com/openimsdk/open-im-server/v3/internal/api/jssdk"
-
-	"github.com/gin-contrib/gzip"
-
-	"github.com/gin-gonic/gin"
-	"github.com/gin-gonic/gin/binding"
-	"github.com/go-playground/validator/v10"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
-	"github.com/openimsdk/protocol/constant"
 	"github.com/openimsdk/tools/apiresp"
 	"github.com/openimsdk/tools/discovery"
+	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/log"
 	"github.com/openimsdk/tools/mw"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 const (
@@ -52,32 +52,32 @@ func prommetricsGin() gin.HandlerFunc {
 	}
 }
 
-func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, config *Config) (*gin.Engine, error) {
-	authConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Auth)
+func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, cfg *Config) (*gin.Engine, error) {
+	authConn, err := client.GetConn(ctx, cfg.Discovery.RpcService.Auth)
 	if err != nil {
 		return nil, err
 	}
-	userConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.User)
+	userConn, err := client.GetConn(ctx, cfg.Discovery.RpcService.User)
 	if err != nil {
 		return nil, err
 	}
-	groupConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Group)
+	groupConn, err := client.GetConn(ctx, cfg.Discovery.RpcService.Group)
 	if err != nil {
 		return nil, err
 	}
-	friendConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Friend)
+	friendConn, err := client.GetConn(ctx, cfg.Discovery.RpcService.Friend)
 	if err != nil {
 		return nil, err
 	}
-	conversationConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Conversation)
+	conversationConn, err := client.GetConn(ctx, cfg.Discovery.RpcService.Conversation)
 	if err != nil {
 		return nil, err
 	}
-	thirdConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Third)
+	thirdConn, err := client.GetConn(ctx, cfg.Discovery.RpcService.Third)
 	if err != nil {
 		return nil, err
 	}
-	msgConn, err := client.GetConn(ctx, config.Share.RpcRegisterName.Msg)
+	msgConn, err := client.GetConn(ctx, cfg.Discovery.RpcService.Msg)
 	if err != nil {
 		return nil, err
 	}
@@ -86,7 +86,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 	if v, ok := binding.Validator.Engine().(*validator.Validate); ok {
 		_ = v.RegisterValidation("required_if", RequiredIf)
 	}
-	switch config.API.Api.CompressionLevel {
+	switch cfg.API.Api.CompressionLevel {
 	case NoCompression:
 	case DefaultCompression:
 		r.Use(gzip.Gzip(gzip.DefaultCompression))
@@ -95,11 +95,12 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 	case BestSpeed:
 		r.Use(gzip.Gzip(gzip.BestSpeed))
 	}
-	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)))
-	u := NewUserApi(user.NewUserClient(userConn), client, config.Share.RpcRegisterName)
-	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), config.Share.IMAdminUserID)
-	userRouterGroup := r.Group("/user")
+	r.Use(prommetricsGin(), gin.RecoveryWithWriter(gin.DefaultErrorWriter, mw.GinPanicErr), mw.CorsHandler(),
+		mw.GinParseOperationID(), GinParseToken(rpcli.NewAuthClient(authConn)))
+
+	u := NewUserApi(user.NewUserClient(userConn), client, cfg.Discovery.RpcService)
 	{
+		userRouterGroup := r.Group("/user")
 		userRouterGroup.POST("/user_register", u.UserRegister)
 		userRouterGroup.POST("/update_user_info", u.UpdateUserInfo)
 		userRouterGroup.POST("/update_user_info_ex", u.UpdateUserInfoEx)
@@ -198,7 +199,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 	}
 	// Third service
 	{
-		t := NewThirdApi(third.NewThirdClient(thirdConn), config.API.Prometheus.GrafanaURL)
+		t := NewThirdApi(third.NewThirdClient(thirdConn), cfg.API.Prometheus.GrafanaURL)
 		thirdGroup := r.Group("/third")
 		thirdGroup.GET("/prometheus", t.GetPrometheus)
 		thirdGroup.POST("/fcm_update_token", t.FcmUpdateToken)
@@ -222,6 +223,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 		objectGroup.GET("/*name", t.ObjectRedirect)
 	}
 	// Message
+	m := NewMessageApi(msg.NewMsgClient(msgConn), rpcli.NewUserClient(userConn), cfg.Share.IMAdminUserID)
 	{
 		msgGroup := r.Group("/msg")
 		msgGroup.POST("/newest_seq", m.GetSeq)
@@ -244,6 +246,8 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 		msgGroup.POST("/batch_send_msg", m.BatchSendMsg)
 		msgGroup.POST("/check_msg_is_send_success", m.CheckMsgIsSendSuccess)
 		msgGroup.POST("/get_server_time", m.GetServerTime)
+		msgGroup.POST("/get_stream_msg", m.GetStreamMsg)
+		msgGroup.POST("/append_stream_msg", m.AppendStreamMsg)
 	}
 	// Conversation
 	{
@@ -278,7 +282,7 @@ func newGinRouter(ctx context.Context, client discovery.SvcDiscoveryRegistry, co
 		jssdk.POST("/get_active_conversations", j.GetActiveConversations)
 	}
 	{
-		pd := NewPrometheusDiscoveryApi(config, client)
+		pd := NewPrometheusDiscoveryApi(cfg, client)
 		proDiscoveryGroup := r.Group("/prometheus_discovery", pd.Enable)
 		proDiscoveryGroup.GET("/api", pd.Api)
 		proDiscoveryGroup.GET("/user", pd.User)

internal/api/user.go

@@ -30,10 +30,10 @@ import (
 type UserApi struct {
 	Client user.UserClient
 	discov discovery.SvcDiscoveryRegistry
-	config config.RpcRegisterName
+	config config.RpcService
 }
 
-func NewUserApi(client user.UserClient, discov discovery.SvcDiscoveryRegistry, config config.RpcRegisterName) UserApi {
+func NewUserApi(client user.UserClient, discov discovery.SvcDiscoveryRegistry, config config.RpcService) UserApi {
 	return UserApi{Client: client, discov: discov, config: config}
 }