复制项目

pkg/common/kdisc/direct/direct_resolver.go

@@ -0,0 +1,96 @@
+// Copyright © 2024 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 direct
+
+import (
+	"context"
+	"math/rand"
+	"strings"
+
+	"github.com/openimsdk/tools/log"
+	"google.golang.org/grpc/resolver"
+)
+
+const (
+	slashSeparator = "/"
+	// EndpointSepChar is the separator char in endpoints.
+	EndpointSepChar = ','
+
+	subsetSize = 32
+	scheme     = "direct"
+)
+
+type ResolverDirect struct {
+}
+
+func NewResolverDirect() *ResolverDirect {
+	return &ResolverDirect{}
+}
+
+func (rd *ResolverDirect) Build(target resolver.Target, cc resolver.ClientConn, _ resolver.BuildOptions) (
+	resolver.Resolver, error) {
+	log.ZDebug(context.Background(), "Build", "target", target)
+	endpoints := strings.FieldsFunc(GetEndpoints(target), func(r rune) bool {
+		return r == EndpointSepChar
+	})
+	endpoints = subset(endpoints, subsetSize)
+	addrs := make([]resolver.Address, 0, len(endpoints))
+
+	for _, val := range endpoints {
+		addrs = append(addrs, resolver.Address{
+			Addr: val,
+		})
+	}
+	if err := cc.UpdateState(resolver.State{
+		Addresses: addrs,
+	}); err != nil {
+		return nil, err
+	}
+
+	return &nopResolver{cc: cc}, nil
+}
+func init() {
+	resolver.Register(&ResolverDirect{})
+}
+func (rd *ResolverDirect) Scheme() string {
+	return scheme // return your custom scheme name
+}
+
+// GetEndpoints returns the endpoints from the given target.
+func GetEndpoints(target resolver.Target) string {
+	return strings.Trim(target.URL.Path, slashSeparator)
+}
+func subset(set []string, sub int) []string {
+	rand.Shuffle(len(set), func(i, j int) {
+		set[i], set[j] = set[j], set[i]
+	})
+	if len(set) <= sub {
+		return set
+	}
+
+	return set[:sub]
+}
+
+type nopResolver struct {
+	cc resolver.ClientConn
+}
+
+func (n nopResolver) ResolveNow(options resolver.ResolveNowOptions) {
+
+}
+
+func (n nopResolver) Close() {
+
+}

pkg/common/kdisc/direct/directconn.go

@@ -0,0 +1,174 @@
+// Copyright © 2024 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 direct
+
+//import (
+//	"context"
+//	"fmt"
+//
+//	config2 "github.com/openimsdk/chat-deploy/v3/pkg/common/config"
+//	"github.com/openimsdk/tools/errs"
+//	"google.golang.org/grpc"
+//	"google.golang.org/grpc/credentials/insecure"
+//)
+//
+//type ServiceAddresses map[string][]int
+//
+//func getServiceAddresses(rpcRegisterName *config2.RpcRegisterName,
+//	rpcPort *config2.RpcPort, longConnSvrPort []int) ServiceAddresses {
+//	return ServiceAddresses{
+//		rpcRegisterName.OpenImUserName:           rpcPort.OpenImUserPort,
+//		rpcRegisterName.OpenImFriendName:         rpcPort.OpenImFriendPort,
+//		rpcRegisterName.OpenImMsgName:            rpcPort.OpenImMessagePort,
+//		rpcRegisterName.OpenImMessageGatewayName: longConnSvrPort,
+//		rpcRegisterName.OpenImGroupName:          rpcPort.OpenImGroupPort,
+//		rpcRegisterName.OpenImAuthName:           rpcPort.OpenImAuthPort,
+//		rpcRegisterName.OpenImPushName:           rpcPort.OpenImPushPort,
+//		rpcRegisterName.OpenImConversationName:   rpcPort.OpenImConversationPort,
+//		rpcRegisterName.OpenImThirdName:          rpcPort.OpenImThirdPort,
+//	}
+//}
+//
+//type ConnDirect struct {
+//	additionalOpts        []grpc.DialOption
+//	currentServiceAddress string
+//	conns                 map[string][]*grpc.ClientConn
+//	resolverDirect        *ResolverDirect
+//	config                *config2.GlobalConfig
+//}
+//
+//func (cd *ConnDirect) GetClientLocalConns() map[string][]*grpc.ClientConn {
+//	return nil
+//}
+//
+//func (cd *ConnDirect) GetUserIdHashGatewayHost(ctx context.Context, userId string) (string, error) {
+//	return "", nil
+//}
+//
+//func (cd *ConnDirect) Register(serviceName, host string, port int, opts ...grpc.DialOption) error {
+//	return nil
+//}
+//
+//func (cd *ConnDirect) UnRegister() error {
+//	return nil
+//}
+//
+//func (cd *ConnDirect) CreateRpcRootNodes(serviceNames []string) error {
+//	return nil
+//}
+//
+//func (cd *ConnDirect) RegisterConf2Registry(key string, conf []byte) error {
+//	return nil
+//}
+//
+//func (cd *ConnDirect) GetConfFromRegistry(key string) ([]byte, error) {
+//	return nil, nil
+//}
+//
+//func (cd *ConnDirect) Close() {
+//
+//}
+//
+//func NewConnDirect(config *config2.GlobalConfig) (*ConnDirect, error) {
+//	return &ConnDirect{
+//		conns:          make(map[string][]*grpc.ClientConn),
+//		resolverDirect: NewResolverDirect(),
+//		config:         config,
+//	}, nil
+//}
+//
+//func (cd *ConnDirect) GetConns(ctx context.Context,
+//	serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) {
+//
+//	if conns, exists := cd.conns[serviceName]; exists {
+//		return conns, nil
+//	}
+//	ports := getServiceAddresses(&cd.config.RpcRegisterName,
+//		&cd.config.RpcPort, cd.config.LongConnSvr.OpenImMessageGatewayPort)[serviceName]
+//	var connections []*grpc.ClientConn
+//	for _, port := range ports {
+//		conn, err := cd.dialServiceWithoutResolver(ctx, fmt.Sprintf(cd.config.Rpc.ListenIP+":%d", port), append(cd.additionalOpts, opts...)...)
+//		if err != nil {
+//			return nil, errs.Wrap(fmt.Errorf("connect to port %d failed,serviceName %s, IP %s", port, serviceName, cd.config.Rpc.ListenIP))
+//		}
+//		connections = append(connections, conn)
+//	}
+//
+//	if len(connections) == 0 {
+//		return nil, errs.New("no connections found for service", "serviceName", serviceName).Wrap()
+//	}
+//	return connections, nil
+//}
+//
+//func (cd *ConnDirect) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+//	// Get service addresses
+//	addresses := getServiceAddresses(&cd.config.RpcRegisterName,
+//		&cd.config.RpcPort, cd.config.LongConnSvr.OpenImMessageGatewayPort)
+//	address, ok := addresses[serviceName]
+//	if !ok {
+//		return nil, errs.New("unknown service name", "serviceName", serviceName).Wrap()
+//	}
+//	var result string
+//	for _, addr := range address {
+//		if result != "" {
+//			result = result + "," + fmt.Sprintf(cd.config.Rpc.ListenIP+":%d", addr)
+//		} else {
+//			result = fmt.Sprintf(cd.config.Rpc.ListenIP+":%d", addr)
+//		}
+//	}
+//	// Try to dial a new connection
+//	conn, err := cd.dialService(ctx, result, append(cd.additionalOpts, opts...)...)
+//	if err != nil {
+//		return nil, errs.WrapMsg(err, "address", result)
+//	}
+//
+//	// Store the new connection
+//	cd.conns[serviceName] = append(cd.conns[serviceName], conn)
+//	return conn, nil
+//}
+//
+//func (cd *ConnDirect) GetSelfConnTarget() string {
+//	return cd.currentServiceAddress
+//}
+//
+//func (cd *ConnDirect) AddOption(opts ...grpc.DialOption) {
+//	cd.additionalOpts = append(cd.additionalOpts, opts...)
+//}
+//
+//func (cd *ConnDirect) CloseConn(conn *grpc.ClientConn) {
+//	if conn != nil {
+//		conn.Close()
+//	}
+//}
+//
+//func (cd *ConnDirect) dialService(ctx context.Context, address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+//	options := append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+//	conn, err := grpc.DialContext(ctx, cd.resolverDirect.Scheme()+":///"+address, options...)
+//
+//	if err != nil {
+//		return nil, errs.WrapMsg(err, "address", address)
+//	}
+//	return conn, nil
+//}
+//
+//func (cd *ConnDirect) dialServiceWithoutResolver(ctx context.Context, address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+//	options := append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+//	conn, err := grpc.DialContext(ctx, address, options...)
+//
+//	if err != nil {
+//		return nil, errs.Wrap(err)
+//	}
+//	return conn, nil
+//}

pkg/common/kdisc/direct/doc.go

@@ -0,0 +1,15 @@
+// Copyright © 2024 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 direct // import "github.com/openimsdk/chat-deploy/v3/pkg/common/discoveryregister/direct"

pkg/common/kdisc/discoveryregister.go

@@ -0,0 +1,51 @@
+// 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 kdisc
+
+import (
+	"time"
+
+	"git.imall.cloud/openim/chat/pkg/common/config"
+	"github.com/openimsdk/tools/discovery"
+	"github.com/openimsdk/tools/discovery/etcd"
+	"github.com/openimsdk/tools/discovery/kubernetes"
+	"github.com/openimsdk/tools/errs"
+)
+
+const (
+	ETCDCONST       = "etcd"
+	KUBERNETESCONST = "kubernetes"
+	DIRECTCONST     = "direct"
+)
+
+// NewDiscoveryRegister creates a new service discovery and registry client based on the provided environment type.
+func NewDiscoveryRegister(discovery *config.Discovery, runtimeEnv string, watchNames []string) (discovery.SvcDiscoveryRegistry, error) {
+	if runtimeEnv == KUBERNETESCONST {
+		return kubernetes.NewKubernetesConnManager(discovery.Kubernetes.Namespace)
+	}
+
+	switch discovery.Enable {
+	case ETCDCONST:
+		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))
+	default:
+		return nil, errs.New("unsupported discovery type", "type", discovery.Enable).Wrap()
+	}
+}

pkg/common/kdisc/etcd/config_manager.go

@@ -0,0 +1,106 @@
+package etcd
+
+import (
+	"context"
+	"os"
+	"os/exec"
+	"runtime"
+	"sync"
+	"syscall"
+
+	"github.com/openimsdk/tools/errs"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/utils/datautil"
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+var (
+	ShutDowns []func() error
+)
+
+func RegisterShutDown(shutDown ...func() error) {
+	ShutDowns = append(ShutDowns, shutDown...)
+}
+
+type ConfigManager struct {
+	client           *clientv3.Client
+	watchConfigNames []string
+	lock             sync.Mutex
+}
+
+func BuildKey(s string) string {
+	return ConfigKeyPrefix + s
+}
+
+func NewConfigManager(client *clientv3.Client, configNames []string) *ConfigManager {
+	return &ConfigManager{
+		client:           client,
+		watchConfigNames: datautil.Batch(func(s string) string { return BuildKey(s) }, append(configNames, RestartKey))}
+}
+
+func (c *ConfigManager) Watch(ctx context.Context) {
+	chans := make([]clientv3.WatchChan, 0, len(c.watchConfigNames))
+	for _, name := range c.watchConfigNames {
+		chans = append(chans, c.client.Watch(ctx, name, clientv3.WithPrefix()))
+	}
+
+	doWatch := func(watchChan clientv3.WatchChan) {
+		for watchResp := range watchChan {
+			if watchResp.Err() != nil {
+				log.ZError(ctx, "watch err", errs.Wrap(watchResp.Err()))
+				continue
+			}
+			for _, event := range watchResp.Events {
+				if event.IsModify() {
+					if datautil.Contain(string(event.Kv.Key), c.watchConfigNames...) {
+						c.lock.Lock()
+						err := restartServer(ctx)
+						if err != nil {
+							log.ZError(ctx, "restart server err", err)
+						}
+						c.lock.Unlock()
+					}
+				}
+			}
+		}
+	}
+	for _, ch := range chans {
+		go doWatch(ch)
+	}
+}
+
+func restartServer(ctx context.Context) error {
+	exePath, err := os.Executable()
+	if err != nil {
+		return errs.New("get executable path fail").Wrap()
+	}
+
+	args := os.Args
+	env := os.Environ()
+
+	cmd := exec.Command(exePath, args[1:]...)
+	cmd.Env = env
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	cmd.Stdin = os.Stdin
+
+	if runtime.GOOS != "windows" {
+		cmd.SysProcAttr = &syscall.SysProcAttr{}
+	}
+	log.ZInfo(ctx, "shutdown server")
+	for _, f := range ShutDowns {
+		if err = f(); err != nil {
+			log.ZError(ctx, "shutdown fail", err)
+		}
+	}
+
+	log.ZInfo(ctx, "restart server")
+	err = cmd.Start()
+	if err != nil {
+		return errs.New("restart server fail").Wrap()
+	}
+	log.ZInfo(ctx, "cmd start over")
+
+	os.Exit(0)
+	return nil
+}

pkg/common/kdisc/etcd/const.go

@@ -0,0 +1,9 @@
+package etcd
+
+const (
+	ConfigKeyPrefix       = "/chat/config/"
+	RestartKey            = "restart"
+	EnableConfigCenterKey = "enable-config-center"
+	Enable                = "enable"
+	Disable               = "disable"
+)