Merge branch 'main' into localcache

# Conflicts:
#	go.mod
#	go.sum
#	internal/push/push_to_client.go

pkg/common/db/controller/msg_test.go

@@ -146,7 +146,7 @@ func Test_BatchInsertChat2DB(t *testing.T) {
 func GetDB() *commonMsgDatabase {
 	config.Config.Mongo.Address = []string{"203.56.175.233:37017"}
 	// config.Config.Mongo.Timeout = 60
-	config.Config.Mongo.Database = "openIM_v3"
+	config.Config.Mongo.Database = "openim_v3"
 	// config.Config.Mongo.Source = "admin"
 	config.Config.Mongo.Username = "root"
 	config.Config.Mongo.Password = "openIM123"

pkg/common/db/controller/user.go

@@ -142,12 +142,12 @@ func (u *userDatabase) Find(ctx context.Context, userIDs []string) (users []*rel
 	return u.cache.GetUsersInfo(ctx, userIDs)
 }
 
-// Find userInfo By Nickname
+// Find userInfo By Nickname.
 func (u *userDatabase) FindByNickname(ctx context.Context, nickname string) (users []*relation.UserModel, err error) {
 	return u.userDB.TakeByNickname(ctx, nickname)
 }
 
-// Find notificationAccouts
+// Find notificationAccouts.
 func (u *userDatabase) FindNotification(ctx context.Context, level int64) (users []*relation.UserModel, err error) {
 	return u.userDB.TakeNotification(ctx, level)
 }

pkg/common/db/s3/cos/cos.go

@@ -36,9 +36,9 @@ import (
 )
 
 const (
-	minPartSize = 1024 * 1024 * 1        // 1MB
-	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  = 1000
+	minPartSize int64 = 1024 * 1024 * 1        // 1MB
+	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
+	maxNumSize  int64 = 1000
 )
 
 const (
@@ -133,7 +133,7 @@ func (c *Cos) PartSize(ctx context.Context, size int64) (int64, error) {
 		return 0, errors.New("size must be greater than 0")
 	}
 	if size > maxPartSize*maxNumSize {
-		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
+		return 0, fmt.Errorf("COS size must be less than the maximum allowed limit")
 	}
 	if size <= minPartSize*maxNumSize {
 		return minPartSize, nil

pkg/common/db/s3/minio/minio.go

@@ -45,9 +45,9 @@ const (
 )
 
 const (
-	minPartSize = 1024 * 1024 * 5        // 1MB
-	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  = 10000
+	minPartSize int64 = 1024 * 1024 * 5        // 1MB
+	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
+	maxNumSize  int64 = 10000
 )
 
 const (
@@ -240,7 +240,7 @@ func (m *Minio) PartSize(ctx context.Context, size int64) (int64, error) {
 		return 0, errors.New("size must be greater than 0")
 	}
 	if size > maxPartSize*maxNumSize {
-		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
+		return 0, fmt.Errorf("MINIO size must be less than the maximum allowed limit")
 	}
 	if size <= minPartSize*maxNumSize {
 		return minPartSize, nil

pkg/common/db/s3/oss/oss.go

@@ -37,9 +37,9 @@ import (
 )
 
 const (
-	minPartSize = 1024 * 1024 * 1        // 1MB
-	maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
-	maxNumSize  = 10000
+	minPartSize int64 = 1024 * 1024 * 1        // 1MB
+	maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
+	maxNumSize  int64 = 10000
 )
 
 const (
@@ -141,7 +141,7 @@ func (o *OSS) PartSize(ctx context.Context, size int64) (int64, error) {
 		return 0, errors.New("size must be greater than 0")
 	}
 	if size > maxPartSize*maxNumSize {
-		return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
+		return 0, fmt.Errorf("OSS size must be less than the maximum allowed limit")
 	}
 	if size <= minPartSize*maxNumSize {
 		return minPartSize, nil

pkg/common/db/s3/s3.go

@@ -24,7 +24,7 @@ import (
 type PartLimit struct {
 	MinPartSize int64 `json:"minPartSize"`
 	MaxPartSize int64 `json:"maxPartSize"`
-	MaxNumSize  int   `json:"maxNumSize"`
+	MaxNumSize  int64 `json:"maxNumSize"`
 }
 
 type InitiateMultipartUploadResult struct {

pkg/common/discoveryregister/direct/directResolver.go

@@ -0,0 +1,81 @@
+package direct
+
+import (
+	"context"
+	"github.com/OpenIMSDK/tools/log"
+	"google.golang.org/grpc/resolver"
+	"math/rand"
+	"strings"
+)
+
+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/discoveryregister/direct/directconn.go

@@ -0,0 +1,154 @@
+package direct
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/OpenIMSDK/tools/errs"
+	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type ServiceAddresses map[string][]int
+
+func getServiceAddresses() ServiceAddresses {
+	return ServiceAddresses{
+		config2.Config.RpcRegisterName.OpenImUserName:           config2.Config.RpcPort.OpenImUserPort,
+		config2.Config.RpcRegisterName.OpenImFriendName:         config2.Config.RpcPort.OpenImFriendPort,
+		config2.Config.RpcRegisterName.OpenImMsgName:            config2.Config.RpcPort.OpenImMessagePort,
+		config2.Config.RpcRegisterName.OpenImMessageGatewayName: config2.Config.LongConnSvr.OpenImMessageGatewayPort,
+		config2.Config.RpcRegisterName.OpenImGroupName:          config2.Config.RpcPort.OpenImGroupPort,
+		config2.Config.RpcRegisterName.OpenImAuthName:           config2.Config.RpcPort.OpenImAuthPort,
+		config2.Config.RpcRegisterName.OpenImPushName:           config2.Config.RpcPort.OpenImPushPort,
+		config2.Config.RpcRegisterName.OpenImConversationName:   config2.Config.RpcPort.OpenImConversationPort,
+		config2.Config.RpcRegisterName.OpenImThirdName:          config2.Config.RpcPort.OpenImThirdPort,
+	}
+}
+
+type ConnDirect struct {
+	additionalOpts        []grpc.DialOption
+	currentServiceAddress string
+	conns                 map[string][]*grpc.ClientConn
+	resolverDirect        *ResolverDirect
+}
+
+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() (*ConnDirect, error) {
+	return &ConnDirect{
+		conns:          make(map[string][]*grpc.ClientConn),
+		resolverDirect: NewResolverDirect(),
+	}, 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()[serviceName]
+	var connections []*grpc.ClientConn
+	for _, port := range ports {
+		conn, err := cd.dialServiceWithoutResolver(ctx, fmt.Sprintf(config2.Config.Rpc.ListenIP+":%d", port), append(cd.additionalOpts, opts...)...)
+		if err != nil {
+			fmt.Printf("connect to port %d failed,serviceName %s, IP %s\n", port, serviceName, config2.Config.Rpc.ListenIP)
+		}
+		connections = append(connections, conn)
+	}
+
+	if len(connections) == 0 {
+		return nil, fmt.Errorf("no connections found for service: %s", serviceName)
+	}
+	return connections, nil
+}
+
+func (cd *ConnDirect) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+	// Get service addresses
+	addresses := getServiceAddresses()
+	address, ok := addresses[serviceName]
+	if !ok {
+		return nil, errs.Wrap(errors.New("unknown service name"), "serviceName", serviceName)
+	}
+	var result string
+	for _, addr := range address {
+		if result != "" {
+			result = result + "," + fmt.Sprintf(config2.Config.Rpc.ListenIP+":%d", addr)
+		} else {
+			result = fmt.Sprintf(config2.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.Wrap(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, err
+	}
+	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, err
+	}
+	return conn, nil
+}

pkg/common/discoveryregister/discoveryregister.go

@@ -16,6 +16,7 @@ package discoveryregister
 
 import (
 	"errors"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/direct"
 	"os"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/kubernetes"
@@ -36,6 +37,8 @@ func NewDiscoveryRegister(envType string) (discoveryregistry.SvcDiscoveryRegistr
 		return zookeeper.NewZookeeperDiscoveryRegister()
 	case "k8s":
 		return kubernetes.NewK8sDiscoveryRegister()
+	case "direct":
+		return direct.NewConnDirect()
 	default:
 		return nil, errors.New("envType not correct")
 	}

pkg/common/discoveryregister/discoveryregister_test.go

@@ -40,6 +40,7 @@ func TestNewDiscoveryRegister(t *testing.T) {
 	}{
 		{"zookeeper", false, true},
 		{"k8s", false, true}, // 假设 k8s 配置也已正确设置
+		{"direct", false, true},
 		{"invalid", true, false},
 	}
 

pkg/common/discoveryregister/kubernetes/kubernetes.go

@@ -104,7 +104,7 @@ func getSelfHost(ctx context.Context) string {
 	return host
 }
 
-// like openimserver-openim-msggateway-0.openimserver-openim-msggateway-headless.openim-lin.svc.cluster.local:88
+// like openimserver-openim-msggateway-0.openimserver-openim-msggateway-headless.openim-lin.svc.cluster.local:88.
 func getMsgGatewayHost(ctx context.Context) []string {
 	port := 88
 	instance := "openimserver"
@@ -132,20 +132,40 @@ func getMsgGatewayHost(ctx context.Context) []string {
 // GetConns returns the gRPC client connections to the specified service.
 func (cli *K8sDR) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) {
 
+	// This conditional checks if the serviceName is not the OpenImMessageGatewayName.
+	// It seems to handle a special case for the OpenImMessageGateway.
 	if serviceName != config.Config.RpcRegisterName.OpenImMessageGatewayName {
+		// DialContext creates a client connection to the given target (serviceName) using the specified context.
+		// 'cli.options' are likely default or common options for all connections in this struct.
+		// 'opts...' allows for additional gRPC dial options to be passed and used.
 		conn, err := grpc.DialContext(ctx, serviceName, append(cli.options, opts...)...)
+
+		// The function returns a slice of client connections with the new connection, or an error if occurred.
 		return []*grpc.ClientConn{conn}, err
 	} else {
+		// This block is executed if the serviceName is OpenImMessageGatewayName.
+		// 'ret' will accumulate the connections to return.
 		var ret []*grpc.ClientConn
+
+		// getMsgGatewayHost presumably retrieves hosts for the message gateway service.
+		// The context is passed, likely for cancellation and timeout control.
 		gatewayHosts := getMsgGatewayHost(ctx)
+
+		// Iterating over the retrieved gateway hosts.
 		for _, host := range gatewayHosts {
+			// Establishes a connection to each host.
+			// Again, appending cli.options with any additional opts provided.
 			conn, err := grpc.DialContext(ctx, host, append(cli.options, opts...)...)
+
+			// If there's an error while dialing any host, the function returns immediately with the error.
 			if err != nil {
 				return nil, err
 			} else {
+				// If the connection is successful, it is added to the 'ret' slice.
 				ret = append(ret, conn)
 			}
 		}
+		// After all hosts are processed, the slice of connections is returned.
 		return ret, nil
 	}
 }
@@ -168,7 +188,7 @@ func (cli *K8sDR) CloseConn(conn *grpc.ClientConn) {
 	conn.Close()
 }
 
-// do not use this method for call rpc
+// do not use this method for call rpc.
 func (cli *K8sDR) GetClientLocalConns() map[string][]*grpc.ClientConn {
 	fmt.Println("should not call this function!!!!!!!!!!!!!!!!!!!!!!!!!")
 	return nil

pkg/common/startrpc/start.go

@@ -71,7 +71,7 @@ func Start(
 	}
 
 	defer client.Close()
-	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
 	registerIP, err := network.GetRpcRegisterIP(config.Config.Rpc.RegisterIP)
 	if err != nil {
 		return err