diff --git a/internal/api/auth.go b/internal/api/auth.go index 22e30b923..458a81eb8 100644 --- a/internal/api/auth.go +++ b/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) } diff --git a/internal/push/push_to_client.go b/internal/push/push_to_client.go index 7a307c91b..c1f6b2b55 100644 --- a/internal/push/push_to_client.go +++ b/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 } diff --git a/internal/rpc/auth/auth.go b/internal/rpc/auth/auth.go index fe5dfd21e..b2ff3fea2 100644 --- a/internal/rpc/auth/auth.go +++ b/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() diff --git a/pkg/discoveryregistry/dns.go b/pkg/discoveryregistry/dns.go deleted file mode 100644 index bc27dcb2a..000000000 --- a/pkg/discoveryregistry/dns.go +++ /dev/null @@ -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...) -// } diff --git a/pkg/discoveryregistry/zookeeper/discover.go b/pkg/discoveryregistry/zookeeper/discover.go index 8a9045a63..9103b884a 100644 --- a/pkg/discoveryregistry/zookeeper/discover.go +++ b/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...)...) } diff --git a/pkg/discoveryregistry/zookeeper/zk.go b/pkg/discoveryregistry/zookeeper/zk.go index cd09e028e..0a665e5bf 100644 --- a/pkg/discoveryregistry/zookeeper/zk.go +++ b/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)