Merge pull request #215 from dimaa6/172-async-execution

Asynchronous execution

Author: TVolden

OCPP-J/src/main/java/eu/chargetime/ocpp/JSONCommunicator.java

@@ -5,6 +5,8 @@
 import eu.chargetime.ocpp.model.CallMessage;
 import eu.chargetime.ocpp.model.CallResultMessage;
 import eu.chargetime.ocpp.model.Message;
+import eu.chargetime.ocpp.model.Exclude;
+
 import java.lang.reflect.Type;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
@@ -99,6 +101,18 @@ public ZonedDateTime deserialize(
   static {
     GsonBuilder builder = new GsonBuilder();
     builder.registerTypeAdapter(ZonedDateTime.class, new ZonedDateTimeSerializer());
+    builder.addSerializationExclusionStrategy(new ExclusionStrategy() {
+      @Override
+      public boolean shouldSkipClass(Class<?> clazz) {
+        return false;
+      }
+
+      @Override
+      public boolean shouldSkipField(FieldAttributes field) {
+        return field.getAnnotation(Exclude.class) != null;
+      }
+    });
+
     gson = builder.disableHtmlEscaping().create();
   }
 

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

@@ -96,6 +96,11 @@ public Confirmation handleRequest(Request request) throws UnsupportedFeatureExce
             }
           }
 
+          @Override
+          public boolean asyncCompleteRequest(String uniqueId, Confirmation confirmation) throws UnsupportedFeatureException, OccurenceConstraintException {
+            return session.completePendingPromise(uniqueId, confirmation);
+          }
+
           @Override
           public void handleError(
               String uniqueId, String errorCode, String errorDescription, Object payload) {
@@ -165,4 +170,8 @@ public CompletableFuture<Confirmation> send(Request request)
   public UUID getSessionId() {
     return this.session.getSessionId();
   }
+
+  public boolean asyncCompleteRequest(String uniqueId, Confirmation confirmation) throws UnsupportedFeatureException, OccurenceConstraintException {
+    return session.completePendingPromise(uniqueId, confirmation);
+  }
 }

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

@@ -1,5 +1,6 @@
 package eu.chargetime.ocpp;
 
+import eu.chargetime.ocpp.model.Confirmation;
 import eu.chargetime.ocpp.model.Request;
 import java.util.UUID;
 
@@ -39,5 +40,7 @@ public interface ISession {
 
   void sendRequest(String action, Request payload, String uuid);
 
+  boolean completePendingPromise(String id, Confirmation confirmation) throws UnsupportedFeatureException, OccurenceConstraintException;
+
   void close();
 }

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

@@ -38,10 +38,8 @@ public RequestDispatcher(PromiseFulfiller fulfiller) {
     this.fulfiller = fulfiller;
   }
 
-  public CompletableFuture<Confirmation> handleRequest(Request request) {
-    CompletableFuture<Confirmation> promise = new CompletableFuture<>();
+  public void handleRequest(CompletableFuture<Confirmation> promise, Request request) {
     fulfiller.fulfill(promise, eventHandler, request);
-    return promise;
   }
 
   public void setEventHandler(SessionEvents eventHandler) {

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

@@ -126,6 +126,11 @@ public Confirmation handleRequest(Request request)
                     }
                   }
 
+                  @Override
+                  public boolean asyncCompleteRequest(String uniqueId, Confirmation confirmation) throws UnsupportedFeatureException, OccurenceConstraintException {
+                    return session.completePendingPromise(uniqueId, confirmation);
+                  }
+
                   @Override
                   public void handleError(
                       String uniqueId, String errorCode, String errorDescription, Object payload) {
@@ -219,6 +224,29 @@ public CompletableFuture<Confirmation> send(UUID sessionIndex, Request request)
     return promise;
   }
 
+  /**
+   * Indicate completion of a pending request.
+   *
+   * @param sessionIndex Session index of the client.
+   * @param uniqueId the unique id used for the original {@link Request}.
+   * @param confirmation the {@link Confirmation} to the original {@link Request}.
+   * @return a boolean indicating if pending request was found.
+   * @throws NotConnectedException Thrown if session with passed sessionIndex is not found
+   */
+  public boolean asyncCompleteRequest(UUID sessionIndex, String uniqueId, Confirmation confirmation) throws NotConnectedException, UnsupportedFeatureException, OccurenceConstraintException {
+    ISession session = sessions.get(sessionIndex);
+
+    if (session == null) {
+      logger.warn("Session not found by index: {}", sessionIndex);
+
+      // No session found means client disconnected and request should be cancelled
+      throw new NotConnectedException();
+    }
+
+    return session.completePendingPromise(uniqueId, confirmation);
+  }
+
+
   public boolean isSessionOpen(UUID sessionIndex) {
     return sessions.containsKey(sessionIndex);
   }

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

@@ -32,6 +32,9 @@ of this software and associated documentation files (the "Software"), to deal
 import eu.chargetime.ocpp.utilities.MoreObjects;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.AbstractMap;
 import java.util.concurrent.CompletableFuture;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -50,6 +53,7 @@ public class Session implements ISession {
   private final RequestDispatcher dispatcher;
   private final IFeatureRepository featureRepository;
   private SessionEvents events;
+  private final Map<String, AbstractMap.SimpleImmutableEntry<String, CompletableFuture<Confirmation>>> pendingPromises = new HashMap<>();
 
   /**
    * Handles required injections.
@@ -198,9 +202,12 @@ public synchronized void onCall(String id, String action, Object payload) {
         try {
           Request request =
               communicator.unpackPayload(payload, featureOptional.get().getRequestType());
+          request.setOcppMessageId(id);
           if (request.validate()) {
-            CompletableFuture<Confirmation> promise = dispatcher.handleRequest(request);
+            CompletableFuture<Confirmation> promise = new CompletableFuture<>();
             promise.whenComplete(new ConfirmationHandler(id, action, communicator));
+            addPendingPromise(id, action, promise);
+            dispatcher.handleRequest(promise, request);
           } else {
             communicator.sendCallError(
                 id, action, "OccurenceConstraintViolation", OCCURENCE_CONSTRAINT_VIOLATION);
@@ -231,6 +238,36 @@ public void onConnected() {
     }
   }
 
+  private void addPendingPromise(String id, String action, CompletableFuture<Confirmation> promise) {
+    synchronized (pendingPromises) {
+      pendingPromises.put(id, new AbstractMap.SimpleImmutableEntry<>(action, promise));
+    }
+  }
+
+  @Override
+  public boolean completePendingPromise(String id, Confirmation confirmation) throws UnsupportedFeatureException, OccurenceConstraintException {
+    AbstractMap.SimpleImmutableEntry<String, CompletableFuture<Confirmation>> promiseAction = null;
+    // synchronization prevents from confirming one promise more than once, as we remove found promise
+    synchronized (pendingPromises) {
+      promiseAction = pendingPromises.get(id);
+      if (promiseAction == null) return false;
+      // remove promise from store
+      pendingPromises.remove(id);
+    }
+    // check confirmation type, it has to correspond to original request type
+    Optional<Feature> featureOptional = featureRepository.findFeature(promiseAction.getKey());
+    if (featureOptional.isPresent()) {
+      if (!featureOptional.get().getConfirmationType().isInstance(confirmation)) {
+        throw new OccurenceConstraintException();
+      }
+    } else {
+      logger.debug("Feature for confirmation with id: {} not found in session: {}", id, this);
+      throw new UnsupportedFeatureException("Error with getting confirmation type by request id = " + id);
+    }
+    promiseAction.getValue().complete(confirmation);
+    return true;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;

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

@@ -49,6 +49,15 @@ public interface SessionEvents {
    */
   Confirmation handleRequest(Request request) throws UnsupportedFeatureException;
 
+  /**
+   * Completes a pending request {@link Request}.
+   *
+   * @param uniqueId the unique id used for the {@link Request}.
+   * @param confirmation the {@link Confirmation} to the {@link Request}.
+   * @return a boolean indicating if pending request was found.
+   */
+  boolean asyncCompleteRequest(String uniqueId, Confirmation confirmation) throws UnsupportedFeatureException, OccurenceConstraintException;
+
   /**
    * Handle a error to a {@link Request}.
    *

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

@@ -39,7 +39,10 @@ public void fulfill(
       CompletableFuture<Confirmation> promise, SessionEvents eventHandler, Request request) {
     try {
       Confirmation conf = eventHandler.handleRequest(request);
-      promise.complete(conf);
+      // Confirmation may be null, in this case asynchronous execution is assumed
+      if (conf != null) {
+        eventHandler.asyncCompleteRequest(request.getOcppMessageId(), conf);
+      }
     } catch (Exception ex) {
       logger.warn("fulfillPromis() failed", ex);
       promise.completeExceptionally(ex);

ocpp-common/src/main/java/eu/chargetime/ocpp/model/Exclude.java

@@ -0,0 +1,10 @@
+package eu.chargetime.ocpp.model;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.FIELD)
+public @interface Exclude {}

ocpp-common/src/main/java/eu/chargetime/ocpp/model/Request.java

@@ -30,4 +30,6 @@ of this software and associated documentation files (the "Software"), to deal
 /** Interface used to flag a model as Request payload. */
 public interface Request extends Validatable {
   boolean transactionRelated();
+  String getOcppMessageId();
+  void setOcppMessageId(String id);
 }