Implement OCPP 2.1 RPC Layer Extensions

Implement the CALLRESULTERROR RPC as a response to a CALLRESULT, in case
that failed internal validation, or if the completion action last added
to the CompletionStage returned by send() using .whenComplete[Async]()
throws any exception.

Add an optional confirmationError() callback method to the ClientEvents
and ServerEvents interfaces to allow the application to get notified
about an incoming CALLRESULTERROR. The application may use the uniqueId
parameter to match it to the Request#getOcppMessageId() of one of the
requests it last responded to, provided it keeps track of them.

Implement the SEND RPC as requests which do not have a confirmation
type. These are sent through the existing send() method, returning a
CompletableFuture which is already completed when the method returns,
to which a completion action can be added to check whether a local
exception prevented the request from being sent.

Fix the only OCPP message using the SEND RPC, NotifyPeriodicEventStream,
and add its missing Feature and Function handlers. The handler for this
message is a void method, as it has no response.

Remove sending CALLERROR in response to anything other than a CALL,
except for the RpcFrameworkError when the message could not be parsed.

Simplify the pendingPromises synchronization in Session by using the
ConcurrentHashMap class rather than HashMap.

Fix a few typos encountered along the way.

Author: robert-s-ubi

ocpp-common/build.gradle

@@ -3,6 +3,7 @@
  */
 
 dependencies {
+    compile 'com.google.code.findbugs:jsr305:3.0.2'
     compile 'org.slf4j:slf4j-api:2.0.17'
     compile 'ch.qos.logback:logback-classic:1.3.16'
     compile group: 'javax.xml.bind', name: 'jaxb-api', version: '2.3.1'

ocpp-common/pom.xml

@@ -38,6 +38,11 @@
      </scm>
 
     <dependencies>
+        <dependency>
+          <groupId>com.google.code.findbugs</groupId>
+          <artifactId>jsr305</artifactId>
+          <version>3.0.2</version>
+        </dependency>
         <dependency>
           <groupId>javax.xml.bind</groupId>
           <artifactId>jaxb-api</artifactId>

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

@@ -30,6 +30,7 @@ of this software and associated documentation files (the "Software"), to deal
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import javax.annotation.Nullable;
 
 public class AsyncPromiseFulfillerDecorator implements PromiseFulfiller {
 
@@ -43,7 +44,9 @@ public static void setExecutor(ExecutorService newExecutor) {
 
   @Override
   public void fulfill(
-      CompletableFuture<Confirmation> promise, SessionEvents eventHandler, Request request) {
+      @Nullable CompletableFuture<Confirmation> promise,
+      SessionEvents eventHandler,
+      Request request) {
     executor.submit(() -> promiseFulfiller.fulfill(promise, eventHandler, request));
   }
 

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

@@ -32,6 +32,8 @@ of this software and associated documentation files (the "Software"), to deal
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import javax.annotation.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -71,11 +73,23 @@ public void connect(String uri, ClientEvents events) {
         new SessionEvents() {
 
           @Override
-          public void handleConfirmation(String uniqueId, Confirmation confirmation) {
+          public void handleConfirmation(String uniqueId, @Nullable Confirmation confirmation) {
             Optional<CompletableFuture<Confirmation>> promiseOptional =
                 promiseRepository.getPromise(uniqueId);
             if (promiseOptional.isPresent()) {
               promiseOptional.get().complete(confirmation);
+              // join completion to catch and rethrow any exceptions thrown in the last added
+              // completion action, so that a CALLRESULTERROR may be produced from it.
+              try {
+                promiseOptional.get().join();
+              } catch (CompletionException e) {
+                Throwable cause = e.getCause() != null ? e.getCause() : e;
+                if (cause instanceof RuntimeException) {
+                  throw (RuntimeException) cause;
+                } else {
+                  throw new RuntimeException(cause);
+                }
+              }
             } else {
               logger.debug("Promise not found for confirmation {}", confirmation);
             }
@@ -112,6 +126,18 @@ public void handleError(
             }
           }
 
+          @Override
+          public void handleConfirmationError(
+              String uniqueId, String errorCode, String errorDescription, Object payload) {
+            logger.error(
+                "Received an error which occurred while processing a call result: "
+                    + "uniqueId {}: errorCode: {}, errorDescription: {}",
+                uniqueId,
+                errorCode,
+                errorDescription);
+            events.confirmationError(uniqueId, errorCode, errorDescription, payload);
+          }
+
           @Override
           public void handleConnectionClosed() {
             if (events != null) events.connectionClosed();
@@ -137,7 +163,8 @@ public void disconnect() {
    * Send a {@link Request} to the server. Can only send {@link Request} that the client supports.
    *
    * @param request outgoing request
-   * @return call back object, will be fulfilled with confirmation when received
+   * @return call back object, will be fulfilled with confirmation when received or {@code null} if
+   *     the request has no confirmation, or exceptionally if a local or remote error occurred.
    * @throws UnsupportedFeatureException trying to send a request from an unsupported feature
    * @throws OccurenceConstraintException Thrown if the request isn't valid.
    * @see CompletableFuture
@@ -166,7 +193,11 @@ public CompletableFuture<Confirmation> send(Request request)
           promiseRepository.removePromise(requestUuid);
         });
 
-    session.sendRequest(featureOptional.get().getAction(), request, requestUuid);
+    if (featureOptional.get().getConfirmationType() != null) {
+      session.sendRequest(featureOptional.get().getAction(), request, requestUuid);
+    } else {
+      session.sendMessage(featureOptional.get().getAction(), request, requestUuid);
+    }
     return promise;
   }
 

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

@@ -31,4 +31,7 @@ public interface ClientEvents {
   void connectionOpened();
 
   void connectionClosed();
+
+  default void confirmationError(
+      String uniqueId, String errorCode, String errorDescription, Object payload) {};
 }

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

@@ -41,9 +41,9 @@ of this software and associated documentation files (the "Software"), to deal
 public abstract class Communicator {
   private static final Logger logger = LoggerFactory.getLogger(Communicator.class);
 
+  private final ArrayDeque<Object> transactionQueue;
   private RetryRunner retryRunner;
   protected Radio radio;
-  private ArrayDeque<Object> transactionQueue;
   private CommunicatorEvents events;
   private boolean failedFlag;
 
@@ -97,6 +97,27 @@ public abstract class Communicator {
   protected abstract Object makeCallError(
       String uniqueId, String action, String errorCode, String errorDescription);
 
+  /**
+   * Create a call result error envelope to transmit.
+   *
+   * @param uniqueId the id the receiver expects.
+   * @param errorCode an OCPP error code.
+   * @param errorDescription an associated error description.
+   * @return a fully packed message ready to send.
+   */
+  protected abstract Object makeCallResultError(
+      String uniqueId, String action, String errorCode, String errorDescription);
+
+  /**
+   * Create a send envelope to transmit to the server.
+   *
+   * @param uniqueId the id of the message.
+   * @param action action name of the feature.
+   * @param payload packed payload.
+   * @return a fully packed message ready to send.
+   */
+  protected abstract Object makeSend(String uniqueId, String action, Object payload);
+
   /**
    * Identify an incoming call and parse it into one of the following: {@link CallMessage} a
    * request. {@link CallResultMessage} a response.
@@ -177,7 +198,7 @@ public synchronized void sendCall(String uniqueId, String action, Request reques
         }
       } else if (request.transactionRelated()
           && transactionQueue != null
-          && transactionQueue.size() > 0) {
+          && !transactionQueue.isEmpty()) {
         transactionQueue.add(call);
         processTransactionQueue();
       } else {
@@ -216,7 +237,7 @@ public void sendCallResult(String uniqueId, String action, Confirmation confirma
           events.onError(
               uniqueId,
               "ConfirmationCompletedHandlerFailed",
-              "The confirmation completed callback handler failed with exception " + e.toString(),
+              "The confirmation completed callback handler failed with exception " + e,
               confirmation);
         }
       }
@@ -240,7 +261,7 @@ public void sendCallResult(String uniqueId, String action, Confirmation confirma
   public void sendCallError(
       String uniqueId, String action, String errorCode, String errorDescription) {
     logger.error(
-        "An error occurred. Sending this information: uniqueId {}: action: {}, errorCore: {}, errorDescription: {}",
+        "An error occurred. Sending this information: uniqueId {}: action: {}, errorCode: {}, errorDescription: {}",
         uniqueId,
         action,
         errorCode,
@@ -257,6 +278,65 @@ public void sendCallError(
     }
   }
 
+  /**
+   * Send a call result error. If offline, the message is thrown away.
+   *
+   * @param uniqueId the id the receiver expects a response to.
+   * @param errorCode an OCPP error Code
+   * @param errorDescription a associated error description.
+   */
+  public void sendCallResultError(
+      String uniqueId, String action, String errorCode, String errorDescription) {
+    logger.error(
+        "An error occurred while processing a call result. Sending this information: "
+            + "uniqueId {}: action: {}, errorCode: {}, errorDescription: {}",
+        uniqueId,
+        action,
+        errorCode,
+        errorDescription);
+    try {
+      radio.send(makeCallResultError(uniqueId, action, errorCode, errorDescription));
+    } catch (NotConnectedException ex) {
+      logger.warn("sendCallResultError() failed", ex);
+      events.onError(
+          uniqueId,
+          "Not connected",
+          "The call result error couldn't be sent due to the lack of connection",
+          errorCode);
+    }
+  }
+
+  /**
+   * Send a {@link Request} which has no confirmation.
+   *
+   * @param uniqueId the id of the {@link Request}.
+   * @param action action name of the {@link eu.chargetime.ocpp.feature.Feature}.
+   * @param request the outgoing {@link Request}
+   */
+  public synchronized void send(String uniqueId, String action, Request request) {
+    Object call = makeSend(uniqueId, action, packPayload(request));
+
+    try {
+      if (radio.isClosed()) {
+        logger.warn("Not connected: can't send request: {}", request);
+        events.onError(
+            uniqueId,
+            "Not connected",
+            "The request can't be sent due to the lack of connection",
+            request);
+      } else {
+        radio.send(call);
+      }
+    } catch (NotConnectedException ex) {
+      logger.warn("sendCall() failed: not connected");
+      events.onError(
+          uniqueId,
+          "Not connected",
+          "The request can't be sent due to the lack of connection",
+          request);
+    }
+  }
+
   /** Close down the connection. Uses the {@link Transmitter}. */
   public void disconnect() {
     radio.disconnect();
@@ -289,6 +369,10 @@ public void receivedMessage(Object input) {
       Message message = parse(input);
       if (message instanceof CallResultMessage) {
         events.onCallResult(message.getId(), message.getAction(), message.getPayload());
+      } else if (message instanceof CallResultErrorMessage) {
+        CallResultErrorMessage call = (CallResultErrorMessage) message;
+        events.onCallResultError(
+            call.getId(), call.getErrorCode(), call.getErrorDescription(), call.getRawPayload());
       } else if (message instanceof CallErrorMessage) {
         failedFlag = true;
         CallErrorMessage call = (CallErrorMessage) message;
@@ -297,6 +381,9 @@ public void receivedMessage(Object input) {
       } else if (message instanceof CallMessage) {
         CallMessage call = (CallMessage) message;
         events.onCall(call.getId(), call.getAction(), call.getPayload());
+      } else if (message instanceof SendMessage) {
+        SendMessage send = (SendMessage) message;
+        events.onSend(send.getId(), send.getAction(), send.getPayload());
       }
     }
 
@@ -318,7 +405,7 @@ private Object getRetryMessage() {
   }
 
   /**
-   * Check if a error message was received.
+   * Check if an error message was received.
    *
    * @return whether a fail flag has been raised.
    */

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

@@ -66,6 +66,31 @@ public interface CommunicatorEvents {
    */
   void onError(String id, String errorCode, String errorDescription, Object payload);
 
+  /**
+   * Handle call result error.
+   *
+   * <p>Hint: Use the id to identify the original call result. You can use {@link Communicator}s
+   * unpackPayload method.
+   *
+   * @param id unique id used to identify the original call result.
+   * @param errorCode short text to categorize the error.
+   * @param errorDescription a longer text to describe the error.
+   * @param payload Object payload attached to the error.
+   */
+  void onCallResultError(String id, String errorCode, String errorDescription, Object payload);
+
+  /**
+   * Handle send.
+   *
+   * <p>Hint: Use the action name to identify the feature, you can then choose to use {@link
+   * Communicator}s unpackPayload method.
+   *
+   * @param id unique id.
+   * @param action action name used to identify the feature.
+   * @param payload raw payload.
+   */
+  void onSend(String id, String action, Object payload);
+
   /** The connection was disconnected. */
   void onDisconnected();
 

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

@@ -73,7 +73,9 @@ public void addFeatureProfile(Profile profile) {
   public void addFeature(Feature feature) {
     actionMap.put(feature.getAction(), feature);
     classMap.put(feature.getRequestType(), feature);
-    classMap.put(feature.getConfirmationType(), feature);
+    if (feature.getConfirmationType() != null) {
+      classMap.put(feature.getConfirmationType(), feature);
+    }
   }
 
   /**

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

@@ -47,5 +47,7 @@ public interface ISession {
   boolean completePendingPromise(String id, Confirmation confirmation)
       throws UnsupportedFeatureException, OccurenceConstraintException;
 
+  void sendMessage(String action, Request payload, String uuid);
+
   void close();
 }