Merge pull request #239 from ubitricity/master

Add OCPP 2.0.1 and multi-protocol support

Author: TVolden

OCPP-J/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>eu.chargetime.ocpp</groupId>
     <artifactId>OCPP-J</artifactId>
-    <version>1.0.2</version>
+    <version>1.2.0</version>
     <packaging>jar</packaging>
 
     <name>Java-OCA-OCPP OCPP-J</name>
@@ -50,7 +50,7 @@
         <dependency>
             <groupId>eu.chargetime.ocpp</groupId>
             <artifactId>common</artifactId>
-            <version>1.0.2</version>
+            <version>1.2.0</version>
         </dependency>
         <dependency>
             <groupId>com.google.code.gson</groupId>

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

@@ -4,9 +4,8 @@
 import eu.chargetime.ocpp.model.CallErrorMessage;
 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 eu.chargetime.ocpp.model.Message;
 import java.lang.reflect.Type;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
@@ -79,6 +78,16 @@ public JSONCommunicator(Radio radio) {
     super(radio);
   }
 
+  /**
+   * Handle required injections.
+   *
+   * @param radio instance of the {@link Radio}.
+   * @param enableTransactionQueue true if transaction queue should be enabled.
+   */
+  public JSONCommunicator(Radio radio, boolean enableTransactionQueue) {
+    super(radio, enableTransactionQueue);
+  }
+
   private static class ZonedDateTimeSerializer
       implements JsonSerializer<ZonedDateTime>, JsonDeserializer<ZonedDateTime> {
 
@@ -101,17 +110,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;
-      }
-    });
+    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();
   }
@@ -147,26 +157,35 @@ protected Message parse(Object json) {
     Message message;
     JsonParser parser = new JsonParser();
     JsonArray array = parser.parse(json.toString()).getAsJsonArray();
-
-    if (array.get(INDEX_MESSAGEID).getAsInt() == TYPENUMBER_CALL) {
-      message = new CallMessage();
-      message.setAction(array.get(INDEX_CALL_ACTION).getAsString());
-      message.setPayload(array.get(INDEX_CALL_PAYLOAD).toString());
-    } else if (array.get(INDEX_MESSAGEID).getAsInt() == TYPENUMBER_CALLRESULT) {
-      message = new CallResultMessage();
-      message.setPayload(array.get(INDEX_CALLRESULT_PAYLOAD).toString());
-    } else if (array.get(INDEX_MESSAGEID).getAsInt() == TYPENUMBER_CALLERROR) {
-      message = new CallErrorMessage();
-      ((CallErrorMessage) message).setErrorCode(array.get(INDEX_CALLERROR_ERRORCODE).getAsString());
-      ((CallErrorMessage) message)
-          .setErrorDescription(array.get(INDEX_CALLERROR_DESCRIPTION).getAsString());
-      ((CallErrorMessage) message).setRawPayload(array.get(INDEX_CALLERROR_PAYLOAD).toString());
-    } else {
-      logger.error("Unknown message type of message: {}", json.toString());
-      throw new IllegalArgumentException("Unknown message type");
+    String messageId = "-1";
+
+    try {
+      messageId = array.get(INDEX_UNIQUEID).getAsString();
+      if (array.get(INDEX_MESSAGEID).getAsInt() == TYPENUMBER_CALL) {
+        message = new CallMessage();
+        message.setAction(array.get(INDEX_CALL_ACTION).getAsString());
+        message.setPayload(array.get(INDEX_CALL_PAYLOAD).toString());
+      } else if (array.get(INDEX_MESSAGEID).getAsInt() == TYPENUMBER_CALLRESULT) {
+        message = new CallResultMessage();
+        message.setPayload(array.get(INDEX_CALLRESULT_PAYLOAD).toString());
+      } else if (array.get(INDEX_MESSAGEID).getAsInt() == TYPENUMBER_CALLERROR) {
+        message = new CallErrorMessage();
+        ((CallErrorMessage) message).setErrorCode(array.get(INDEX_CALLERROR_ERRORCODE).getAsString());
+        ((CallErrorMessage) message)
+            .setErrorDescription(array.get(INDEX_CALLERROR_DESCRIPTION).getAsString());
+        ((CallErrorMessage) message).setRawPayload(array.get(INDEX_CALLERROR_PAYLOAD).toString());
+      } else {
+        logger.error("Unknown message type of message: {}", json.toString());
+        sendCallError(messageId, null, "MessageTypeNotSupported", null);
+        return null;
+      }
+    } catch (Exception e) {
+      logger.error("Exception while parsing message: {}", json.toString());
+      sendCallError(messageId, null, "RpcFrameworkError", e.getMessage());
+      return null;
     }
 
-    message.setId(array.get(INDEX_UNIQUEID).getAsString());
+    message.setId(messageId);
 
     return message;
   }

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

@@ -36,6 +36,7 @@ public class JSONConfiguration {
   public static final String PING_INTERVAL_PARAMETER = "PING_INTERVAL";
   public static final String USERNAME_PARAMETER = "USERNAME";
   public static final String PASSWORD_PARAMETER = "PASSWORD";
+  public static final String CONNECT_NON_BLOCKING_PARAMETER = "CONNECT_NON_BLOCKING";
   public static final String CONNECT_TIMEOUT_IN_MS_PARAMETER = "CONNECT_TIMEOUT_IN_MS";
   public static final String WEBSOCKET_WORKER_COUNT = "WEBSOCKET_WORKER_COUNT";
 

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

@@ -134,13 +134,24 @@ public void onError(Exception ex) {
 
     configure();
 
-    logger.debug("Trying to connect to: {}", resource);
+    boolean isNonBlocking = isNonBlockingParameterSet();
 
-    try {
-      client.connectBlocking();
-      closed = false;
-    } catch (Exception ex) {
-      logger.warn("client.connectBlocking() failed", ex);
+    logger.debug("Trying to connect to: {}{}", resource, isNonBlocking ? "" : " [blocking]");
+
+    if (isNonBlocking) {
+      try {
+        client.connect();
+        closed = false;
+      } catch (Exception ex) {
+        logger.warn("client.connect() failed", ex);
+      }
+    } else {
+      try {
+        client.connectBlocking();
+        closed = false;
+      } catch (Exception ex) {
+        logger.warn("client.connectBlocking() failed", ex);
+      }
     }
   }
 
@@ -175,16 +186,37 @@ public void disconnect() {
     if (client == null) {
       return;
     }
-    try {
-      client.closeBlocking();
-    } catch (Exception ex) {
-      logger.info("client.closeBlocking() failed", ex);
-    } finally {
-      client = null;
-      closed = true;
+
+    boolean isNonBlocking = isNonBlockingParameterSet();
+
+    logger.debug("Disconnecting{}", isNonBlocking ? "" : " [blocking]");
+
+    if (isNonBlocking) {
+      try {
+        client.close();
+      } catch (Exception ex) {
+        logger.info("client.close() failed", ex);
+      } finally {
+        client = null;
+        closed = true;
+      }
+    } else {
+      try {
+        client.closeBlocking();
+      } catch (Exception ex) {
+        logger.info("client.closeBlocking() failed", ex);
+      } finally {
+        client = null;
+        closed = true;
+      }
     }
   }
 
+  private boolean isNonBlockingParameterSet() {
+    Object rawParam = configuration.getParameter(JSONConfiguration.CONNECT_NON_BLOCKING_PARAMETER);
+    return rawParam instanceof Boolean ? (Boolean) rawParam : false;
+  }
+
   @Override
   public void send(Object request) throws NotConnectedException {
     if (client == null) {

build.gradle

@@ -15,7 +15,7 @@ buildscript {
 
 allprojects {
     group = 'eu.chargetime.ocpp'
-    version = '1.1'
+    version = '1.2'
 }
 
 subprojects {

ocpp-common/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>eu.chargetime.ocpp</groupId>
     <artifactId>common</artifactId>
-    <version>1.0.2</version>
+    <version>1.2.0</version>
 
     <name>Java-OCA-OCPP common</name>
     <description>Implementation of Open Charge-Point Protocol common library.</description>

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

@@ -44,22 +44,20 @@ of this software and associated documentation files (the "Software"), to deal
 public class Client {
   private static final Logger logger = LoggerFactory.getLogger(Client.class);
 
-  private ISession session;
-  private final IFeatureRepository featureRepository;
+  private final ISession session;
   private final IPromiseRepository promiseRepository;
 
   /**
    * Handle required injections.
    *
    * @param session Inject session object
+   * @param promiseRepository Inject promise repository
    * @see Session
    */
   public Client(
       ISession session,
-      IFeatureRepository featureRepository,
       IPromiseRepository promiseRepository) {
     this.session = session;
-    this.featureRepository = featureRepository;
     this.promiseRepository = promiseRepository;
   }
 
@@ -88,7 +86,7 @@ public void handleConfirmation(String uniqueId, Confirmation confirmation) {
 
           @Override
           public Confirmation handleRequest(Request request) throws UnsupportedFeatureException {
-            Optional<Feature> featureOptional = featureRepository.findFeature(request);
+            Optional<Feature> featureOptional = session.getFeatureRepository().findFeature(request);
             if (featureOptional.isPresent()) {
               return featureOptional.get().handleRequest(getSessionId(), request);
             } else {
@@ -149,7 +147,7 @@ public void disconnect() {
    */
   public CompletableFuture<Confirmation> send(Request request)
       throws UnsupportedFeatureException, OccurenceConstraintException {
-    Optional<Feature> featureOptional = featureRepository.findFeature(request);
+    Optional<Feature> featureOptional = session.getFeatureRepository().findFeature(request);
     if (!featureOptional.isPresent()) {
       logger.error("Can't send request: unsupported feature. Payload: {}", request);
       throw new UnsupportedFeatureException();

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

@@ -115,9 +115,20 @@ protected abstract Object makeCallError(
    * @param transmitter Injected {@link Transmitter}
    */
   public Communicator(Radio transmitter) {
+    this(transmitter, true);
+  }
+
+  /**
+   * Handle required injections.
+   *
+   * @param transmitter Injected {@link Transmitter}
+   * @param enableTransactionQueue flag to enable/disable the transaction queue and associated
+   *     processing
+   */
+  public Communicator(Radio transmitter, boolean enableTransactionQueue) {
     this.radio = transmitter;
-    this.transactionQueue = new ArrayDeque<>();
-    this.retryRunner = new RetryRunner();
+    this.transactionQueue = enableTransactionQueue ? new ArrayDeque<>() : null;
+    this.retryRunner = enableTransactionQueue ? new RetryRunner() : null;
     this.failedFlag = false;
   }
 
@@ -164,7 +175,7 @@ public synchronized void sendCall(String uniqueId, String action, Request reques
 
     try {
       if (radio.isClosed()) {
-        if (request.transactionRelated()) {
+        if (request.transactionRelated() && transactionQueue != null) {
           logger.warn("Not connected: storing request to queue: {}", request);
           transactionQueue.add(call);
         } else {
@@ -175,15 +186,17 @@ public synchronized void sendCall(String uniqueId, String action, Request reques
               "The request can't be sent due to the lack of connection",
               request);
         }
-      } else if (request.transactionRelated() && transactionQueue.size() > 0) {
+      } else if (request.transactionRelated()
+          && transactionQueue != null
+          && transactionQueue.size() > 0) {
         transactionQueue.add(call);
         processTransactionQueue();
       } else {
         radio.send(call);
       }
     } catch (NotConnectedException ex) {
       logger.warn("sendCall() failed: not connected");
-      if (request.transactionRelated()) {
+      if (request.transactionRelated() && transactionQueue != null) {
         transactionQueue.add(call);
       } else {
         events.onError(
@@ -211,7 +224,11 @@ public void sendCallResult(String uniqueId, String action, Confirmation confirma
         try {
           completedHandler.onConfirmationCompleted();
         } catch (Throwable e) {
-          events.onError(uniqueId, "ConfirmationCompletedHandlerFailed", "The confirmation completed callback handler failed with exception " + e.toString(), confirmation);
+          events.onError(
+              uniqueId,
+              "ConfirmationCompletedHandlerFailed",
+              "The confirmation completed callback handler failed with exception " + e.toString(),
+              confirmation);
         }
       }
     } catch (NotConnectedException ex) {
@@ -257,7 +274,7 @@ public void disconnect() {
   }
 
   private synchronized void processTransactionQueue() {
-    if (!retryRunner.isAlive()) {
+    if (retryRunner != null && !retryRunner.isAlive()) {
       if (retryRunner.getState() != Thread.State.NEW) {
         retryRunner = new RetryRunner();
       }
@@ -315,7 +332,7 @@ public void disconnected() {
    */
   private Object getRetryMessage() {
     Object result = null;
-    if (!transactionQueue.isEmpty()) result = transactionQueue.peek();
+    if (transactionQueue != null && !transactionQueue.isEmpty()) result = transactionQueue.peek();
     return result;
   }
 
@@ -329,7 +346,7 @@ private boolean hasFailed() {
   }
 
   private void popRetryMessage() {
-    if (!transactionQueue.isEmpty()) transactionQueue.pop();
+    if (transactionQueue != null && !transactionQueue.isEmpty()) transactionQueue.pop();
   }
 
   /** Will resend transaction related requests. */

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

@@ -36,9 +36,23 @@ of this software and associated documentation files (the "Software"), to deal
 
 public class FeatureRepository implements IFeatureRepository {
 
+  private final ProtocolVersion protocolVersion;
   private final Map<String, Feature> actionMap = new HashMap<>();
   private final Map<Class<?>, Feature> classMap = new HashMap<>();
 
+  public FeatureRepository() {
+    this(ProtocolVersion.OCPP1_6);
+  }
+
+  public FeatureRepository(ProtocolVersion protocolVersion) {
+    this.protocolVersion = protocolVersion;
+  }
+
+  @Override
+  public ProtocolVersion getProtocolVersion() {
+    return protocolVersion;
+  }
+
   /**
    * Add {@link Profile} to support a group of features.
    *
@@ -89,6 +103,7 @@ public Optional<Feature> findFeature(Object needle) {
   @Override
   public String toString() {
     return MoreObjects.toStringHelper("FeatureRepository")
+        .add("protocolVersion", protocolVersion)
         .add("actionMap", actionMap)
         .add("classMap", classMap)
         .toString();