Fixed subscriber shutdown.

Author: neocoretechs

src/main/java/org/ros/internal/node/client/Registrar.java

@@ -37,7 +37,7 @@
  * which creates a SlaveClient of type SlaveRpcEndpointImpl to
  * contact the publishers. If successful, call signalOnMasterRegistrationSuccess for the subscriber.
  * 
- * @author jg
+ * @author Jonathan Groff Copyright (C) NeoCoreTechs 2015,2021
  */
 public class Registrar implements TopicParticipantManagerListener, ServiceManagerListener {
 
@@ -252,7 +252,15 @@ public void run() {
       });
     }
   }
-
+  
+  @Override
+  public void onSubscriberRemoved(final DefaultSubscriber<?> subscriber, boolean remove) {
+	   if (DEBUG) {
+		      log.info("Unregistering subscriber for shutdown: " + subscriber);
+	   }
+	   masterClient.unregisterSubscriber(nodeIdentifier, subscriber);
+  }
+  
   @Override
   public void onServiceServerAdded(final DefaultServiceServer<?, ?> serviceServer) {
     if (DEBUG) {

src/main/java/org/ros/internal/node/topic/SubscriberFactory.java

@@ -58,7 +58,7 @@ public void onNewPublisher(Subscriber<T> subscriber, PublisherIdentifier publish
           @Override
           public void onShutdown(Subscriber<T> subscriber) {
         	assert(subscriber != null );
-            topicParticipantManager.removeSubscriber((DefaultSubscriber<T>) subscriber);
+            topicParticipantManager.removeSubscriber((DefaultSubscriber<T>) subscriber, true);
           }
         });
         topicParticipantManager.addSubscriber(subscriber);

src/main/java/org/ros/internal/node/topic/TopicParticipantManager.java

@@ -8,7 +8,6 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -104,7 +103,16 @@ public void removeSubscriber(DefaultSubscriber<?> subscriber) {
       listener.onSubscriberRemoved(subscriber);
     }
   }
-
+  /**
+   * Variation for removal of subscriber during executor service shutdown.<p/>
+   * Dont fire onSubscriberRemoved method of Registrar, which resubmits 
+   * @param subscriber
+   * @param b
+   */
+  public void removeSubscriber(DefaultSubscriber<?> subscriber, boolean b) {
+	  subscribers.remove(subscriber.getTopicName());
+  }
+  
   public void addSubscriberConnection(DefaultSubscriber<?> subscriber, PublisherIdentifier publisherIdentifier) {
 	  if(DEBUG)
 		  log.info("Connecting subscriber:"+subscriber+" to publisher Identifier:"+publisherIdentifier+" for "+this);
@@ -163,4 +171,6 @@ public Collection<DefaultPublisher<?>> getPublishers() {
   public Collection<SubscriberIdentifier> getPublisherConnections(DefaultPublisher<?> publisher) {
     return publisherConnections.get(publisher);
   }
+
+
 }

src/main/java/org/ros/internal/node/topic/TopicParticipantManagerListener.java

@@ -37,8 +37,14 @@ public interface TopicParticipantManagerListener {
   /**
    * Called when a {@link Subscriber} is removed.
    * 
-   * @param subscriber
-   *          the {@link Subscriber} that was removed
+   * @param subscriber the {@link Subscriber} that was removed
    */
   void onSubscriberRemoved(DefaultSubscriber<?> subscriber);
+
+  /**
+   * Called when a {@link Subscriber} is removed for shutdown.
+   * 
+   * @param subscriber the {@link Subscriber} that was removed
+   */
+  void onSubscriberRemoved(DefaultSubscriber<?> subscriber, boolean remove);
 }