Autoremove EventHandler from parent

Signed-off-by: Ivan Santiago Paunovic <ivanpauno@ekumenlabs.com>

Author: ivanpauno

rcljava/src/main/java/org/ros2/rcljava/events/EventHandlerImpl.java

@@ -20,6 +20,7 @@
 
 import org.ros2.rcljava.common.JNIUtils;
 import org.ros2.rcljava.consumers.Consumer;
+import org.ros2.rcljava.consumers.Consumer;
 import org.ros2.rcljava.events.EventHandler;
 import org.ros2.rcljava.events.EventStatus;
 import org.ros2.rcljava.interfaces.Disposable;
@@ -64,16 +65,19 @@ public class EventHandlerImpl<
    * @param eventStatusFactory Factory of an event status.
    * @param callback The callback function that will be called when the event
    *     is triggered.
+   * @param removeCallback Callback that will be called when this object is being disposed.
    */
   public EventHandlerImpl(
       final WeakReference<ParentT> parentReference,
       final long handle,
       final Supplier<T> eventStatusFactory,
-      final Consumer<T> callback) {
+      final Consumer<T> callback,
+      final Consumer<EventHandler> removeCallback) {
     this.parentReference = parentReference;
     this.handle = handle;
     this.eventStatusFactory = eventStatusFactory;
     this.callback = callback;
+    this.removeCallback = removeCallback;
   }
 
   /**
@@ -102,9 +106,11 @@ public final long getHandle() {
    * {@inheritDoc}
    */
   public synchronized final void dispose() {
-    if (this.handle != 0) {
-      nativeDispose(this.handle);
+    if (this.handle == 0) {
+      return;
     }
+    this.removeCallback.accept(this);
+    nativeDispose(this.handle);
     this.handle = 0;
   }
 
@@ -126,11 +132,12 @@ public synchronized final void executeCallback() {
     nativeTake(this.handle, nativeEventStatusHandle);
     eventStatus.fromRCLEvent(nativeEventStatusHandle);
     eventStatus.deallocateRCLStatusEvent(nativeEventStatusHandle);
-    callback.accept(eventStatus);
+    this.callback.accept(eventStatus);
   }
 
   private final Supplier<T> eventStatusFactory;
   private final WeakReference<ParentT> parentReference;
   private long handle;
   private final Consumer<T> callback;
+  private final Consumer<EventHandler> removeCallback;
 }

rcljava/src/main/java/org/ros2/rcljava/publisher/PublisherImpl.java

@@ -117,10 +117,20 @@ public final WeakReference<Node> getNodeReference() {
   public final
   <T extends PublisherEventStatus> EventHandler<T, Publisher>
   createEventHandler(Supplier<T> factory, Consumer<T> callback) {
+    final WeakReference<Collection<EventHandler>> weakEventHandlers = new WeakReference(
+      this.eventHandlers);
+    Consumer<EventHandler> removeCallback = new Consumer<EventHandler>() {
+      public void accept(EventHandler eventHandler) {
+        Collection<EventHandler> eventHandlers = weakEventHandlers.get();
+        if (eventHandlers != null) {
+          eventHandlers.remove(eventHandler);
+        }
+      }
+    };
     T status = factory.get();
     long eventHandle = nativeCreateEvent(this.handle, status.getPublisherEventType());
     EventHandler<T, Publisher> eventHandler = new EventHandlerImpl(
-      new WeakReference<Publisher>(this), eventHandle, factory, callback);
+      new WeakReference<Publisher>(this), eventHandle, factory, callback, removeCallback);
     this.eventHandlers.add(eventHandler);
     return eventHandler;
   }

rcljava/src/main/java/org/ros2/rcljava/subscription/SubscriptionImpl.java

@@ -22,6 +22,7 @@
 
 import org.ros2.rcljava.RCLJava;
 import org.ros2.rcljava.common.JNIUtils;
+import org.ros2.rcljava.consumers.BiConsumer;
 import org.ros2.rcljava.consumers.Consumer;
 import org.ros2.rcljava.events.EventHandler;
 import org.ros2.rcljava.events.EventHandlerImpl;
@@ -128,10 +129,20 @@ public final WeakReference<Node> getNodeReference() {
   public final
   <T extends SubscriptionEventStatus> EventHandler<T, Subscription>
   createEventHandler(Supplier<T> factory, Consumer<T> callback) {
+    final WeakReference<Collection<EventHandler>> weakEventHandlers = new WeakReference(
+      this.eventHandlers);
+    Consumer<EventHandler> removeCallback = new Consumer<EventHandler>() {
+      public void accept(EventHandler eventHandler) {
+        Collection<EventHandler> eventHandlers = weakEventHandlers.get();
+        if (eventHandlers != null) {
+          eventHandlers.remove(eventHandler);
+        }
+      }
+    };
     T status = factory.get();
     long eventHandle = nativeCreateEvent(this.handle, status.getSubscriptionEventType());
     EventHandler<T, Subscription> eventHandler = new EventHandlerImpl(
-      new WeakReference<Subscription>(this), eventHandle, factory, callback);
+      new WeakReference<Subscription>(this), eventHandle, factory, callback, removeCallback);
     this.eventHandlers.add(eventHandler);
     return eventHandler;
   }

rcljava/src/test/java/org/ros2/rcljava/publisher/PublisherTest.java

@@ -90,9 +90,12 @@ public void accept(final OfferedQosIncompatible status) {
       }
     );
     assertNotEquals(0, eventHandler.getHandle());
+    assertEquals(1, publisher.getEventHandlers().size());
     // force executing the callback, so we check that taking an event works
     eventHandler.executeCallback();
-    RCLJava.shutdown();
+    eventHandler.dispose();
     assertEquals(0, eventHandler.getHandle());
+    assertEquals(0, publisher.getEventHandlers().size());
+    RCLJava.shutdown();
   }
 }