Fix compile errors after updates of opcua driver.

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

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/Conversation.java

@@ -50,7 +50,7 @@
 import org.apache.plc4x.java.opcua.readwrite.NodeIdFourByte;
 import org.apache.plc4x.java.opcua.readwrite.NodeIdTwoByte;
 import org.apache.plc4x.java.opcua.readwrite.NodeIdTypeDefinition;
-import org.apache.plc4x.java.opcua.readwrite.NullExtension;
+import org.apache.plc4x.java.opcua.readwrite.NullExtensionObject;
 import org.apache.plc4x.java.opcua.readwrite.OpcuaAPU;
 import org.apache.plc4x.java.opcua.readwrite.OpcuaAcknowledgeResponse;
 import org.apache.plc4x.java.opcua.readwrite.OpcuaCloseRequest;
@@ -66,10 +66,12 @@
 import org.apache.plc4x.java.opcua.readwrite.Payload;
 import org.apache.plc4x.java.opcua.readwrite.RequestHeader;
 import org.apache.plc4x.java.opcua.readwrite.ResponseHeader;
+import org.apache.plc4x.java.opcua.readwrite.RootExtensionObject;
 import org.apache.plc4x.java.opcua.readwrite.SecurityHeader;
 import org.apache.plc4x.java.opcua.readwrite.SequenceHeader;
 import org.apache.plc4x.java.opcua.readwrite.ServiceFault;
 import org.apache.plc4x.java.opcua.readwrite.SignatureData;
+import org.apache.plc4x.java.opcua.readwrite.WireExtensionObject;
 import org.apache.plc4x.java.opcua.security.MessageSecurity;
 import org.apache.plc4x.java.opcua.security.SecurityPolicy;
 import org.apache.plc4x.java.spi.ConversationContext;
@@ -89,10 +91,10 @@ public class Conversation {
         null
     );
 
-    protected static final ExtensionObject NULL_EXTENSION_OBJECT = new ExtensionObject(
+    protected static final ExtensionObject NULL_EXTENSION_OBJECT = new WireExtensionObject(
         NULL_EXPANDED_NODE_ID,
         new ExtensionObjectEncodingMask(false, false, false),
-        new NullExtension());               // Body
+        new NullExtensionObject());               // Body
 
 
     private final Logger logger = LoggerFactory.getLogger(Conversation.class);
@@ -274,14 +276,13 @@ private CompletableFuture<Object> submit(ExtensionObjectDefinition requestDefini
         ExpandedNodeId expandedNodeId = new ExpandedNodeId(
             false,           //Namespace Uri Specified
             false,            //Server Index Specified
-            new NodeIdFourByte((short) 0, Integer.parseInt(requestDefinition.getExtensionId())),
+            new NodeIdFourByte((short) 0, requestDefinition.getExtensionId()),
             null,
             null
         );
-        ExtensionObject requestObject = new ExtensionObject(expandedNodeId, null, requestDefinition);
         ExtensiblePayload payload = new ExtensiblePayload(
             new SequenceHeader(tm.getSequenceSupplier().get(), requestId),
-            requestObject
+            new RootExtensionObject(expandedNodeId, requestDefinition)
         );
 
         MemoryChunkStorage chunkStorage = new MemoryChunkStorage();

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannel.java

@@ -79,6 +79,8 @@ public class SecureChannel {
     private static final PascalString PRODUCT_URI = new PascalString("urn:apache:plc4x:client");
     private static final PascalString APPLICATION_TEXT = new PascalString("OPCUA client for the Apache PLC4X:PLC4J project");
     public static final ScheduledExecutorService KEEP_ALIVE_EXECUTOR = newSingleThreadScheduledExecutor(runnable -> new Thread(runnable, "plc4x-opcua-keep-alive"));
+    public static final ExtensionObjectEncodingMask BINARY_ENCODING_MASK = new ExtensionObjectEncodingMask(
+        false, false, true);
     private final String sessionName = "UaSession:" + APPLICATION_TEXT.getStringValue() + ":" + RandomStringUtils.random(20, true, true);
     private final PascalByteString localCertificateString;
     private final PascalByteString remoteCertificateThumbprint;
@@ -182,10 +184,9 @@ public CompletableFuture<OpenSecureChannelResponse> onConnectOpenSecureChannel(S
         }
 
         ExpandedNodeId expandedNodeId = new ExpandedNodeId(false, false,
-            new NodeIdFourByte((short) 0, Integer.parseInt(openSecureChannelRequest.getExtensionId())),
+            new NodeIdFourByte((short) 0, openSecureChannelRequest.getExtensionId()),
             null, null
         );
-        ExtensionObject extObject = new ExtensionObject(expandedNodeId, null, openSecureChannelRequest);
 
         Function<CallContext, OpcuaOpenRequest> openRequest = context -> {
             LOGGER.debug("Submitting OpenSecureChannel with id of {}", context.getRequestId());
@@ -196,7 +197,7 @@ public CompletableFuture<OpenSecureChannelResponse> onConnectOpenSecureChannel(S
             ),
             new ExtensiblePayload(
                 new SequenceHeader(context.getNextSequenceNumber(), context.getRequestId()),
-                extObject
+                new RootExtensionObject(expandedNodeId, openSecureChannelRequest)
             ));
         };
 
@@ -375,15 +376,15 @@ private void onDisconnectCloseSecureChannel() {
         CloseSecureChannelRequest closeSecureChannelRequest = new CloseSecureChannelRequest(requestHeader);
 
         ExpandedNodeId expandedNodeId = new ExpandedNodeId(false, false,
-            new NodeIdFourByte((short) 0, Integer.parseInt(closeSecureChannelRequest.getExtensionId())),
+            new NodeIdFourByte((short) 0, closeSecureChannelRequest.getExtensionId()),
             null, null
         );
 
         Function<CallContext, OpcuaCloseRequest> closeRequest = ctx ->
             new OpcuaCloseRequest(FINAL, ctx.getSecurityHeader(),
             new ExtensiblePayload(
                 new SequenceHeader(ctx.getNextSequenceNumber(), ctx.getRequestId()),
-                new ExtensionObject(expandedNodeId, null, closeSecureChannelRequest)
+                new RootExtensionObject(expandedNodeId, closeSecureChannelRequest)
             )
         );
 
@@ -600,10 +601,9 @@ private ExtensionObject getIdentityToken(UserTokenType tokenType, String securit
                     null
                 );
 
-                return new ExtensionObject(
-                    extExpandedNodeId,
-                    new ExtensionObjectEncodingMask(false, false, true),
-                    new UserIdentityToken(new PascalString(securityPolicy), anonymousIdentityToken));
+                return new WireExtensionObject(extExpandedNodeId, BINARY_ENCODING_MASK, new BinaryWireExtensionObject(
+                    new UserIdentityToken(new PascalString(securityPolicy), anonymousIdentityToken)
+                ));
             case userTokenTypeUserName:
                 //Encrypt the password using the server nonce and server public key
                 byte[] remoteNonce = conversation.getRemoteNonce();
@@ -630,10 +630,9 @@ private ExtensionObject getIdentityToken(UserTokenType tokenType, String securit
                     null,
                     null);
 
-                return new ExtensionObject(
-                    extExpandedNodeId,
-                    new ExtensionObjectEncodingMask(false, false, true),
-                    new UserIdentityToken(new PascalString(securityPolicy), userNameIdentityToken));
+                return new WireExtensionObject(extExpandedNodeId, BINARY_ENCODING_MASK, new BinaryWireExtensionObject(
+                    new UserIdentityToken(new PascalString(securityPolicy), userNameIdentityToken)
+                ));
         }
         return null;
     }

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaProtocolLogic.java

@@ -72,10 +72,10 @@ public class OpcuaProtocolLogic extends Plc4xProtocolBase<OpcuaAPU> implements H
         null
     );
 
-    protected static final ExtensionObject NULL_EXTENSION_OBJECT = new ExtensionObject(
+    protected static final ExtensionObject NULL_EXTENSION_OBJECT = new WireExtensionObject(
         NULL_EXPANDED_NODEID,
         new ExtensionObjectEncodingMask(false, false, false),
-        new NullExtension());               // Body
+        new NullExtensionObject());               // Body
 
     private static final long EPOCH_OFFSET = 116444736000000000L;         //Offset between OPC UA epoch time and linux epoch time.
     private final Map<Long, OpcuaSubscriptionHandle> subscriptions = new ConcurrentHashMap<>();

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java

@@ -93,7 +93,7 @@ public CompletableFuture<OpcuaSubscriptionHandle> onSubscribeCreateMonitoredItem
 
             ReadValueId readValueId = new ReadValueId(
                 idNode,
-                0xD,
+                AttributeId.Value.getValue(),
                 OpcuaProtocolLogic.NULL_STRING,
                 new QualifiedName(0, OpcuaProtocolLogic.NULL_STRING));
 
@@ -114,19 +114,50 @@ public CompletableFuture<OpcuaSubscriptionHandle> onSubscribeCreateMonitoredItem
 
             ExtensionObject eventFilter = OpcuaProtocolLogic.NULL_EXTENSION_OBJECT;
             if (tagDefaultPlcSubscription.getPlcSubscriptionType() == PlcSubscriptionType.EVENT) {
-                FilterOperand filterOperand = new FilterOperand();
-                EventFilter eventFilter1 = new EventFilter(
-                    Collections.emptyList(), //Arrays.asList(filterOperand),
-                    null //new NullExtension()
+                NodeId nodeId = new NodeId(new NodeIdFourByte((short) 0, OpcuaNodeIdServicesObjectType.BaseEventType.getValue()));
+                List<SimpleAttributeOperand> filterOperand = Arrays.asList(
+                    new SimpleAttributeOperand(nodeId,
+                        Arrays.asList(new QualifiedName(0, new PascalString("EventId"))),
+                        AttributeId.Value.getValue(),
+                        null
+                    ),
+                    new SimpleAttributeOperand(nodeId,
+                        Arrays.asList(new QualifiedName(0, new PascalString("EventType"))),
+                        AttributeId.Value.getValue(),
+                        null
+                    ),
+                    new SimpleAttributeOperand(nodeId,
+                        Arrays.asList(new QualifiedName(0, new PascalString("Severity"))),
+                        AttributeId.Value.getValue(),
+                        null
+                    ),
+                    new SimpleAttributeOperand(nodeId,
+                        Arrays.asList(new QualifiedName(0, new PascalString("Time"))),
+                        AttributeId.Value.getValue(),
+                        null
+                    ),
+                    new SimpleAttributeOperand(nodeId,
+                        Arrays.asList(new QualifiedName(0, new PascalString("Message"))),
+                        AttributeId.Value.getValue(),
+                        null
+                    )
                 );
+
+                EventFilter filterPayload = new EventFilter(filterOperand, new ContentFilter(null));
                 ExpandedNodeId expandedNodeId = new ExpandedNodeId(false, false,
-                    new NodeIdFourByte((short) 0, Integer.parseInt(eventFilter.getIdentifier())),
+                    new NodeIdFourByte((short) 0, filterPayload.getExtensionId()),
                     null, null
                 );
-                eventFilter = new ExtensionObject(expandedNodeId,
+                eventFilter = new WireExtensionObject(
+                    expandedNodeId,
                     new ExtensionObjectEncodingMask(false, false, true),
-                    eventFilter1
+                    new BinaryWireExtensionObject(filterPayload)
                 );
+                readValueId = new ReadValueId(
+                    idNode,
+                    AttributeId.EventNotifier.getValue(),
+                    OpcuaProtocolLogic.NULL_STRING,
+                    new QualifiedName(0, OpcuaProtocolLogic.NULL_STRING));
             }
 
             long clientHandle = clientHandles.getAndIncrement();
@@ -138,8 +169,7 @@ public CompletableFuture<OpcuaSubscriptionHandle> onSubscribeCreateMonitoredItem
                 true        // discard oldest
             );
 
-            MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(
-                readValueId, monitoringMode, parameters);
+            MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(readValueId, monitoringMode, parameters);
 
             requestList.add(request);
         }
@@ -160,10 +190,7 @@ public CompletableFuture<OpcuaSubscriptionHandle> onSubscribeCreateMonitoredItem
                     LOGGER.info("Error while sending the Create Monitored Item Subscription Message", error);
                 }
             }).thenApply(responseMessage -> {
-                MonitoredItemCreateResult[] array = responseMessage.getResults().stream()
-                    .filter(extensionObjectDefinition -> extensionObjectDefinition instanceof MonitoredItemCreateResult)
-                    .map(extensionObjectDefinition -> (MonitoredItemCreateResult) extensionObjectDefinition)
-                    .toArray(MonitoredItemCreateResult[]::new);
+                MonitoredItemCreateResult[] array = responseMessage.getResults().stream().toArray(MonitoredItemCreateResult[]::new);
                 for (int index = 0, arrayLength = array.length; index < arrayLength; index++) {
                     MonitoredItemCreateResult result = array[index];
                     if (OpcuaStatusCode.enumForValue(result.getStatusCode().getStatusCode()) != OpcuaStatusCode.Good) {
@@ -219,7 +246,7 @@ private void sendPublishRequest() {
                             onSubscriptionValue(items.stream().toArray(MonitoredItemNotification[]::new));
                         } else if (notification instanceof NotificationData) {
                             NotificationData data = (NotificationData) notification;
-
+                            System.out.println("Received notification data");
                         } else {
                             LOGGER.warn("Unsupported Notification type");
                         }

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/chunk/PayloadConverter.java

@@ -25,6 +25,7 @@
 import org.apache.plc4x.java.opcua.readwrite.ExtensionObject;
 import org.apache.plc4x.java.opcua.readwrite.MessagePDU;
 import org.apache.plc4x.java.opcua.readwrite.Payload;
+import org.apache.plc4x.java.opcua.readwrite.RootExtensionObject;
 import org.apache.plc4x.java.spi.generation.ByteOrder;
 import org.apache.plc4x.java.spi.generation.Message;
 import org.apache.plc4x.java.spi.generation.ParseException;
@@ -56,7 +57,7 @@ public static ExtensiblePayload toExtensible(BinaryPayload binary) throws ParseE
         byte[] payload = binary.getPayload();
 
         ReadBufferByteBased buffer = new ReadBufferByteBased(payload, ByteOrder.LITTLE_ENDIAN);
-        ExtensionObject extensionObject = ExtensionObject.staticParse(buffer, false);
+        RootExtensionObject extensionObject = (RootExtensionObject) RootExtensionObject.staticParse(buffer, false);
 
         return new ExtensiblePayload(binary.getSequenceHeader(), extensionObject);
     }

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/readwrite/utils/StaticHelper.java

@@ -19,6 +19,8 @@
 package org.apache.plc4x.java.opcua.readwrite.utils;
 
 import java.nio.charset.StandardCharsets;
+import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
+import org.apache.plc4x.java.opcua.readwrite.ExpandedNodeId;
 
 public class StaticHelper {
 
@@ -27,14 +29,18 @@ public static int utf8LengthToPascalLength(String stringValue) {
         if (stringValue == null) {
             return -1;
         }
-        int nBytes = stringValue.getBytes(StandardCharsets.UTF_8).length;
-        if (nBytes == 0) {
-            return -1;
-        }
-        return nBytes;
+        return stringValue.getBytes(StandardCharsets.UTF_8).length;
     }
 
     public static int pascalLengthToUtf8Length(int slength) {
         return Math.max(slength, 0);
     }
+
+    public static int extensionId(ExpandedNodeId typeId) {
+        try {
+            return Integer.parseInt(typeId.getIdentifier());
+        } catch (NumberFormatException e) {
+            throw new PlcRuntimeException("Invalid node id, expected integer index, found " + typeId.getIdentifier().getClass());
+        }
+    }
 }

plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandleTest.java

@@ -148,6 +148,29 @@ public void subscribeDoesNotExists() throws Exception {
         subscriptionHandle.stopSubscriber();
     }
 
+    @Test
+    public void subscribeEvent() throws Exception {
+        // Create Subscription
+        PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+        builder.addEventTagAddress("ev1", "ns=0;i=2253");
+        PlcSubscriptionRequest request = builder.build();
+
+        // Get result of creating subscription
+        PlcSubscriptionResponse response = request.execute().get(1000, TimeUnit.MILLISECONDS);
+        final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle("ev1");
+
+        // Create handler for returned value
+        subscriptionHandle.register(plcSubscriptionEvent -> {
+            System.out.println("Subscription handle " + plcSubscriptionEvent);
+            assert plcSubscriptionEvent.getResponseCode("ev1").equals(PlcResponseCode.OK);
+        });
+
+        //Wait for value to be returned from server
+        Thread.sleep(1200);
+
+        subscriptionHandle.stopSubscriber();
+    }
+
     // ! If this test fails, see comment at the top of the class before investigating.
     @Test
     public void subscribeMultiple() throws Exception {

plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/protocol/chunk/PayloadConverterTest.java

@@ -28,6 +28,7 @@
 import org.apache.plc4x.java.opcua.readwrite.ExtensionObject;
 import org.apache.plc4x.java.opcua.readwrite.HistoryEvent;
 import org.apache.plc4x.java.opcua.readwrite.NodeIdFourByte;
+import org.apache.plc4x.java.opcua.readwrite.RootExtensionObject;
 import org.apache.plc4x.java.opcua.readwrite.SequenceHeader;
 import org.apache.plc4x.java.spi.utils.hex.Hex;
 import org.junit.jupiter.api.Test;
@@ -46,9 +47,8 @@ void convert() throws Exception {
             null
         );
 
-        ExtensionObject extObject = new ExtensionObject(
+        RootExtensionObject extObject = new RootExtensionObject(
             expandedNodeId,
-            null,
             new HistoryEvent(Collections.emptyList())
         );