Feat/direct notify

cluster/router/chain/chain.go

@@ -269,7 +269,7 @@ func NewRouterChain(url *common.URL) (*RouterChain, error) {
 // rule doesn't change), and the address list doesn't change, then the existing data will be re-used.
 func poolRouter(p router.Poolable, origin *InvokerCache, invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
 	name := p.Name()
-	if isCacheMiss(origin, name) || p.ShouldPool() || &(origin.invokers) != &invokers {
+	if isCacheMiss(origin, name) || &(origin.invokers) != &invokers {
 		logger.Debugf("build address cache for router %q", name)
 		return p.Pool(invokers)
 	}

cluster/router/healthcheck/health_check_route.go

@@ -91,11 +91,6 @@ func (r *HealthCheckRouter) Pool(invokers []protocol.Invoker) (router.AddrPool,
 	return rb, nil
 }
 
-// ShouldPool will always return true to make sure healthy check constantly.
-func (r *HealthCheckRouter) ShouldPool() bool {
-	return r.enabled
-}
-
 func (r *HealthCheckRouter) Name() string {
 	return name
 }

cluster/router/router.go

@@ -61,10 +61,6 @@ type Poolable interface {
 	// Pool created address pool and address metadata from the invokers.
 	Pool([]protocol.Invoker) (AddrPool, AddrMetadata)
 
-	// ShouldPool returns if it should pool. One typical scenario is a router rule changes, in this case, a pooling
-	// is necessary, even if the addresses not changed at all.
-	ShouldPool() bool
-
 	// Name return the Poolable's name.
 	Name() string
 }

cluster/router/tag/tag_router.go

@@ -74,7 +74,6 @@ type tagRouter struct {
 	enabled       bool
 	priority      int64
 	application   string
-	ruleChanged   bool
 	mutex         sync.RWMutex
 }
 
@@ -190,7 +189,6 @@ func (c *tagRouter) Process(event *config_center.ConfigChangeEvent) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 	c.tagRouterRule = routerRule
-	c.ruleChanged = true
 }
 
 // URL gets the url of tagRouter
@@ -213,7 +211,6 @@ func (c *tagRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.A
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 	poolWithDynamicTag(invokers, c.tagRouterRule, rb)
-	c.ruleChanged = false
 	// create metadata in order to avoid lock in route()
 	meta := addrMetadata{application: c.application}
 	if c.tagRouterRule != nil {
@@ -268,13 +265,6 @@ func (c *tagRouter) fetchRuleIfNecessary(invokers []protocol.Invoker) {
 	}
 }
 
-// ShouldPool returns false, to make sure address cache for tag router happens once and only once.
-func (c *tagRouter) ShouldPool() bool {
-	c.mutex.RLock()
-	defer c.mutex.RUnlock()
-	return c.ruleChanged
-}
-
 // Name returns pool's name
 func (c *tagRouter) Name() string {
 	return name

common/dubbo_dir_refresh.go

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 common
+
+import "sync"
+
+// DirMap is used by dubbo-getty, it can notify delete event to dubbo directory directly by getty session
+// look up at:
+// remoting/getty/listener.go
+// registry/directory/directory.go
+var DirMap sync.Map

registry/directory/directory.go

@@ -69,7 +69,7 @@ type RegistryDirectory struct {
 }
 
 // NewRegistryDirectory will create a new RegistryDirectory
-func NewRegistryDirectory(url *common.URL, registry registry.Registry) (cluster.Directory, error) {
+func NewRegistryDirectory(url *common.URL, reg registry.Registry) (cluster.Directory, error) {
 	if url.SubURL == nil {
 		return nil, perrors.Errorf("url is invalid, suburl can not be nil")
 	}
@@ -79,7 +79,7 @@ func NewRegistryDirectory(url *common.URL, registry registry.Registry) (cluster.
 		cacheInvokers:    []protocol.Invoker{},
 		cacheInvokersMap: &sync.Map{},
 		serviceType:      url.SubURL.Service(),
-		registry:         registry,
+		registry:         reg,
 	}
 
 	dir.consumerURL = dir.getConsumerUrl(url.SubURL)
@@ -233,6 +233,20 @@ func (dir *RegistryDirectory) setNewInvokers() {
 	dir.RouterChain().SetInvokers(newInvokers)
 }
 
+// AddToDirMap is called after receive Add event
+// this function add a way to delete service in directory by getty-session, not only by registry.
+// the way of using getty-session to delete is faster than registry to delete
+// this function register the delete function to common sync.Map, and it read by getty-session, to push url
+func (dir *RegistryDirectory) AddToDirMap(url *common.URL) {
+	common.DirMap.Store(url.Key(), func(url *common.URL) {
+		eve := &registry.ServiceEvent{
+			Service: url,
+			Action:  remoting.EventTypeDel,
+		}
+		dir.Notify(eve)
+	})
+}
+
 // cacheInvokerByEvent caches invokers from the service event
 func (dir *RegistryDirectory) cacheInvokerByEvent(event *registry.ServiceEvent) (protocol.Invoker, error) {
 	// judge is override or others
@@ -243,6 +257,7 @@ func (dir *RegistryDirectory) cacheInvokerByEvent(event *registry.ServiceEvent)
 			logger.Infof("selector add service url{%s}", event.Service)
 			// FIXME: routers are built in every address notification?
 			dir.configRouters()
+			dir.AddToDirMap(event.Service)
 			return dir.cacheInvoker(u), nil
 		case remoting.EventTypeDel:
 			logger.Infof("selector delete service url{%s}", event.Service)

remoting/getty/getty_client.go

@@ -146,7 +146,7 @@ func (c *Client) Connect(url *common.URL) error {
 	initClient(url.Protocol)
 	c.conf = *clientConf
 	// new client
-	c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL))
+	c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL), url)
 	c.pool.sslEnabled = url.GetParamBool(constant.SSL_ENABLED_KEY, false)
 
 	// codec

remoting/getty/getty_client_test.go

@@ -94,7 +94,7 @@ func getClient(url *common.URL) *Client {
 }
 
 func testClient_Call(t *testing.T, svr *Server, url *common.URL, c *Client) {
-	c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL))
+	c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL), &common.URL{})
 
 	testGetBigPkg(t, c)
 	testGetUser(t, c)

remoting/getty/listener.go

@@ -31,6 +31,7 @@ import (
 )
 
 import (
+	"github.com/apache/dubbo-go/common"
 	"github.com/apache/dubbo-go/common/constant"
 	"github.com/apache/dubbo-go/common/logger"
 	"github.com/apache/dubbo-go/protocol/invocation"
@@ -69,11 +70,15 @@ func (s *rpcSession) GetReqNum() int32 {
 type RpcClientHandler struct {
 	conn         *gettyRPCClient
 	timeoutTimes int
+	url          *common.URL
 }
 
 // nolint
-func NewRpcClientHandler(client *gettyRPCClient) *RpcClientHandler {
-	return &RpcClientHandler{conn: client}
+func NewRpcClientHandler(client *gettyRPCClient, url *common.URL) *RpcClientHandler {
+	return &RpcClientHandler{
+		conn: client,
+		url:  url,
+	}
 }
 
 // OnOpen call the getty client session opened, add the session to getty client session list
@@ -85,15 +90,25 @@ func (h *RpcClientHandler) OnOpen(session getty.Session) error {
 // OnError the getty client session has errored, so remove the session from the getty client session list
 func (h *RpcClientHandler) OnError(session getty.Session, err error) {
 	logger.Infof("session{%s} got error{%v}, will be closed.", session.Stat(), err)
+	h.tryDirectDelete()
 	h.conn.removeSession(session)
 }
 
 // OnClose close the session, remove it from the getty session list
 func (h *RpcClientHandler) OnClose(session getty.Session) {
 	logger.Infof("session{%s} is closing......", session.Stat())
+	h.tryDirectDelete()
 	h.conn.removeSession(session)
 }
 
+func (h *RpcClientHandler) tryDirectDelete() {
+	f, ok := common.DirMap.Load(h.url.Key())
+	if !ok {
+		return
+	}
+	f.(func(url *common.URL))(h.url)
+}
+
 // OnMessage get response from getty server, and update the session to the getty client session list
 func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) {
 	result, ok := pkg.(remoting.DecodeResult)

remoting/getty/pool.go

@@ -33,6 +33,7 @@ import (
 )
 
 import (
+	"github.com/apache/dubbo-go/common"
 	"github.com/apache/dubbo-go/common/logger"
 	"github.com/apache/dubbo-go/config"
 )
@@ -54,7 +55,7 @@ var (
 	errClientPoolClosed = perrors.New("client pool closed")
 )
 
-func newGettyRPCClientConn(pool *gettyRPCClientPool, addr string) (*gettyRPCClient, error) {
+func newGettyRPCClientConn(pool *gettyRPCClientPool, addr string, url *common.URL) (*gettyRPCClient, error) {
 	var (
 		gettyClient getty.Client
 		sslEnabled  bool
@@ -134,7 +135,7 @@ func (c *gettyRPCClient) newSession(session getty.Session) error {
 		session.SetName(conf.GettySessionParam.SessionName)
 		session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
 		session.SetPkgHandler(NewRpcClientPackageHandler(c.pool.rpcClient))
-		session.SetEventListener(NewRpcClientHandler(c))
+		session.SetEventListener(NewRpcClientHandler(c, c.pool.url))
 		session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
 		session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
 		session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6))
@@ -167,7 +168,7 @@ func (c *gettyRPCClient) newSession(session getty.Session) error {
 	session.SetName(conf.GettySessionParam.SessionName)
 	session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
 	session.SetPkgHandler(NewRpcClientPackageHandler(c.pool.rpcClient))
-	session.SetEventListener(NewRpcClientHandler(c))
+	session.SetEventListener(NewRpcClientHandler(c, c.pool.url))
 	session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
 	session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
 	session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6))
@@ -333,15 +334,17 @@ type gettyRPCClientPool struct {
 
 	sync.Mutex
 	conns []*gettyRPCClient
+	url   *common.URL
 }
 
-func newGettyRPCClientConnPool(rpcClient *Client, size int, ttl time.Duration) *gettyRPCClientPool {
+func newGettyRPCClientConnPool(rpcClient *Client, size int, ttl time.Duration, url *common.URL) *gettyRPCClientPool {
 	return &gettyRPCClientPool{
 		rpcClient: rpcClient,
 		size:      size,
 		ttl:       int64(ttl.Seconds()),
 		// init capacity : 2
 		conns: make([]*gettyRPCClient, 0, 2),
+		url:   url,
 	}
 }
 
@@ -359,7 +362,7 @@ func (p *gettyRPCClientPool) getGettyRpcClient(addr string) (*gettyRPCClient, er
 	conn, connErr := p.get()
 	if connErr == nil && conn == nil {
 		// create new conn
-		rpcClientConn, rpcErr := newGettyRPCClientConn(p, addr)
+		rpcClientConn, rpcErr := newGettyRPCClientConn(p, addr, p.url)
 		if rpcErr == nil {
 			p.put(rpcClientConn)
 		}

remoting/getty/pool_test.go

@@ -26,10 +26,14 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+import (
+	"github.com/apache/dubbo-go/common"
+)
+
 func TestGetConnFromPool(t *testing.T) {
 	var rpcClient Client
 
-	clientPoll := newGettyRPCClientConnPool(&rpcClient, 1, time.Duration(5*time.Second))
+	clientPoll := newGettyRPCClientConnPool(&rpcClient, 1, time.Duration(5*time.Second), &common.URL{})
 
 	var conn1 gettyRPCClient
 	conn1.active = time.Now().Unix()