test conn

internal/api/auth.go

@@ -12,38 +12,32 @@ import (
 )
 
 func NewAuth(discov discoveryregistry.SvcDiscoveryRegistry) *Auth {
-	// conn, err := discov.GetConn(context.Background(), config.Config.RpcRegisterName.OpenImAuthName)
-	// if err != nil {
-	// panic(err)
-	// }
-	return &Auth{discov: discov}
+	conn, err := discov.GetConn(context.Background(), config.Config.RpcRegisterName.OpenImAuthName)
+	if err != nil {
+		panic(err)
+	}
+	client := auth.NewAuthClient(conn)
+	return &Auth{discov: discov, conn: conn, client: client}
 }
 
 type Auth struct {
 	conn   *grpc.ClientConn
+	client auth.AuthClient
 	discov discoveryregistry.SvcDiscoveryRegistry
 }
 
-func (o *Auth) client(ctx context.Context) (auth.AuthClient, error) {
-	c, err := o.discov.GetConn(ctx, config.Config.RpcRegisterName.OpenImAuthName)
-	if err != nil {
-		return nil, err
-	}
-	return auth.NewAuthClient(c), nil
-}
-
-func (o *Auth) UserRegister(c *gin.Context) {
-	//a2r.Call(auth.AuthClient.UserRegister, o.userClient, c) // todo
+func (o *Auth) Client(ctx context.Context) (auth.AuthClient, error) {
+	return o.client, nil
 }
 
 func (o *Auth) UserToken(c *gin.Context) {
-	a2r.Call(auth.AuthClient.UserToken, o.client, c)
+	a2r.Call(auth.AuthClient.UserToken, o.Client, c)
 }
 
 func (o *Auth) ParseToken(c *gin.Context) {
-	a2r.Call(auth.AuthClient.ParseToken, o.client, c)
+	a2r.Call(auth.AuthClient.ParseToken, o.Client, c)
 }
 
 func (o *Auth) ForceLogout(c *gin.Context) {
-	a2r.Call(auth.AuthClient.ForceLogout, o.client, c)
+	a2r.Call(auth.AuthClient.ForceLogout, o.Client, c)
 }

internal/push/push_to_client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+
 	"github.com/OpenIMSDK/Open-IM-Server/internal/push/offlinepush"
 	"github.com/OpenIMSDK/Open-IM-Server/internal/push/offlinepush/fcm"
 	"github.com/OpenIMSDK/Open-IM-Server/internal/push/offlinepush/getui"
@@ -255,6 +256,7 @@ func (p *Pusher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
 	for _, v := range conns {
 		msgClient := msggateway.NewMsgGatewayClient(v)
 		reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: pushToUserIDs})
+		v.Close()
 		if err != nil {
 			continue
 		}
@@ -262,6 +264,7 @@ func (p *Pusher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
 		if reply != nil && reply.SinglePushResult != nil {
 			wsResults = append(wsResults, reply.SinglePushResult...)
 		}
+
 	}
 	return wsResults, nil
 }

internal/rpc/auth/auth.go

@@ -108,6 +108,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID
 		client := msggateway.NewMsgGatewayClient(v)
 		kickReq := &msggateway.KickUserOfflineReq{KickUserIDList: []string{userID}, PlatformID: platformID}
 		_, err := client.KickUserOffline(ctx, kickReq)
+		v.Close()
 		return utils.Wrap(err, "")
 	}
 	return errs.ErrInternalServer.Wrap()

pkg/discoveryregistry/dns.go

@@ -1,51 +0,0 @@
-package discoveryregistry
-
-// type DnsDiscoveryRegistry struct {
-// 	opts      []grpc.DialOption
-// 	namespace string
-// 	clientset *kubernetes.Clientset
-// }
-
-// func NewDnsDiscoveryRegistry(namespace string, opts []grpc.DialOption) (*DnsDiscoveryRegistry, error) {
-// 	config, err := rest.InClusterConfig()
-// 	if err != nil {
-// 		return nil, err
-// 	}
-// 	clientset, err := kubernetes.NewForConfig(config)
-// 	if err != nil {
-// 		return nil, err
-// 	}
-// 	return &DnsDiscoveryRegistry{
-// 		clientset: clientset,
-// 		namespace: namespace,
-// 		opts:      opts,
-// 	}, nil
-// }
-
-// func (d DnsDiscoveryRegistry) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) {
-// 	endpoints, err := d.clientset.CoreV1().Endpoints(d.namespace).Get(context.TODO(), serviceName, metav1.GetOptions{})
-// 	if err != nil {
-// 		return nil, err
-// 	}
-// 	var conns []*grpc.ClientConn
-// 	for _, subset := range endpoints.Subsets {
-// 		for _, address := range subset.Addresses {
-// 			for _, port := range subset.Ports {
-// 				conn, err := grpc.DialContext(ctx, net.JoinHostPort(address.IP, string(port.Port)), append(d.opts, opts...)...)
-// 				if err != nil {
-// 					return nil, err
-// 				}
-// 				conns = append(conns, conn)
-// 			}
-// 		}
-// 	}
-// 	return conns, nil
-// }
-
-// func (d DnsDiscoveryRegistry) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
-// 	return grpc.DialContext(ctx, fmt.Sprintf("%s.%s.svc.cluster.local", serviceName, d.namespace), append(d.opts, opts...)...)
-// }
-
-// func (d *DnsDiscoveryRegistry) AddOption(opts ...grpc.DialOption) {
-// 	d.opts = append(d.opts, opts...)
-// }

pkg/discoveryregistry/zookeeper/discover.go

@@ -22,7 +22,6 @@ func (s *ZkClient) watch(wg *sync.WaitGroup) {
 		switch event.Type {
 		case zk.EventSession:
 			s.logger.Printf("zk session event: %+v", event)
-		case zk.EventNodeCreated:
 		case zk.EventNodeChildrenChanged:
 			s.logger.Printf("zk event: %s", event.Path)
 			l := strings.Split(event.Path, "/")
@@ -37,6 +36,7 @@ func (s *ZkClient) watch(wg *sync.WaitGroup) {
 			}
 			s.logger.Printf("zk event handle success: %s", event.Path)
 		case zk.EventNodeDataChanged:
+		case zk.EventNodeCreated:
 		case zk.EventNodeDeleted:
 		case zk.EventNotWatching:
 		}
@@ -101,24 +101,6 @@ func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grp
 }
 
 func (s *ZkClient) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
-	conns, err := s.GetConns(ctx, serviceName, opts...)
-	if err != nil {
-		return nil, err
-	}
-	if len(conns) == 0 {
-		return nil, ErrConnIsNil
-	}
-	s.logger.Printf("get conn from conns, serviceName: %s", serviceName)
-	return s.getConnBalance(conns)
-}
-
-func (s *ZkClient) GetFirstConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
-	conns, err := s.GetConns(ctx, serviceName, opts...)
-	if err != nil {
-		return nil, err
-	}
-	if len(conns) == 0 {
-		return nil, ErrConnIsNil
-	}
-	return conns[0], nil
+	newOpts := append(s.options, grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, s.balancerName)))
+	return grpc.DialContext(ctx, fmt.Sprintf("%s:///%s", s.scheme, serviceName), append(newOpts, opts...)...)
 }

pkg/discoveryregistry/zookeeper/zk.go

@@ -113,7 +113,7 @@ func NewClient(zkServers []string, zkRoot string, options ...ZkOption) (*ZkClien
 		client.CloseZK()
 		return nil, err
 	}
-	// resolver.Register(client)
+	resolver.Register(client)
 	var wg sync.WaitGroup
 	go client.refresh(&wg)
 	go client.watch(&wg)