Subscribe error handling (#2785)

* [fix] etcd config source prefix issue (#2389)

* http transport data race issue (#2436)

* [fix] #2431 http transport data race issue

* [feature] Ability to close connection while receiving.
Ability to send messages while receiving.
Icreased r channel limit to 100 to more fluently communication.
Do not dropp sent request if r channel is full.

* [feature] always subscribes to all topics and if there is an error in one of them, the unsuccessful topics will be subscribed to again

---------

Co-authored-by: Johnson C <chengqiaosheng@gmail.com>

server/rpc_events.go

@@ -101,7 +101,7 @@ func (s *rpcServer) Subscribe(sb Subscriber) error {
 
 // subscribeServer will subscribe the server to the topic with its own name.
 func (s *rpcServer) subscribeServer(config Options) error {
-	if s.opts.Router != nil {
+	if s.opts.Router != nil && s.subscriber == nil {
 		sub, err := s.opts.Broker.Subscribe(config.Name, s.HandleEvent)
 		if err != nil {
 			return err
@@ -115,8 +115,11 @@ func (s *rpcServer) subscribeServer(config Options) error {
 }
 
 // reSubscribe itterates over subscribers and re-subscribes then.
-func (s *rpcServer) reSubscribe(config Options) error {
+func (s *rpcServer) reSubscribe(config Options) {
 	for sb := range s.subscribers {
+		if s.subscribers[sb] != nil {
+			continue
+		}
 		var opts []broker.SubscribeOption
 		if queue := sb.Options().Queue; len(queue) > 0 {
 			opts = append(opts, broker.Queue(queue))
@@ -133,12 +136,15 @@ func (s *rpcServer) reSubscribe(config Options) error {
 		config.Logger.Logf(log.InfoLevel, "Subscribing to topic: %s", sb.Topic())
 		sub, err := config.Broker.Subscribe(sb.Topic(), s.HandleEvent, opts...)
 		if err != nil {
-			return err
+			config.Logger.Logf(log.WarnLevel, "Unable to subscribing to topic: %s, error: %s", sb.Topic(), err)
+			continue
+		}
+		err = s.router.Subscribe(sb)
+		if err != nil {
+			config.Logger.Logf(log.WarnLevel, "Unable to subscribing to topic: %s, error: %s", sb.Topic(), err)
+			sub.Unsubscribe()
+			continue
 		}
-
 		s.subscribers[sb] = []broker.Subscriber{sub}
-		s.router.Subscribe(sb)
 	}
-
-	return nil
 }

server/rpc_server.go

@@ -446,17 +446,6 @@ func (s *rpcServer) Register() error {
 	// Set what we're advertising
 	s.opts.Advertise = addr
 
-	// Router can exchange messages on broker
-	// Subscribe to the topic with its own name
-	if err := s.subscribeServer(config); err != nil {
-		return errors.Wrap(err, "failed to subscribe to service name topic")
-	}
-
-	// Subscribe for all of the subscribers
-	if err := s.reSubscribe(config); err != nil {
-		return errors.Wrap(err, "failed to resubscribe")
-	}
-
 	return nil
 }
 
@@ -606,19 +595,29 @@ func (s *rpcServer) newRegFuc(config Options) func(service *registry.Service) er
 		// Attempt to register. If registration fails, back off and try again.
 		// TODO: see if we can improve the retry mechanism. Maybe retry lib, maybe config values
 		for i := 0; i < 3; i++ {
-			if err := config.Registry.Register(service, rOpts...); err != nil {
-				regErr = err
-
+			if regErr = config.Registry.Register(service, rOpts...); regErr != nil {
 				time.Sleep(backoff.Do(i + 1))
-
 				continue
 			}
-
-			return nil
+			break
 		}
 
+		if regErr != nil {
 			return regErr
 		}
+
+		s.Lock()
+		defer s.Unlock()
+		// Router can exchange messages on broker
+		// Subscribe to the topic with its own name
+		if err := s.subscribeServer(config); err != nil {
+			return errors.Wrap(err, "failed to subscribe to service name topic")
+		}
+		// Subscribe for all of the subscribers
+		s.reSubscribe(config)
+
+		return nil
+	}
 }
 
 // getAddr will take the advertise or service address, and return it.