Add remove pending request method (#81)

Signed-off-by: Ivan Santiago Paunovic <[email protected]>

Author: ivanpauno

rcljava/CMakeLists.txt

@@ -143,6 +143,7 @@ set(${PROJECT_NAME}_sources
   "src/main/java/org/ros2/rcljava/action/GoalStatus.java"
   "src/main/java/org/ros2/rcljava/client/Client.java"
   "src/main/java/org/ros2/rcljava/client/ClientImpl.java"
+  "src/main/java/org/ros2/rcljava/client/ResponseFuture.java"
   "src/main/java/org/ros2/rcljava/concurrent/Callback.java"
   "src/main/java/org/ros2/rcljava/concurrent/RCLFuture.java"
   "src/main/java/org/ros2/rcljava/contexts/Context.java"

rcljava/src/main/java/org/ros2/rcljava/client/Client.java

@@ -30,11 +30,13 @@ public interface Client<T extends ServiceDefinition> extends Disposable {
 
   <U extends MessageDefinition> void handleResponse(RMWRequestId header, U response);
 
-  <U extends MessageDefinition, V extends MessageDefinition> Future<V> asyncSendRequest(
+  <U extends MessageDefinition, V extends MessageDefinition> ResponseFuture<V> asyncSendRequest(
       final U request);
 
-  <U extends MessageDefinition, V extends MessageDefinition> Future<V> asyncSendRequest(
+  <U extends MessageDefinition, V extends MessageDefinition> ResponseFuture<V> asyncSendRequest(
       final U request, final Consumer<Future<V>> callback);
+  
+  <V extends MessageDefinition> boolean removePendingRequest(ResponseFuture<V> future);
 
   /**
    * Check if the service server is available.

rcljava/src/main/java/org/ros2/rcljava/client/ClientImpl.java

@@ -28,7 +28,6 @@
 
 import org.ros2.rcljava.RCLJava;
 import org.ros2.rcljava.common.JNIUtils;
-import org.ros2.rcljava.concurrent.RCLFuture;
 import org.ros2.rcljava.consumers.Consumer;
 import org.ros2.rcljava.interfaces.MessageDefinition;
 import org.ros2.rcljava.interfaces.ServiceDefinition;
@@ -53,7 +52,7 @@ public class ClientImpl<T extends ServiceDefinition> implements Client<T> {
   private final WeakReference<Node> nodeReference;
   private long handle;
   private final String serviceName;
-  private Map<Long, Map.Entry<Consumer, RCLFuture>> pendingRequests;
+  private Map<Long, Map.Entry<Consumer, ResponseFuture>> pendingRequests;
 
   private final ServiceDefinition serviceDefinition;
 
@@ -67,43 +66,52 @@ public ClientImpl(
     this.handle = handle;
     this.serviceName = serviceName;
     this.serviceDefinition = serviceDefinition;
-    this.pendingRequests = new HashMap<Long, Map.Entry<Consumer, RCLFuture>>();
+    this.pendingRequests = new HashMap<Long, Map.Entry<Consumer, ResponseFuture>>();
   }
 
   public ServiceDefinition getServiceDefinition() {
     return this.serviceDefinition;
   }
 
-  public final <U extends MessageDefinition, V extends MessageDefinition> Future<V>
+  public final <U extends MessageDefinition, V extends MessageDefinition> ResponseFuture<V>
   asyncSendRequest(final U request) {
     return asyncSendRequest(request, new Consumer<Future<V>>() {
       public void accept(Future<V> input) {}
     });
   }
 
-  public final <U extends MessageDefinition, V extends MessageDefinition> Future<V>
+  public final <U extends MessageDefinition, V extends MessageDefinition> ResponseFuture<V>
   asyncSendRequest(final U request, final Consumer<Future<V>> callback) {
     synchronized (pendingRequests) {
       long sequenceNumber = nativeSendClientRequest(
           handle, request.getFromJavaConverterInstance(),
           request.getDestructorInstance(), request);
-      RCLFuture<V> future = new RCLFuture<V>();
+      ResponseFuture<V> future = new ResponseFuture<V>(sequenceNumber);
 
-      Map.Entry<Consumer, RCLFuture> entry =
-          new AbstractMap.SimpleEntry<Consumer, RCLFuture>(callback, future);
+      Map.Entry<Consumer, ResponseFuture> entry =
+          new AbstractMap.SimpleEntry<Consumer, ResponseFuture>(callback, future);
       pendingRequests.put(sequenceNumber, entry);
       return future;
     }
   }
 
+  public final <V extends MessageDefinition> boolean
+  removePendingRequest(ResponseFuture<V> future) {
+    synchronized (pendingRequests) {
+      Map.Entry<Consumer, ResponseFuture> entry = pendingRequests.remove(
+        future.getRequestSequenceNumber());
+      return entry != null;
+    }
+  }
+
   public final <U extends MessageDefinition> void handleResponse(
       final RMWRequestId header, final U response) {
     synchronized (pendingRequests) {
       long sequenceNumber = header.sequenceNumber;
-      Map.Entry<Consumer, RCLFuture> entry = pendingRequests.remove(sequenceNumber);
+      Map.Entry<Consumer, ResponseFuture> entry = pendingRequests.remove(sequenceNumber);
       if (entry != null) {
         Consumer<Future> callback = entry.getKey();
-        RCLFuture<U> future = entry.getValue();
+        ResponseFuture<U> future = entry.getValue();
         future.set(response);
         callback.accept(future);
         return;

rcljava/src/main/java/org/ros2/rcljava/client/ResponseFuture.java

@@ -0,0 +1,29 @@
+/* Copyright 2021 Open Source Robotics Foundation, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.ros2.rcljava.client;
+
+import org.ros2.rcljava.concurrent.RCLFuture;
+
+
+public class ResponseFuture<V> extends RCLFuture<V> {
+  public ResponseFuture(long requestSequenceNumber) {
+    this.requestSequenceNumber = requestSequenceNumber;
+  }
+  public long getRequestSequenceNumber() {
+    return this.requestSequenceNumber;
+  }
+  private long requestSequenceNumber;
+}

rcljava/src/test/java/org/ros2/rcljava/client/ClientTest.java

@@ -16,6 +16,7 @@
 package org.ros2.rcljava.client;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
 
@@ -131,4 +132,29 @@ public final void testAdd() throws Exception {
     service.dispose();
     assertEquals(0, service.getHandle());
   }
+
+  @Test
+  public final void testRemovePendingRequest() throws Exception {
+    RCLFuture<rcljava.srv.AddTwoInts_Response> consumerFuture =
+        new RCLFuture<rcljava.srv.AddTwoInts_Response>();
+
+    TestClientConsumer clientConsumer = new TestClientConsumer(consumerFuture);
+
+    Service<rcljava.srv.AddTwoInts> service = node.<rcljava.srv.AddTwoInts>createService(
+        rcljava.srv.AddTwoInts.class, "add_two_ints", clientConsumer);
+
+    rcljava.srv.AddTwoInts_Request request = new rcljava.srv.AddTwoInts_Request();
+    request.setA(2);
+    request.setB(3);
+
+    Client<rcljava.srv.AddTwoInts> client =
+        node.<rcljava.srv.AddTwoInts>createClient(rcljava.srv.AddTwoInts.class, "add_two_ints");
+
+    assertTrue(client.waitForService(Duration.ofSeconds(10)));
+
+    ResponseFuture<rcljava.srv.AddTwoInts_Response> responseFuture = client.asyncSendRequest(request);
+
+    assertTrue(client.removePendingRequest(responseFuture));
+    assertFalse(client.removePendingRequest(responseFuture));
+  }
 }