Adding the capability to remove a messageListener from a subscriber.

Author: Juan Ignacio Ubeira

rosjava/src/main/java/org/ros/concurrent/ListenerGroup.java

@@ -16,6 +16,7 @@
 
 package org.ros.concurrent;
 
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 
 import java.util.Collection;
@@ -107,11 +108,10 @@ public Collection<EventDispatcher<T>> addAll(Collection<T> listeners) {
    * @param listener the listener to remove
    * @return flag indicating successful removal
      */
-  public boolean remove(T listener)
-  {
+  public boolean remove(T listener) {
+    Preconditions.checkNotNull(listener);
     for (EventDispatcher<T> eventDispatcher : eventDispatchers) {
-      if(listener.equals(eventDispatcher.getListener()))
-      {
+      if (listener.equals(eventDispatcher.getListener())) {
         eventDispatcher.cancel();
         eventDispatchers.remove(eventDispatcher);
         return true;

rosjava/src/main/java/org/ros/internal/node/topic/DefaultSubscriber.java

@@ -139,6 +139,16 @@ public void addMessageListener(MessageListener<T> messageListener) {
     addMessageListener(messageListener, 1);
   }
 
+  @Override
+  public boolean removeMessageListener(MessageListener<T> messageListener) {
+    return incomingMessageQueue.removeListener(messageListener);
+  }
+
+  @Override
+  public void removeAllMessageListeners() {
+    incomingMessageQueue.removeAllListeners();
+  }
+
   @VisibleForTesting
   public void addPublisher(PublisherIdentifier publisherIdentifier, InetSocketAddress address) {
     synchronized (mutex) {

rosjava/src/main/java/org/ros/internal/transport/queue/IncomingMessageQueue.java

@@ -71,6 +71,20 @@ public void addListener(final MessageListener<T> messageListener, int queueCapac
     messageDispatcher.addListener(messageListener, queueCapacity);
   }
 
+  /**
+   * @see MessageDispatcher#removeListener(MessageListener)
+   */
+  public boolean removeListener(MessageListener<T> messageListener) {
+    return messageDispatcher.removeListener(messageListener);
+  }
+
+  /**
+   * @see MessageDispatcher#removeAllListeners()
+   */
+  public void removeAllListeners() {
+    messageDispatcher.removeAllListeners();
+  }
+
   public void shutdown() {
     messageDispatcher.cancel();
   }

rosjava/src/main/java/org/ros/internal/transport/queue/MessageDispatcher.java

@@ -79,6 +79,37 @@ public void addListener(MessageListener<T> messageListener, int limit) {
     }
   }
 
+  /**
+   * Removes the specified {@link MessageListener} from the internal
+   * {@link ListenerGroup}.
+   * @param messageListener {@link MessageListener} to remove.
+   * @return True if the listener was removed, false if it wasn't registered before.
+   *
+   * @see ListenerGroup#remove(Object)
+   */
+  public boolean removeListener(MessageListener<T> messageListener) {
+    if (DEBUG) {
+      log.info("Removing listener.");
+    }
+    synchronized (mutex) {
+      return messageListeners.remove(messageListener);
+    }
+  }
+
+  /**
+   * Removes all the registered {@link MessageListener}s.
+   *
+   * @see ListenerGroup#shutdown()
+   */
+  public void removeAllListeners() {
+    if (DEBUG) {
+      log.info("Removing all listeners.");
+    }
+    synchronized (mutex) {
+      messageListeners.shutdown();
+    }
+  }
+
   /**
    * Returns a newly allocated {@link SignalRunnable} for the specified
    * {@link LazyMessage}.

rosjava/src/main/java/org/ros/node/topic/Subscriber.java

@@ -61,6 +61,18 @@ public interface Subscriber<T> extends TopicParticipant {
    */
   void addMessageListener(MessageListener<T> messageListener);
 
+  /**
+   * Removes a previously added {@link MessageListener}.
+   * @param messageListener {@link MessageListener} to remove.
+   * @return True if the listener was removed, false if it wasn't registered before.
+   */
+  boolean removeMessageListener(MessageListener<T> messageListener);
+
+  /**
+   * Removes all registered {@link MessageListener}s.
+   */
+  void removeAllMessageListeners();
+
   /**
    * Shuts down and unregisters the {@link Subscriber}. using the default
    * timeout Shutdown is delayed by at most the specified timeout to allow