fix: init grpc conn multiple in one process will cause zero address when rpc is called (#604)

* new feature: add batch send msg

* new feature: add batch send msg

* new feature: add batch send msg

* new feature: add batch send msg

* new feature: add batch send msg

* new feature: add batch send msg

* fix bug: multiple gateway kick user

* fix bug: multiple gateway kick user

* fix bug: multiple gateway kick user

* fix bug: multiple gateway kick user

* fix bug: multiple gateway kick user

* MsgDestructTime

* fix bug: msg destruct sql

* fix bug: msg destruct

* fix bug: msg destruct

* fix bug: msg destruct sql

* fix bug: msg destruct sql

* fix bug: msg destruct sql

* fix bug: msg destruct sql

* debug: print stack

* debug: print stack

* debug: print stack

* fix bug: msg destruct sql

Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com>

* fix bug: msg notification self 2 self push twice

* fix bug: heartbeat get self notification

* fix bug: init grpc conn in one process

* fix bug: grpc conn

Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com>

* fix bug: zk client recreate node when reconn

* fix bug: set friend mark args error

---------

Signed-off-by: wangchuxiao <wangchuxiao97@outlook.com>
pull/632/head
WangchuXiao 1 year ago committed by GitHub
parent 4629b03272
commit 7ee40bd485
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -40,7 +40,17 @@ func (s *ZkClient) watch() {
event := <-s.eventChan
switch event.Type {
case zk.EventSession:
if event.State == zk.StateHasSession && s.isRegistered {
s.logger.Printf("zk session event stateHasSession: %+v, client prepare to create new temp node", event)
node, err := s.CreateTempNode(s.rpcRegisterName, s.rpcRegisterAddr)
if err != nil {
s.logger.Printf("zk session event stateHasSession: %+v, create temp node error: %v", event, err)
} else {
s.node = node
}
} else {
s.logger.Printf("zk session event: %+v", event)
}
case zk.EventNodeChildrenChanged:
s.logger.Printf("zk event: %s", event.Path)
l := strings.Split(event.Path, "/")

@ -30,6 +30,14 @@ func (s *ZkClient) CreateRpcRootNodes(serviceNames []string) error {
return nil
}
func (s *ZkClient) CreateTempNode(rpcRegisterName, addr string) (node string, err error) {
return s.conn.CreateProtectedEphemeralSequential(
s.getPath(rpcRegisterName)+"/"+addr+"_",
[]byte(addr),
zk.WorldACL(zk.PermAll),
)
}
func (s *ZkClient) Register(rpcRegisterName, host string, port int, opts ...grpc.DialOption) error {
if err := s.ensureName(rpcRegisterName); err != nil {
return err
@ -39,15 +47,14 @@ func (s *ZkClient) Register(rpcRegisterName, host string, port int, opts ...grpc
if err != nil {
return err
}
node, err := s.conn.CreateProtectedEphemeralSequential(
s.getPath(rpcRegisterName)+"/"+addr+"_",
[]byte(addr),
zk.WorldACL(zk.PermAll),
)
node, err := s.CreateTempNode(rpcRegisterName, addr)
if err != nil {
return err
}
s.rpcRegisterName = rpcRegisterName
s.rpcRegisterAddr = addr
s.node = node
s.isRegistered = true
return nil
}
@ -60,6 +67,9 @@ func (s *ZkClient) UnRegister() error {
}
time.Sleep(time.Second)
s.node = ""
s.rpcRegisterName = ""
s.rpcRegisterAddr = ""
s.isRegistered = false
s.localConns = make(map[string][]grpc.ClientConnInterface)
s.resolvers = make(map[string]*Resolver)
return nil

@ -73,7 +73,11 @@ func (s *ZkClient) Build(
opts resolver.BuildOptions,
) (resolver.Resolver, error) {
s.logger.Printf("build resolver: %+v, cc: %+v", target, cc.UpdateState)
// log.ZDebug(context.Background(), "build resolver start", "target", target, "cc", cc.UpdateState)
serviceName := strings.TrimLeft(target.URL.Path, "/")
if oldResolver, ok := s.resolvers[serviceName]; ok {
s.logger.Printf("rpc resolver exist: %+v, cc: %+v, key: %s", target, cc.UpdateState, serviceName)
return oldResolver, nil
}
r := &Resolver{}
r.target = target
r.cc = cc
@ -81,11 +85,8 @@ func (s *ZkClient) Build(
r.ResolveNowZK(resolver.ResolveNowOptions{})
s.lock.Lock()
defer s.lock.Unlock()
serviceName := strings.TrimLeft(target.URL.Path, "/")
s.resolvers[serviceName] = r
s.logger.Printf("build resolver finished: %+v, cc: %+v, key: %s", target, cc.UpdateState, serviceName)
// log.ZDebug(context.Background(), "build resolver finished", "target", target, "cc", cc.UpdateState,
// "serviceName", serviceName)
return r, nil
}

@ -40,6 +40,9 @@ type ZkClient struct {
userName string
password string
rpcRegisterName string
rpcRegisterAddr string
isRegistered bool
scheme string
timeout int
@ -136,6 +139,7 @@ func NewClient(zkServers []string, zkRoot string, options ...ZkOption) (*ZkClien
resolver.Register(client)
go client.refresh()
go client.watch()
time.Sleep(time.Millisecond * 50)
return client, nil
}

@ -152,9 +152,6 @@ func (x *SetFriendRemarkReq) Check() error {
if x.FriendUserID == "" {
return errs.ErrArgs.Wrap("fromUserID is empty")
}
if x.Remark == "" {
return errs.ErrArgs.Wrap("remark is empty")
}
return nil
}

Loading…
Cancel
Save