nacos registry handleServiceEvents is a blocking method, fix multiple service name not subscribe as expect

Author: FoghostCn

registry/nacos/registry.go

@@ -161,62 +161,39 @@ func (nr *nacosRegistry) UnRegister(url *common.URL) error {
 
 // Subscribe returns nil if subscribing registry successfully. If not returns an error.
 func (nr *nacosRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) error {
-	// TODO
 	role, _ := strconv.Atoi(url.GetParam(constant.RegistryRoleKey, ""))
 	if role != common.CONSUMER {
 		return nil
 	}
 	serviceName := url.GetParam(constant.InterfaceKey, "")
-	if serviceName == "*" {
-		// Subscribe to all services
-		for {
-			if !nr.IsAvailable() {
-				logger.Warnf("event listener game over.")
-				return perrors.New("nacosRegistry is not available.")
-			}
-
-			services, err := nr.getAllSubscribeServiceNames(url)
-			if err != nil {
-				if !nr.IsAvailable() {
-					logger.Warnf("event listener game over.")
-					return err
-				}
-				logger.Warnf("getAllServices() = err:%v", perrors.WithStack(err))
-				time.Sleep(time.Duration(RegistryConnDelay) * time.Second)
-				continue
-			}
-
-			for _, service := range services {
-				listener, err := NewNacosListenerWithServiceName(service, nr.URL, nr.namingClient)
-				metrics.Publish(metricsRegistry.NewSubscribeEvent(err == nil))
-				if err != nil {
-					logger.Warnf("Failed to subscribe to service '%s': %v", service, err)
-					continue
-				}
-
-				nr.handleServiceEvents(listener, notifyListener)
-			}
-		}
+	var serviceNames []string
+	var err error
+	if serviceName == constant.AnyValue {
+		serviceNames, err = nr.getAllSubscribeServiceNames(url)
 	} else {
-		// Subscribe to a specific service
-		for {
-			if !nr.IsAvailable() {
-				logger.Warnf("event listener game over.")
-				return perrors.New("nacosRegistry is not available.")
-			}
+		serviceNames = []string{getSubscribeName(url)}
+	}
+	err = nr.subscribe(serviceNames, notifyListener)
+	return err
+}
 
-			listener, err := NewNacosListenerWithServiceName(getSubscribeName(url), nr.URL, nr.namingClient)
+// subscribe subscribe services
+func (nr *nacosRegistry) subscribe(serviceNames []string, notifyListener registry.NotifyListener) error {
+	for {
+		// Subscribe to a specific service
+		if !nr.IsAvailable() {
+			logger.Warnf("event listener game over.")
+			return perrors.New("nacosRegistry is not available.")
+		}
+		for _, serviceName := range serviceNames {
+			listener, err := NewNacosListenerWithServiceName(serviceName, nr.URL, nr.namingClient)
 			metrics.Publish(metricsRegistry.NewSubscribeEvent(err == nil))
 			if err != nil {
-				if !nr.IsAvailable() {
-					logger.Warnf("event listener game over.")
-					return err
-				}
-				logger.Warnf("getListener() = err:%v", perrors.WithStack(err))
+				logger.Warnf("getAllServices() = err:%v", perrors.WithStack(err))
 				time.Sleep(time.Duration(RegistryConnDelay) * time.Second)
-				continue
+				break
 			}
-			nr.handleServiceEvents(listener, notifyListener)
+			go nr.handleServiceEvents(listener, notifyListener)
 		}
 	}
 }