You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by hu...@apache.org on 2021/05/26 10:35:11 UTC
[plc4x] branch feature/native_opua_client updated: Add test cases
for subscriptions, improved subscription disconnect.
This is an automated email from the ASF dual-hosted git repository.
hutcheb pushed a commit to branch feature/native_opua_client
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/feature/native_opua_client by this push:
new 0afe310 Add test cases for subscriptions, improved subscription disconnect.
0afe310 is described below
commit 0afe310137105f375fb75857faf554d33a446044
Author: hutcheb <be...@gmail.com>
AuthorDate: Wed May 26 06:34:17 2021 -0400
Add test cases for subscriptions, improved subscription disconnect.
---
.../language/java/JavaLanguageTemplateHelper.java | 23 -
.../templates/java/data-io-template.java.ftlh | 4 +-
.../resources/templates/java/io-template.java.ftlh | 7 -
.../templates/java/pojo-template.java.ftlh | 3 -
.../plc4x/java/opcua/context/SecureChannel.java | 81 ++++
.../context/SecureChannelTransactionManager.java | 16 +
.../java/opcua/protocol/OpcuaProtocolLogic.java | 15 +-
.../opcua/protocol/OpcuaSubscriptionHandle.java | 119 ++++-
.../protocol/OpcuaSubscriptionHandleTest.java | 498 +++++++++++++++++----
.../opcua/src/test/resources/log4j.properties | 24 +
plc4j/drivers/opcua/src/test/resources/logback.xml | 31 ++
.../plc4x/java/spi/generation/ReadBuffer.java | 6 -
.../plc4x/java/spi/generation/WriteBuffer.java | 6 -
13 files changed, 674 insertions(+), 159 deletions(-)
diff --git a/build-utils/language-java/src/main/java/org/apache/plc4x/language/java/JavaLanguageTemplateHelper.java b/build-utils/language-java/src/main/java/org/apache/plc4x/language/java/JavaLanguageTemplateHelper.java
index f6df481..19bbedd 100644
--- a/build-utils/language-java/src/main/java/org/apache/plc4x/language/java/JavaLanguageTemplateHelper.java
+++ b/build-utils/language-java/src/main/java/org/apache/plc4x/language/java/JavaLanguageTemplateHelper.java
@@ -503,34 +503,23 @@ public class JavaLanguageTemplateHelper extends BaseFreemarkerLanguageTemplateHe
}
public String toSerializationExpression(TypedField field, Term term, Argument[] serializerArgments) {
- System.out.println(field);
- System.out.println(term);
return toExpression(field, term, term1 -> toVariableSerializationExpression(field, term1, serializerArgments));
}
private String toExpression(TypedField field, Term term, Function<Term, String> variableExpressionGenerator) {
- System.out.println(field);
- System.out.println(term);
-
if (term == null) {
return "";
}
if (term instanceof Literal) {
if (term instanceof NullLiteral) {
- System.out.println("Blah10");
return "null";
} else if (term instanceof BooleanLiteral) {
- System.out.println("Blah11");
return Boolean.toString(((BooleanLiteral) term).getValue());
} else if (term instanceof NumericLiteral) {
- System.out.println("Blah12");
- System.out.println(((NumericLiteral) term).getNumber().toString());
return ((NumericLiteral) term).getNumber().toString();
} else if (term instanceof StringLiteral) {
- System.out.println("Blah13");
return "\"" + ((StringLiteral) term).getValue() + "\"";
} else if (term instanceof VariableLiteral) {
- System.out.println("Blah14");
VariableLiteral variableLiteral = (VariableLiteral) term;
// If this literal references an Enum type, then we have to output it differently.
if (getTypeDefinitions().get(variableLiteral.getName()) instanceof EnumTypeDefinition) {
@@ -544,7 +533,6 @@ public class JavaLanguageTemplateHelper extends BaseFreemarkerLanguageTemplateHe
throw new RuntimeException("Unsupported Literal type " + term.getClass().getName());
}
} else if (term instanceof UnaryTerm) {
- System.out.println("Blah15");
UnaryTerm ut = (UnaryTerm) term;
Term a = ut.getA();
switch (ut.getOperation()) {
@@ -558,7 +546,6 @@ public class JavaLanguageTemplateHelper extends BaseFreemarkerLanguageTemplateHe
throw new RuntimeException("Unsupported unary operation type " + ut.getOperation());
}
} else if (term instanceof BinaryTerm) {
- System.out.println("Blah16");
BinaryTerm bt = (BinaryTerm) term;
Term a = bt.getA();
Term b = bt.getB();
@@ -570,7 +557,6 @@ public class JavaLanguageTemplateHelper extends BaseFreemarkerLanguageTemplateHe
return "(" + toExpression(field, a, variableExpressionGenerator) + ") " + operation + " (" + toExpression(field, b, variableExpressionGenerator) + ")";
}
} else if (term instanceof TernaryTerm) {
- System.out.println("Blah17");
TernaryTerm tt = (TernaryTerm) term;
if ("if".equals(tt.getOperation())) {
Term a = tt.getA();
@@ -588,7 +574,6 @@ public class JavaLanguageTemplateHelper extends BaseFreemarkerLanguageTemplateHe
private String toVariableParseExpression(TypedField field, Term term, Argument[] parserArguments) {
VariableLiteral vl = (VariableLiteral) term;
// CAST expressions are special as we need to add a ".class" to the second parameter in Java.
- System.out.println("Blah1");
if ("CAST".equals(vl.getName())) {
StringBuilder sb = new StringBuilder(vl.getName());
if ((vl.getArgs() == null) || (vl.getArgs().size() != 2)) {
@@ -657,12 +642,10 @@ public class JavaLanguageTemplateHelper extends BaseFreemarkerLanguageTemplateHe
}
// If we are accessing implicit fields, we need to rely on a local variable instead.
else if (isVariableLiteralImplicitField(vl)) {
- System.out.println("Blah2");
return vl.getName();
}
// All uppercase names are not fields, but utility methods.
else if (vl.getName().equals(vl.getName().toUpperCase())) {
- System.out.println("Blah3");
StringBuilder sb = new StringBuilder(vl.getName());
if (vl.getArgs() != null) {
sb.append("(");
@@ -679,18 +662,12 @@ public class JavaLanguageTemplateHelper extends BaseFreemarkerLanguageTemplateHe
if (vl.getIndex() != VariableLiteral.NO_INDEX) {
sb.append("[").append(vl.getIndex()).append("]");
}
- System.out.println("Blah4");
return sb.toString() + ((vl.getChild() != null) ? "." + toVariableExpressionRest(vl.getChild()) : "");
}
- System.out.println("Blah5");
return vl.getName() + ((vl.getChild() != null) ? "." + toVariableExpressionRest(vl.getChild()) : "");
}
private String toVariableSerializationExpression(TypedField field, Term term, Argument[] serialzerArguments) {
- System.out.println(field);
- System.out.println(term);
- System.out.println(serialzerArguments);
-
VariableLiteral vl = (VariableLiteral) term;
if ("STATIC_CALL".equals(vl.getName())) {
StringBuilder sb = new StringBuilder();
diff --git a/build-utils/language-java/src/main/resources/templates/java/data-io-template.java.ftlh b/build-utils/language-java/src/main/resources/templates/java/data-io-template.java.ftlh
index b8a72bf..2aabc97 100644
--- a/build-utils/language-java/src/main/resources/templates/java/data-io-template.java.ftlh
+++ b/build-utils/language-java/src/main/resources/templates/java/data-io-template.java.ftlh
@@ -118,7 +118,7 @@ public class ${type.name}IO {
${field.name}.add(new ${helper.getPlcValueTypeForTypeReference(field.type)}((${helper.getNonPrimitiveLanguageTypeNameForField(field)}) <#if helper.isSimpleTypeReference(field.type)>${helper.getReadBufferReadMethodCall(field.type, "", field)})<#else>${field.type.name}IO.staticParse(io<#if field.params?has_content>, <#list field.params as parserArgument>(${helper.getLanguageTypeNameForTypeReference(helper.getArgumentType(field.type, parserArgument?index), true)}) (${hel [...]
}
}
- LOGGER.info("${field.name} - {}", ${field.name});
+
<#-- In all other cases do we have to work with a list, that is later converted to an array -->
<#else>
<#-- For a length array, we read data till the read position of the buffer reaches a given position -->
@@ -167,7 +167,6 @@ public class ${type.name}IO {
if(${field.name} != ${type.name}.${field.name?upper_case}) {
throw new ParseException("Expected constant value " + ${type.name}.${field.name?upper_case} + " but got " + ${field.name});
}
- LOGGER.info("${field.name} - {}", ${field.name});
<#if field.name == "value">
<#assign valueDefined=true>
</#if>
@@ -184,7 +183,6 @@ public class ${type.name}IO {
// Manual Field (${field.name})
${helper.getLanguageTypeNameForField(field)} ${field.name} = (${helper.getLanguageTypeNameForField(field)}) (${helper.toParseExpression(field, field.parseExpression, type.parserArguments)});
- LOGGER.info("${field.name} - {}", ${field.name});
<#if field.name == "value">
<#assign valueDefined=true>
</#if>
diff --git a/build-utils/language-java/src/main/resources/templates/java/io-template.java.ftlh b/build-utils/language-java/src/main/resources/templates/java/io-template.java.ftlh
index 9ad3550..75cf202 100644
--- a/build-utils/language-java/src/main/resources/templates/java/io-template.java.ftlh
+++ b/build-utils/language-java/src/main/resources/templates/java/io-template.java.ftlh
@@ -198,7 +198,6 @@ public class ${type.name}IO implements <#if outputFlavor != "passive">MessageIO<
}
}
- LOGGER.trace("${arrayField.name} - {}", ${arrayField.name});
<#-- In all other cases do we have to work with a list, that is later converted to an array -->
<#else>
@@ -257,7 +256,6 @@ public class ${type.name}IO implements <#if outputFlavor != "passive">MessageIO<
if(checksum != checksumRef) {
throw new ParseException(String.format("Checksum verification failed. Expected %04X but got %04X", checksumRef & 0xFFFF, checksum & 0xFFFF));
}
- LOGGER.trace("CheckSum - {}", checksum);
}
<#break>
<#case "const">
@@ -269,7 +267,6 @@ public class ${type.name}IO implements <#if outputFlavor != "passive">MessageIO<
if(${constField.name} != ${type.name}.${constField.name?upper_case}) {
throw new ParseException("Expected constant value " + ${type.name}.${constField.name?upper_case} + " but got " + ${constField.name});
}
- LOGGER.trace("${constField.name} - {}", ${constField.name});
<#break>
<#case "discriminator">
@@ -364,7 +361,6 @@ public class ${type.name}IO implements <#if outputFlavor != "passive">MessageIO<
// Manual Field (${manualField.name})
${helper.getLanguageTypeNameForField(field)} ${manualField.name} = (${helper.getLanguageTypeNameForField(field)}) (${helper.toParseExpression(manualField, manualField.parseExpression, type.parserArguments)});
- LOGGER.trace("${manualField.name} - {}", ${manualField.name});
<#break>
<#case "optional">
<#assign optionalField = field>
@@ -381,7 +377,6 @@ public class ${type.name}IO implements <#if outputFlavor != "passive">MessageIO<
${optionalField.name} = <#if helper.isSimpleTypeReference(optionalField.type)>${helper.getReadBufferReadMethodCall(optionalField.type, "", optionalField)}<#else>${optionalField.type.name}IO.staticParse(io<#if field.params?has_content>, <#list field.params as parserArgument>(${helper.getLanguageTypeNameForTypeReference(helper.getArgumentType(optionalField.type, parserArgument?index), true)}) (${helper.toParseExpression(optionalField, parserArgument, type.parserArguments)})<#se [...]
</#if>
}
- LOGGER.trace("${optionalField.name} - {}", ${optionalField.name});
<#break>
<#case "padding">
<#assign paddingField = field>
@@ -437,7 +432,6 @@ public class ${type.name}IO implements <#if outputFlavor != "passive">MessageIO<
</...@compress>
</#if>
- LOGGER.trace("${simpleField.name} - {}", ${simpleField.name});
<#break>
<#case "switch">
<#assign switchField = field>
@@ -634,7 +628,6 @@ public class ${type.name}IO implements <#if outputFlavor != "passive">MessageIO<
${complexTypeReference.name}IO.staticSerialize(io, ${simpleField.name});
</#if>
</#if>
- LOGGER.trace("${simpleField.name} - {} - Position {}", ${simpleField.name}, io.getPos());
<#break>
<#case "switch">
<#assign switchField = field>
diff --git a/build-utils/language-java/src/main/resources/templates/java/pojo-template.java.ftlh b/build-utils/language-java/src/main/resources/templates/java/pojo-template.java.ftlh
index 9f7c5ad..f61ff7e 100644
--- a/build-utils/language-java/src/main/resources/templates/java/pojo-template.java.ftlh
+++ b/build-utils/language-java/src/main/resources/templates/java/pojo-template.java.ftlh
@@ -204,9 +204,6 @@ public<#if helper.isDiscriminatedParentTypeDefinition()> abstract</#if> class ${
<#assign simpleTypeReference = arrayField.type>
for(${arrayField.type.name} element : ${arrayField.name}) {
boolean last = ++i >= ${arrayField.name}.length;
- System.out.println(last);
- System.out.println(element);
- System.out.println(element.getLengthInBitsConditional(last));
lengthInBits += element.getLengthInBitsConditional(last);
}
<#else>
diff --git a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannel.java b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannel.java
index 59bec79..233fb95 100644
--- a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannel.java
+++ b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannel.java
@@ -308,6 +308,15 @@ public class SecureChannel {
.unwrap(apuMessage -> encryptionHandler.decodeMessage(apuMessage))
.check(p -> p.getMessage() instanceof OpcuaOpenResponse)
.unwrap(p -> (OpcuaOpenResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaOpenResponse -> {
try {
ReadBuffer readBuffer = new ReadBuffer(opcuaOpenResponse.getMessage(), true);
@@ -427,6 +436,15 @@ public class SecureChannel {
.expectResponse(OpcuaAPU.class, REQUEST_TIMEOUT)
.check(p -> p.getMessage() instanceof OpcuaMessageResponse)
.unwrap(p -> (OpcuaMessageResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaMessageResponse -> {
try {
ExtensionObject message = ExtensionObjectIO.staticParse(new ReadBuffer(opcuaMessageResponse.getMessage(), true), false);
@@ -552,6 +570,15 @@ public class SecureChannel {
.expectResponse(OpcuaAPU.class, REQUEST_TIMEOUT)
.check(p -> p.getMessage() instanceof OpcuaMessageResponse)
.unwrap(p -> (OpcuaMessageResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaActivateResponse -> {
LOGGER.debug("Got Activate Session Response Connection Response");
try {
@@ -633,6 +660,15 @@ public class SecureChannel {
.expectResponse(OpcuaAPU.class, REQUEST_TIMEOUT)
.check(p -> p.getMessage() instanceof OpcuaMessageResponse)
.unwrap(p -> (OpcuaMessageResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaMessageResponse -> {
LOGGER.trace("Got Close Session Response Connection Response" + opcuaMessageResponse.toString());
onDisconnectCloseSecureChannel(context);
@@ -680,6 +716,15 @@ public class SecureChannel {
.expectResponse(OpcuaAPU.class, REQUEST_TIMEOUT)
.check(p -> p.getMessage() instanceof OpcuaMessageResponse)
.unwrap(p -> (OpcuaMessageResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaMessageResponse -> {
LOGGER.trace("Got Close Secure Channel Response" + opcuaMessageResponse.toString());
});
@@ -763,6 +808,15 @@ public class SecureChannel {
.expectResponse(OpcuaAPU.class, REQUEST_TIMEOUT)
.check(p -> p.getMessage() instanceof OpcuaOpenResponse)
.unwrap(p -> (OpcuaOpenResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaOpenResponse -> {
try {
ExtensionObject message = ExtensionObjectIO.staticParse(new ReadBuffer(opcuaOpenResponse.getMessage(), true), false);
@@ -845,6 +899,15 @@ public class SecureChannel {
.expectResponse(OpcuaAPU.class, REQUEST_TIMEOUT)
.check(p -> p.getMessage() instanceof OpcuaMessageResponse)
.unwrap(p -> (OpcuaMessageResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaMessageResponse -> {
try {
ExtensionObject message = ExtensionObjectIO.staticParse(new ReadBuffer(opcuaMessageResponse.getMessage(), true), false);
@@ -921,6 +984,15 @@ public class SecureChannel {
.expectResponse(OpcuaAPU.class, REQUEST_TIMEOUT)
.check(p -> p.getMessage() instanceof OpcuaMessageResponse)
.unwrap(p -> (OpcuaMessageResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaMessageResponse -> {
LOGGER.trace("Got Close Secure Channel Response" + opcuaMessageResponse.toString());
// Send an event that connection setup is complete.
@@ -1008,6 +1080,15 @@ public class SecureChannel {
.unwrap(apuMessage -> encryptionHandler.decodeMessage(apuMessage))
.check(p -> p.getMessage() instanceof OpcuaOpenResponse)
.unwrap(p -> (OpcuaOpenResponse) p.getMessage())
+ .check(p -> {
+ LOGGER.info("{}", p.getSequenceNumber());
+ LOGGER.info("{}", transactionId);
+ if (p.getRequestId() == transactionId) {
+ return true;
+ } else {
+ return false;
+ }
+ })
.handle(opcuaOpenResponse -> {
try {
ReadBuffer readBuffer = new ReadBuffer(opcuaOpenResponse.getMessage(), true);
diff --git a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannelTransactionManager.java b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannelTransactionManager.java
index a033dec..c7c852b 100644
--- a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannelTransactionManager.java
+++ b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/context/SecureChannelTransactionManager.java
@@ -46,6 +46,7 @@ public class SecureChannelTransactionManager {
private AtomicInteger transactionIdentifierGenerator = new AtomicInteger(0);
+ private AtomicInteger requestIdentifierGenerator = new AtomicInteger(0);
private AtomicInteger activeTransactionId = new AtomicInteger(0);
private Map<Integer, Transaction> queue = new HashMap<>();
@@ -84,6 +85,21 @@ public class SecureChannelTransactionManager {
}
/**
+ * Returns the next request identifier.
+ *
+ * @return the next sequential request identifier
+ */
+ public int getRequestIdentifier() {
+ int requestId = requestIdentifierGenerator.getAndIncrement();
+ if(requestIdentifierGenerator.get() == DEFAULT_MAX_REQUEST_ID) {
+ requestIdentifierGenerator.set(1);
+ }
+ return requestId;
+ }
+
+
+
+ /**
* Returns the next transaction identifier.
*
* @return the next sequential transaction identifier
diff --git a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaProtocolLogic.java b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaProtocolLogic.java
index 0368587..dc751c6 100644
--- a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaProtocolLogic.java
+++ b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaProtocolLogic.java
@@ -786,13 +786,8 @@ public class OpcuaProtocolLogic extends Plc4xProtocolBase<OpcuaAPU> implements H
CreateSubscriptionResponse response = subscription.get(SecureChannel.REQUEST_TIMEOUT_LONG, TimeUnit.MILLISECONDS);
subscriptionId = response.getSubscriptionId();
subscriptions.put(subscriptionId, new OpcuaSubscriptionHandle(context, this, channel, subscriptionRequest, subscriptionId, cycleTime));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- LOGGER.warn("Unable to subscribe Elements because of: {}", e.getMessage());
- } catch (ExecutionException e) {
- LOGGER.warn("Unable to subscribe Elements because of: {}", e.getMessage());
- } catch (TimeoutException e) {
- LOGGER.warn("Unable to subscribe Elements because of: {}", e.getMessage());
+ } catch (Exception e) {
+ throw new PlcRuntimeException("Unable to subscribe because of: " + e.getMessage());
}
for (String fieldName : subscriptionRequest.getFieldNames()) {
@@ -811,6 +806,7 @@ public class OpcuaProtocolLogic extends Plc4xProtocolBase<OpcuaAPU> implements H
private CompletableFuture<CreateSubscriptionResponse> onSubscribeCreateSubscription(long cycleTime) {
CompletableFuture<CreateSubscriptionResponse> future = new CompletableFuture<>();
+ LOGGER.trace("Enetering creating subscription request");
RequestHeader requestHeader = new RequestHeader(channel.getAuthenticationToken(),
SecureChannel.getCurrentDateTime(),
@@ -891,11 +887,14 @@ public class OpcuaProtocolLogic extends Plc4xProtocolBase<OpcuaAPU> implements H
unsubscriptionRequest.getSubscriptionHandles().forEach(o -> {
OpcuaSubscriptionHandle opcuaSubHandle = (OpcuaSubscriptionHandle) o;
opcuaSubHandle.stopSubscriber();
- subscriptions.remove(opcuaSubHandle);
});
return null;
}
+ public void removeSubscription(Long subscriptionId) {
+ subscriptions.remove(subscriptionId);
+ }
+
@Override
public PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer, Collection<PlcSubscriptionHandle> handles) {
List<PlcConsumerRegistration> registrations = new LinkedList<>();
diff --git a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java
index f9fcda6..6d5f961 100644
--- a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java
+++ b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java
@@ -145,7 +145,7 @@ public class OpcuaSubscriptionHandle extends DefaultPlcSubscriptionHandle {
CreateMonitoredItemsRequest createMonitoredItemsRequest = new CreateMonitoredItemsRequest(
requestHeader,
subscriptionId,
- TimestampsToReturn.timestampsToReturnBoth,
+ TimestampsToReturn.timestampsToReturnNeither,
requestList.length,
requestList
);
@@ -184,7 +184,7 @@ public class OpcuaSubscriptionHandle extends DefaultPlcSubscriptionHandle {
}
for (MonitoredItemCreateResult result : Arrays.stream(responseMessage.getResults()).toArray(MonitoredItemCreateResult[]::new)) {
if (OpcuaStatusCode.enumForValue(result.getStatusCode().getStatusCode()) != OpcuaStatusCode.Good) {
- throw new PlcInvalidFieldException(fieldNames.get((int) result.getMonitoredItemId()));
+ LOGGER.error("Invalid Field {}, subscription created without this field", fieldNames.get((int) result.getMonitoredItemId()));
} else {
LOGGER.debug("Field {} was added to the subscription", fieldNames.get((int) result.getMonitoredItemId() - 1));
}
@@ -214,9 +214,9 @@ public class OpcuaSubscriptionHandle extends DefaultPlcSubscriptionHandle {
return future;
}
- private void sleep() {
+ private void sleep(long length) {
try {
- Thread.sleep(this.revisedCycleTime);
+ Thread.sleep(length);
} catch (InterruptedException e) {
LOGGER.trace("Interrupted Exception");
}
@@ -279,35 +279,38 @@ public class OpcuaSubscriptionHandle extends DefaultPlcSubscriptionHandle {
/* Create Consumer for the response message, error and timeout to be sent to the Secure Channel */
Consumer<OpcuaMessageResponse> consumer = opcuaResponse -> {
PublishResponse responseMessage = null;
+ ServiceFault serviceFault = null;
try {
ExtensionObjectDefinition unknownExtensionObject = ExtensionObjectIO.staticParse(new ReadBuffer(opcuaResponse.getMessage(), true), false).getBody();
if (unknownExtensionObject instanceof PublishResponse) {
responseMessage = (PublishResponse) unknownExtensionObject;
} else {
- ServiceFault serviceFault = (ServiceFault) unknownExtensionObject;
+ serviceFault = (ServiceFault) unknownExtensionObject;
ResponseHeader header = (ResponseHeader) serviceFault.getResponseHeader();
- LOGGER.error("Subscription ServiceFault return from server with error code, '{}'", header.getServiceResult().toString());
- plcSubscriber.onDisconnect(context);
+ LOGGER.error("Subscription ServiceFault returned from server with error code, '{}'", header.getServiceResult().toString());
+ //plcSubscriber.onDisconnect(context);
}
} catch (ParseException e) {
LOGGER.error("Unable to parse the returned Subscription response");
e.printStackTrace();
plcSubscriber.onDisconnect(context);
}
- outstandingRequests.remove(((ResponseHeader) responseMessage.getResponseHeader()).getRequestHandle());
+ if (serviceFault == null) {
+ outstandingRequests.remove(((ResponseHeader) responseMessage.getResponseHeader()).getRequestHandle());
- for (long availableSequenceNumber : responseMessage.getAvailableSequenceNumbers()) {
- outstandingAcknowledgements.add(new SubscriptionAcknowledgement(this.subscriptionId, availableSequenceNumber));
- }
+ for (long availableSequenceNumber : responseMessage.getAvailableSequenceNumbers()) {
+ outstandingAcknowledgements.add(new SubscriptionAcknowledgement(this.subscriptionId, availableSequenceNumber));
+ }
- for (ExtensionObject notificationMessage : ((NotificationMessage) responseMessage.getNotificationMessage()).getNotificationData()) {
- ExtensionObjectDefinition notification = notificationMessage.getBody();
- if (notification instanceof DataChangeNotification) {
- LOGGER.trace("Found a Data Change notification");
- ExtensionObjectDefinition[] items = ((DataChangeNotification) notification).getMonitoredItems();
- onSubscriptionValue(Arrays.stream(items).toArray(MonitoredItemNotification[]::new));
- } else {
- LOGGER.warn("Unsupported Notification type");
+ for (ExtensionObject notificationMessage : ((NotificationMessage) responseMessage.getNotificationMessage()).getNotificationData()) {
+ ExtensionObjectDefinition notification = notificationMessage.getBody();
+ if (notification instanceof DataChangeNotification) {
+ LOGGER.trace("Found a Data Change notification");
+ ExtensionObjectDefinition[] items = ((DataChangeNotification) notification).getMonitoredItems();
+ onSubscriptionValue(Arrays.stream(items).toArray(MonitoredItemNotification[]::new));
+ } else {
+ LOGGER.warn("Unsupported Notification type");
+ }
}
}
};
@@ -333,10 +336,10 @@ public class OpcuaSubscriptionHandle extends DefaultPlcSubscriptionHandle {
}
}
/* Put the subscriber loop to sleep for the rest of the cycle. */
- sleep();
+ sleep(this.revisedCycleTime);
}
- //Wait for any outstanding responses to arrive
- sleep();
+ //Wait for any outstanding responses to arrive, using the request timeout length
+ //sleep(this.revisedCycleTime * 10);
complete = true;
} catch (Exception e) {
LOGGER.error("Failed :(");
@@ -354,9 +357,77 @@ public class OpcuaSubscriptionHandle extends DefaultPlcSubscriptionHandle {
*/
public void stopSubscriber() {
this.destroy.set(true);
- while (!complete) {
- sleep();
+
+ long requestHandle = channel.getRequestHandle();
+
+ RequestHeader requestHeader = new RequestHeader(channel.getAuthenticationToken(),
+ SecureChannel.getCurrentDateTime(),
+ requestHandle,
+ 0L,
+ OpcuaProtocolLogic.NULL_STRING,
+ this.revisedCycleTime * 10,
+ OpcuaProtocolLogic.NULL_EXTENSION_OBJECT);
+
+ long[] subscriptions = new long[1];
+ subscriptions[0] = subscriptionId;
+ DeleteSubscriptionsRequest deleteSubscriptionrequest = new DeleteSubscriptionsRequest(requestHeader,
+ 1,
+ subscriptions
+ );
+
+ ExpandedNodeId extExpandedNodeId = new ExpandedNodeId(false, //Namespace Uri Specified
+ false, //Server Index Specified
+ new NodeIdFourByte((short) 0, Integer.valueOf(deleteSubscriptionrequest.getIdentifier())),
+ null,
+ null);
+
+ ExtensionObject extObject = new ExtensionObject(
+ extExpandedNodeId,
+ null,
+ deleteSubscriptionrequest);
+
+ try {
+ WriteBuffer buffer = new WriteBuffer(extObject.getLengthInBytes(), true);
+ ExtensionObjectIO.staticSerialize(buffer, extObject);
+
+ // Create Consumer for the response message, error and timeout to be sent to the Secure Channel
+ Consumer<OpcuaMessageResponse> consumer = opcuaResponse -> {
+ DeleteSubscriptionsResponse responseMessage = null;
+ try {
+ ExtensionObjectDefinition unknownExtensionObject = ExtensionObjectIO.staticParse(new ReadBuffer(opcuaResponse.getMessage(), true), false).getBody();
+ if (unknownExtensionObject instanceof DeleteSubscriptionsResponse) {
+ responseMessage = (DeleteSubscriptionsResponse) unknownExtensionObject;
+ } else {
+ ServiceFault serviceFault = (ServiceFault) unknownExtensionObject;
+ ResponseHeader header = (ResponseHeader) serviceFault.getResponseHeader();
+ LOGGER.error("Fault when deleteing Subscription ServiceFault return from server with error code, '{}'", header.getServiceResult().toString());
+ }
+ } catch (ParseException e) {
+ LOGGER.error("Unable to parse the returned Delete Subscriptions Response");
+ e.printStackTrace();
+ }
+ };
+
+ Consumer<TimeoutException> timeout = e -> {
+ LOGGER.error("Timeout while waiting for delete subscription response");
+ e.printStackTrace();
+ plcSubscriber.onDisconnect(context);
+ };
+
+ BiConsumer<OpcuaAPU, Throwable> error = (message, e) -> {
+ LOGGER.error("Error while waiting for delete subscription response");
+ e.printStackTrace();
+ plcSubscriber.onDisconnect(context);
+ };
+
+ channel.submit(context, timeout, error, consumer, buffer);
+ } catch (ParseException e) {
+ LOGGER.warn("Unable to serialize subscription request");
+ e.printStackTrace();
}
+
+ sleep(500);
+ plcSubscriber.removeSubscription(subscriptionId);
}
/**
diff --git a/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandleTest.java b/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandleTest.java
index 26abf38..ca90e22 100644
--- a/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandleTest.java
+++ b/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandleTest.java
@@ -40,14 +40,14 @@ public class OpcuaSubscriptionHandleTest {
private static ExampleServer exampleServer;
// Address of local milo server
- private String miloLocalAddress = "127.0.0.1:12686/milo";
+ private static String miloLocalAddress = "127.0.0.1:12686/milo";
//Tcp pattern of OPC UA
- private String opcPattern = "opcua:tcp://";
+ private static String opcPattern = "opcua:tcp://";
private String paramSectionDivider = "?";
private String paramDivider = "&";
- private String tcpConnectionAddress = opcPattern + miloLocalAddress;
+ private static String tcpConnectionAddress = opcPattern + miloLocalAddress;
// Read only variables of milo example server of version 3.6
private static final String BOOL_IDENTIFIER_READ_WRITE = "ns=2;s=HelloWorld/ScalarTypes/Boolean";
@@ -66,6 +66,8 @@ public class OpcuaSubscriptionHandleTest {
private static final String UINTEGER_IDENTIFIER_READ_WRITE = "ns=2;s=HelloWorld/ScalarTypes/UInteger";
private static final String DOES_NOT_EXIST_IDENTIFIER_READ_WRITE = "ns=2;i=12512623";
+ private static PlcConnection opcuaConnection;
+
@BeforeEach
public void before() {
}
@@ -80,6 +82,9 @@ public class OpcuaSubscriptionHandleTest {
try {
exampleServer = new ExampleServer();
exampleServer.startup().get();
+ //Connect
+ opcuaConnection = new PlcDriverManager().getConnection(tcpConnectionAddress);
+ assert opcuaConnection.isConnected();
} catch (Exception e) {
}
@@ -88,6 +93,10 @@ public class OpcuaSubscriptionHandleTest {
@AfterAll
public static void tearDown() {
try {
+ // Close Connection
+ opcuaConnection.close();
+ assert !opcuaConnection.isConnected();
+
exampleServer.shutdown().get();
} catch (Exception e) {
@@ -95,108 +104,439 @@ public class OpcuaSubscriptionHandleTest {
}
@Test
- public void subscribeVariables() throws Exception {
- PlcConnection opcuaConnection = new PlcDriverManager().getConnection(tcpConnectionAddress);
- assert opcuaConnection.isConnected();
+ public void subscribeBool() throws Exception {
+ String field = "Bool";
+ String identifier = BOOL_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+ // Create Subscription
PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
- builder.addChangeOfStateField("Bool", BOOL_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Byte", BYTE_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Double", DOUBLE_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Float", FLOAT_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Int16", INT16_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Int32", INT32_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Int64", INT64_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Integer", INTEGER_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("SByte", SBYTE_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("String", STRING_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("UInt16", UINT16_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("UInt32", UINT32_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("UInt64", UINT64_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("UInteger", UINTEGER_IDENTIFIER_READ_WRITE);
- //builder.addChangeOfStateField("DoesNotExists", DOES_NOT_EXIST_IDENTIFIER_READ_WRITE);
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeByte() throws Exception {
+ String field = "Byte";
+ String identifier = BYTE_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
PlcSubscriptionRequest request = builder.build();
+ // Get result of creating subscription
PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
- for (String subscriptionName : response.getFieldNames()) {
- final PlcSubscriptionHandle subscriptionHandle = response.getSubscriptionHandle(subscriptionName);
- subscriptionHandle.register(plcSubscriptionEvent -> {
- for (String fieldName : plcSubscriptionEvent.getFieldNames()) {
- assert plcSubscriptionEvent.getResponseCode(fieldName).equals(PlcResponseCode.OK);
- }
- });
- }
- Thread.sleep(10000);
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeDouble() throws Exception {
+ String field = "Double";
+ String identifier = DOUBLE_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
- opcuaConnection.close();
- assert !opcuaConnection.isConnected();
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
}
@Test
- public void dualSubscriptions() throws Exception {
- PlcConnection opcuaConnection = new PlcDriverManager().getConnection(tcpConnectionAddress);
- assert opcuaConnection.isConnected();
+ public void subscribeFloat() throws Exception {
+ String field = "Float";
+ String identifier = FLOAT_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+ // Create Subscription
PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
- builder.addChangeOfStateField("Bool", BOOL_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Byte", BYTE_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Double", DOUBLE_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Float", FLOAT_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Int16", INT16_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Int32", INT32_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Int64", INT64_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("Integer", INTEGER_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("SByte", SBYTE_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("String", STRING_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("UInt16", UINT16_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("UInt32", UINT32_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("UInt64", UINT64_IDENTIFIER_READ_WRITE);
- builder.addChangeOfStateField("UInteger", UINTEGER_IDENTIFIER_READ_WRITE);
- //builder.addChangeOfStateField("DoesNotExists", DOES_NOT_EXIST_IDENTIFIER_READ_WRITE);
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+ @Test
+ public void subscribeInt16() throws Exception {
+ String field = "Int16";
+ String identifier = INT16_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
PlcSubscriptionRequest request = builder.build();
+ // Get result of creating subscription
PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
- PlcSubscriptionRequest.Builder builderTwo = opcuaConnection.subscriptionRequestBuilder();
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
- builderTwo.addChangeOfStateField("Bool", BOOL_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("Byte", BYTE_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("Double", DOUBLE_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("Float", FLOAT_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("Int16", INT16_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("Int32", INT32_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("Int64", INT64_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("Integer", INTEGER_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("SByte", SBYTE_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("String", STRING_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("UInt16", UINT16_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("UInt32", UINT32_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("UInt64", UINT64_IDENTIFIER_READ_WRITE);
- builderTwo.addChangeOfStateField("UInteger", UINTEGER_IDENTIFIER_READ_WRITE);
- //builder.addChangeOfStateField("DoesNotExists", DOES_NOT_EXIST_IDENTIFIER_READ_WRITE);
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
- PlcSubscriptionRequest requestTwo = builderTwo.build();
+ subscriptionHandle.stopSubscriber();
+ }
- PlcSubscriptionResponse responseTwo = requestTwo.execute().get();
+ @Test
+ public void subscribeInt32() throws Exception {
+ String field = "Int32";
+ String identifier = INT32_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
- for (String subscriptionName : response.getFieldNames()) {
- final PlcSubscriptionHandle subscriptionHandle = response.getSubscriptionHandle(subscriptionName);
- subscriptionHandle.register(plcSubscriptionEvent -> {
- for (String fieldName : plcSubscriptionEvent.getFieldNames()) {
- assert plcSubscriptionEvent.getResponseCode(fieldName).equals(PlcResponseCode.OK);
- }
- });
- }
- Thread.sleep(1500);
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeInt64() throws Exception {
+ String field = "Int64";
+ String identifier = INT64_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeInteger() throws Exception {
+ String field = "Integer";
+ String identifier = INTEGER_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeSByte() throws Exception {
+ String field = "SByte";
+ String identifier = SBYTE_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeString() throws Exception {
+ String field = "String";
+ String identifier = STRING_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeUInt16() throws Exception {
+ String field = "Uint16";
+ String identifier = UINT16_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
- opcuaConnection.close();
- assert !opcuaConnection.isConnected();
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
}
+
+ @Test
+ public void subscribeUInt32() throws Exception {
+ String field = "UInt32";
+ String identifier = UINT32_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeUInt64() throws Exception {
+ String field = "UInt64";
+ String identifier = UINT64_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeUInteger() throws Exception {
+ String field = "UInteger";
+ String identifier = UINTEGER_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field).equals(PlcResponseCode.OK);
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeDoesNotExists() throws Exception {
+ String field = "DoesNotExists";
+ String identifier = DOES_NOT_EXIST_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} test", field);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field, identifier);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ //This should never be called,
+ assert false;
+ LOGGER.info("Received a response from {} test {}", field, plcSubscriptionEvent.getPlcValue(field).toString());
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
+ @Test
+ public void subscribeMultiple() throws Exception {
+ String field1 = "UInteger";
+ String identifier1 = UINTEGER_IDENTIFIER_READ_WRITE;
+ String field2 = "Integer";
+ String identifier2 = INTEGER_IDENTIFIER_READ_WRITE;
+ LOGGER.info("Starting subscription {} and {} test", field1, field2);
+
+ // Create Subscription
+ PlcSubscriptionRequest.Builder builder = opcuaConnection.subscriptionRequestBuilder();
+ builder.addChangeOfStateField(field1, identifier1);
+ builder.addChangeOfStateField(field2, identifier2);
+ PlcSubscriptionRequest request = builder.build();
+
+ // Get result of creating subscription
+ PlcSubscriptionResponse response = request.execute().get();
+ final OpcuaSubscriptionHandle subscriptionHandle = (OpcuaSubscriptionHandle) response.getSubscriptionHandle(field1);
+
+ // Create handler for returned value
+ subscriptionHandle.register(plcSubscriptionEvent -> {
+ assert plcSubscriptionEvent.getResponseCode(field1).equals(PlcResponseCode.OK);
+ assert plcSubscriptionEvent.getResponseCode(field2).equals(PlcResponseCode.OK);
+ });
+
+ //Wait for value to be returned from server
+ Thread.sleep(1200);
+
+ subscriptionHandle.stopSubscriber();
+ }
+
}
diff --git a/plc4j/drivers/opcua/src/test/resources/log4j.properties b/plc4j/drivers/opcua/src/test/resources/log4j.properties
new file mode 100644
index 0000000..593679f
--- /dev/null
+++ b/plc4j/drivers/opcua/src/test/resources/log4j.properties
@@ -0,0 +1,24 @@
+#
+# 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.
+#
+appender.out.type=Console
+appender.out.name=out
+appender.out.layout.type=PatternLayout
+appender.out.layout.pattern=[%30.30t] %-30.30c{1} %-5p %m%n
+rootLogger.level=TRACE
+rootLogger.appenderRef.out.ref=out
diff --git a/plc4j/drivers/opcua/src/test/resources/logback.xml b/plc4j/drivers/opcua/src/test/resources/logback.xml
new file mode 100644
index 0000000..99d73cb
--- /dev/null
+++ b/plc4j/drivers/opcua/src/test/resources/logback.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+
+<configuration>
+ <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>%d %5p | %t | %-55logger{55} | %m %n</pattern>
+ </encoder>
+ </appender>
+ <root>
+ <level value="TRACE"/>
+ <appender-ref ref="CONSOLE"/>
+ </root>
+</configuration>
\ No newline at end of file
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/generation/ReadBuffer.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/generation/ReadBuffer.java
index 3e77df0..dbb6f76 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/generation/ReadBuffer.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/generation/ReadBuffer.java
@@ -28,13 +28,8 @@ import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.charset.Charset;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
public class ReadBuffer {
- private static final Logger LOGGER = LoggerFactory.getLogger(ReadBuffer.class);
-
private final MyDefaultBitInput bi;
private final boolean littleEndian;
private final long totalBytes;
@@ -44,7 +39,6 @@ public class ReadBuffer {
}
public ReadBuffer(byte[] input, boolean littleEndian) {
- LOGGER.info("Creating read buffer " + input.length);
ArrayByteInput abi = new ArrayByteInput(input);
this.bi = new MyDefaultBitInput(abi);
this.littleEndian = littleEndian;
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/generation/WriteBuffer.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/generation/WriteBuffer.java
index 38163c0..21fee13 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/generation/WriteBuffer.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/generation/WriteBuffer.java
@@ -28,13 +28,8 @@ import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
public class WriteBuffer {
- private static final Logger LOGGER = LoggerFactory.getLogger(WriteBuffer.class);
-
private final ByteBuffer bb;
private final BufferByteOutput bbo;
private final MyDefaultBitOutput bo;
@@ -45,7 +40,6 @@ public class WriteBuffer {
}
public WriteBuffer(int size, boolean littleEndian) {
- LOGGER.info("Creating write buffer " + size);
bb = ByteBuffer.allocate(size);
bbo = new BufferByteOutput(bb);
bo = new MyDefaultBitOutput(bbo);