feat(plc4j/opcua): Chunking and encryption of request/response calls.

Update of protocol logic to reflect chunked payloads.

Threading updates within driver itself should allow us to have better control over how results are being processed.. and make error handling more consistent.

Signed-off-by: Łukasz Dywicki <[email protected]>

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/config/Limits.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.plc4x.java.opcua.config;
+
+import org.apache.plc4x.java.spi.configuration.Configuration;
+import org.apache.plc4x.java.spi.configuration.annotations.ConfigurationParameter;
+
+public class Limits implements Configuration {
+
+    private static final int DEFAULT_MAX_CHUNK_COUNT = 64;
+    private static final int DEFAULT_MAX_MESSAGE_SIZE = 2097152;
+    private static final int DEFAULT_RECEIVE_BUFFER_SIZE = 65535;
+    private static final int DEFAULT_SEND_BUFFER_SIZE = 65535;
+
+    public static final Limits DEFAULT = new Limits(
+        DEFAULT_RECEIVE_BUFFER_SIZE,
+        DEFAULT_SEND_BUFFER_SIZE,
+        DEFAULT_MAX_MESSAGE_SIZE,
+        DEFAULT_MAX_CHUNK_COUNT
+    );
+
+    private final int receiveBufferSize;
+    private final int sendBufferSize;
+    private final int maxMessageSize;
+    private final int maxChunkCount;
+
+    public Limits() {
+        this(DEFAULT_RECEIVE_BUFFER_SIZE, DEFAULT_SEND_BUFFER_SIZE, DEFAULT_MAX_MESSAGE_SIZE, DEFAULT_MAX_CHUNK_COUNT);
+    }
+
+    public Limits(int receiveBufferSize, int sendBufferSize, int maxMessageSize, int maxChunkCount) {
+        this.receiveBufferSize = receiveBufferSize;
+        this.sendBufferSize = sendBufferSize;
+        this.maxMessageSize = maxMessageSize;
+        this.maxChunkCount = maxChunkCount;
+    }
+
+    public int getReceiveBufferSize() {
+        return receiveBufferSize;
+    }
+
+    public int getSendBufferSize() {
+        return sendBufferSize;
+    }
+
+    public int getMaxMessageSize() {
+        return maxMessageSize;
+    }
+
+    public int getMaxChunkCount() {
+        return maxChunkCount;
+    }
+
+    @Override
+    public String toString() {
+        return "Limits{" +
+            " receiveBufferSize=" + receiveBufferSize +
+            ", sendBufferSize=" + sendBufferSize +
+            ", maxMessageSize=" + maxMessageSize +
+            ", maxChunkCount=" + maxChunkCount +
+            '}';
+    }
+}

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/config/OpcuaConfiguration.java

@@ -18,15 +18,25 @@
  */
 package org.apache.plc4x.java.opcua.config;
 
+import java.security.cert.X509Certificate;
+import java.time.Duration;
+import org.apache.plc4x.java.opcua.context.SecureChannel;
 import org.apache.plc4x.java.opcua.readwrite.PascalByteString;
+import org.apache.plc4x.java.opcua.security.MessageSecurity;
 import org.apache.plc4x.java.opcua.security.SecurityPolicy;
 import org.apache.plc4x.java.spi.configuration.Configuration;
+import org.apache.plc4x.java.spi.configuration.annotations.ComplexConfigurationParameter;
 import org.apache.plc4x.java.spi.configuration.annotations.ConfigurationParameter;
 import org.apache.plc4x.java.spi.configuration.annotations.defaults.BooleanDefaultValue;
-import org.apache.plc4x.java.spi.configuration.annotations.defaults.StringDefaultValue;
 
 public class OpcuaConfiguration implements Configuration {
 
+    public static final long DEFAULT_CONNECTION_LIFETIME = 36000000;
+
+    public static final long DEFAULT_REQUEST_TIMEOUT = 5 * 60 * 1000L;
+
+    public static final long DEFAULT_SESSION_TIMEOUT = 120000L;
+
     @ConfigurationParameter("protocolCode")
     private String protocolCode;
 
@@ -49,6 +59,9 @@ public class OpcuaConfiguration implements Configuration {
     @ConfigurationParameter("securityPolicy")
     private SecurityPolicy securityPolicy = SecurityPolicy.NONE;
 
+    @ConfigurationParameter("messageSecurity")
+    private MessageSecurity messageSecurity = MessageSecurity.SIGN_ENCRYPT;
+
     @ConfigurationParameter("keyStoreFile")
     private String keyStoreFile;
 
@@ -57,8 +70,22 @@ public class OpcuaConfiguration implements Configuration {
 
     @ConfigurationParameter("keyStorePassword")
     private String keyStorePassword;
-    private byte[] senderCertificate;
-    private PascalByteString thumbprint;
+    private X509Certificate serverCertificate;
+
+    @ConfigurationParameter("channelLifetime")
+    private long channelLifetime = DEFAULT_CONNECTION_LIFETIME;
+
+    @ConfigurationParameter("requestTimeout")
+    private long requestTimeout = DEFAULT_REQUEST_TIMEOUT;
+
+    @ConfigurationParameter("sessionTimeout")
+    private long sessionTimeout = DEFAULT_SESSION_TIMEOUT;
+
+    @ConfigurationParameter("openChannelTimeout")
+    private long openChannelTimeout = DEFAULT_REQUEST_TIMEOUT;
+
+    @ComplexConfigurationParameter(prefix = "encoding", defaultOverrides = {}, requiredOverrides = {})
+    private Limits limits = new Limits();
 
     public String getProtocolCode() {
         return protocolCode;
@@ -92,6 +119,10 @@ public SecurityPolicy getSecurityPolicy() {
         return securityPolicy;
     }
 
+    public MessageSecurity getMessageSecurity() {
+        return messageSecurity;
+    }
+
     public String getKeyStoreFile() {
         return keyStoreFile;
     }
@@ -100,6 +131,30 @@ public String getKeyStorePassword() {
         return keyStorePassword;
     }
 
+    public Limits getEncodingLimits() {
+        return limits;
+    }
+
+    public X509Certificate getServerCertificate() {
+        return serverCertificate;
+    }
+
+    public void setServerCertificate(X509Certificate serverCertificate) {
+        this.serverCertificate = serverCertificate;
+    }
+
+    public long getChannelLifetime() {
+        return channelLifetime;
+    }
+
+    public long getRequestTimeout() {
+        return requestTimeout;
+    }
+
+    public long getOpenChannelTimeout() {
+        return openChannelTimeout;
+    }
+
     @Override
     public String toString() {
         return "OpcuaConfiguration{" +
@@ -110,24 +165,8 @@ public String toString() {
             ", keyStoreFile='" + keyStoreFile + '\'' +
             ", certDirectory='" + certDirectory + '\'' +
             ", keyStorePassword='" + (keyStorePassword != null ? "******" : null) + '\'' +
+            ", limits=" + limits +
             '}';
     }
-
-    public byte[] getSenderCertificate() {
-        return senderCertificate;
-    }
-
-    public void setSenderCertificate(byte[] senderCertificate) {
-        this.senderCertificate = senderCertificate;
-    }
-
-    public PascalByteString getThumbprint() {
-        return this.thumbprint;
-    }
-
-    public void setThumbprint(PascalByteString thumbprint) {
-        this.thumbprint = thumbprint;
-    }
-
 }