Cleanup state after future has completed

- Clean up request and promise maps whenever future completes
- Enables timeout for Client/Server.send()

(cherry picked from commit c801fbb221b4090405e58ed37d414fab06b16ce3)

Author: martinfkaeser

ocpp-common/src/main/java/eu/chargetime/ocpp/Client.java

@@ -78,7 +78,6 @@ public void handleConfirmation(String uniqueId, Confirmation confirmation) {
                 promiseRepository.getPromise(uniqueId);
             if (promiseOptional.isPresent()) {
               promiseOptional.get().complete(confirmation);
-              promiseRepository.removePromise(uniqueId);
             } else {
               logger.debug("Promise not found for confirmation {}", confirmation);
             }
@@ -105,11 +104,9 @@ public void handleError(
             Optional<CompletableFuture<Confirmation>> promiseOptional =
                 promiseRepository.getPromise(uniqueId);
             if (promiseOptional.isPresent()) {
-              promiseOptional
-                  .get()
+              promiseOptional.get()
                   .completeExceptionally(
                       new CallErrorException(errorCode, errorDescription, payload));
-              promiseRepository.removePromise(uniqueId);
             } else {
               logger.debug("Promise not found for error {}", errorDescription);
             }
@@ -158,10 +155,16 @@ public CompletableFuture<Confirmation> send(Request request)
       throw new OccurenceConstraintException();
     }
 
-    String id = session.storeRequest(request);
-    CompletableFuture<Confirmation> promise = promiseRepository.createPromise(id);
+    String requestUuid = session.storeRequest(request);
+    CompletableFuture<Confirmation> promise = promiseRepository.createPromise(requestUuid);
 
-    session.sendRequest(featureOptional.get().getAction(), request, id);
+    // Clean up after the promise has completed, no matter if it was successful or had an error or a timeout.
+    promise.whenComplete((confirmation, throwable) -> {
+      session.removeRequest(requestUuid);
+      promiseRepository.removePromise(requestUuid);
+    });
+
+    session.sendRequest(featureOptional.get().getAction(), request, requestUuid);
     return promise;
   }
 

ocpp-common/src/main/java/eu/chargetime/ocpp/ISession.java

@@ -40,6 +40,8 @@ public interface ISession {
 
   String storeRequest(Request payload);
 
+  void removeRequest(String ticket);
+
   void sendRequest(String action, Request payload, String uuid);
 
   boolean completePendingPromise(String id, Confirmation confirmation) throws UnsupportedFeatureException, OccurenceConstraintException;

ocpp-common/src/main/java/eu/chargetime/ocpp/Queue.java

@@ -89,6 +89,16 @@ public Optional<Request> restoreRequest(String ticket) {
     return Optional.empty();
   }
 
+  /**
+   * Remove a stored {@link Request} using a unique identifier.
+   * If no request is found for the identifier this method has no effect.
+   *
+   * @param ticket unique identifier returned when {@link Request} was initially stored.
+   */
+  public void removeRequest(String ticket) {
+    requestQueue.remove(ticket);
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("requestQueue", requestQueue).toString();

ocpp-common/src/main/java/eu/chargetime/ocpp/Server.java

@@ -97,7 +97,6 @@ public void handleConfirmation(String uniqueId, Confirmation confirmation) {
                         promiseRepository.getPromise(uniqueId);
                     if (promiseOptional.isPresent()) {
                       promiseOptional.get().complete(confirmation);
-                      promiseRepository.removePromise(uniqueId);
                     } else {
                       logger.debug("Promise not found for confirmation {}", confirmation);
                     }
@@ -135,11 +134,9 @@ public void handleError(
                     Optional<CompletableFuture<Confirmation>> promiseOptional =
                         promiseRepository.getPromise(uniqueId);
                     if (promiseOptional.isPresent()) {
-                      promiseOptional
-                          .get()
+                      promiseOptional.get()
                           .completeExceptionally(
                               new CallErrorException(errorCode, errorDescription, payload));
-                      promiseRepository.removePromise(uniqueId);
                     } else {
                       logger.debug("Promise not found for error {}", errorDescription);
                     }
@@ -216,9 +213,16 @@ public CompletableFuture<Confirmation> send(UUID sessionIndex, Request request)
       throw new OccurenceConstraintException();
     }
 
-    String id = session.storeRequest(request);
-    CompletableFuture<Confirmation> promise = promiseRepository.createPromise(id);
-    session.sendRequest(featureOptional.get().getAction(), request, id);
+    String requestUuid = session.storeRequest(request);
+    CompletableFuture<Confirmation> promise = promiseRepository.createPromise(requestUuid);
+
+    // Clean up after the promise has completed, no matter if it was successful or had an error or a timeout.
+    promise.whenComplete((confirmation, throwable) -> {
+      session.removeRequest(requestUuid);
+      promiseRepository.removePromise(requestUuid);
+    });
+
+    session.sendRequest(featureOptional.get().getAction(), request, requestUuid);
     return promise;
   }
 

ocpp-common/src/main/java/eu/chargetime/ocpp/Session.java

@@ -114,6 +114,16 @@ public String storeRequest(Request payload) {
     return queue.store(payload);
   }
 
+  /**
+   * Remove a stored {@link Request} using a unique identifier.
+   * If no request is found for the identifier this method has no effect.
+   *
+   * @param ticket unique identifier returned when {@link Request} was initially stored.
+   */
+  public void removeRequest(String ticket) {
+    queue.removeRequest(ticket);
+  }
+
   /**
    * Send a {@link Confirmation} to a {@link Request}
    *

ocpp-v1_6/src/main/java/eu/chargetime/ocpp/TimeoutSessionDecorator.java

@@ -92,6 +92,11 @@ public void sendRequest(String action, Request payload, String uuid) {
     this.session.sendRequest(action, payload, uuid);
   }
 
+  @Override
+  public void removeRequest(String ticket) {
+    this.session.removeRequest(ticket);
+  }
+
   @Override
   public boolean completePendingPromise(String id, Confirmation confirmation) throws UnsupportedFeatureException, OccurenceConstraintException {
     return this.session.completePendingPromise(id, confirmation);