Skip to content

Commit

Permalink
fix(plc4j/opcua): More rebust MonitoredItem handling within subscript…
Browse files Browse the repository at this point in the history 
…ion handle.

Fix for PLC4X-342. Minor adjustments of logging so stack traces go through slf4j.

Signed-off-by: Łukasz Dywicki <luke@code-house.org>
  • Loading branch information
@splatch
splatch committed May 15, 2022
1 parent e1629b9 commit 066c395
Showing 1 changed file with 20 additions and 32 deletions.
52 changes: 20 additions & 32 deletions ...ers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java
View file
Expand Up @@ -77,8 +77,7 @@ public OpcuaSubscriptionHandle(ConversationContext<OpcuaAPU> context, OpcuaProto
try {
onSubscribeCreateMonitoredItemsRequest().get();
} catch (Exception e) {
LOGGER.info("Unable to serialize the Create Monitored Item Subscription Message");
e.printStackTrace();
LOGGER.info("Unable to serialize the Create Monitored Item Subscription Message", e);
plcSubscriber.onDisconnect(context);
}
startSubscriber();
Expand Down Expand Up @@ -141,7 +140,7 @@ private CompletableFuture<CreateMonitoredItemsResponse> onSubscribeCreateMonitor
CreateMonitoredItemsRequest createMonitoredItemsRequest = new CreateMonitoredItemsRequest(
requestHeader,
subscriptionId,
TimestampsToReturn.timestampsToReturnNeither,
TimestampsToReturn.timestampsToReturnBoth,
requestList.size(),
requestList
);
Expand Down Expand Up @@ -175,37 +174,35 @@ private CompletableFuture<CreateMonitoredItemsResponse> onSubscribeCreateMonitor
plcSubscriber.onDisconnect(context);
}
} catch (ParseException e) {
LOGGER.error("Unable to parse the returned Subscription response");
e.printStackTrace();
LOGGER.error("Unable to parse the returned Subscription response", e);
plcSubscriber.onDisconnect(context);
}
for (MonitoredItemCreateResult result : responseMessage.getResults().toArray(new MonitoredItemCreateResult[0])) {
MonitoredItemCreateResult[] array = responseMessage.getResults().toArray(new MonitoredItemCreateResult[0]);
for (int index = 0, arrayLength = array.length; index < arrayLength; index++) {
MonitoredItemCreateResult result = array[index];
if (OpcuaStatusCode.enumForValue(result.getStatusCode().getStatusCode()) != OpcuaStatusCode.Good) {
LOGGER.error("Invalid Field {}, subscription created without this field", fieldNames.get((int) result.getMonitoredItemId()));
LOGGER.error("Invalid Field {}, subscription created without this field", fieldNames.get(index));
} else {
LOGGER.debug("Field {} was added to the subscription", fieldNames.get((int) result.getMonitoredItemId() - 1));
LOGGER.debug("Field {} was added to the subscription", fieldNames.get(index));
}
}
future.complete(responseMessage);
};

Consumer<TimeoutException> timeout = e -> {
LOGGER.info("Timeout while sending the Create Monitored Item Subscription Message");
e.printStackTrace();
LOGGER.info("Timeout while sending the Create Monitored Item Subscription Message", e);
plcSubscriber.onDisconnect(context);
};

BiConsumer<OpcuaAPU, Throwable> error = (message, e) -> {
LOGGER.info("Error while sending the Create Monitored Item Subscription Message");
e.printStackTrace();
LOGGER.info("Error while sending the Create Monitored Item Subscription Message", e);
plcSubscriber.onDisconnect(context);
};

channel.submit(context, timeout, error, consumer, buffer);

} catch (SerializationException e) {
LOGGER.info("Unable to serialize the Create Monitored Item Subscription Message");
e.printStackTrace();
LOGGER.info("Unable to serialize the Create Monitored Item Subscription Message", e);
plcSubscriber.onDisconnect(context);
}
return future;
Expand Down Expand Up @@ -287,8 +284,7 @@ public void startSubscriber() {
//plcSubscriber.onDisconnect(context);
}
} catch (ParseException e) {
LOGGER.error("Unable to parse the returned Subscription response");
e.printStackTrace();
LOGGER.error("Unable to parse the returned Subscription response", e);
plcSubscriber.onDisconnect(context);
}
if (serviceFault == null) {
Expand All @@ -312,23 +308,20 @@ public void startSubscriber() {
};

Consumer<TimeoutException> timeout = e -> {
LOGGER.error("Timeout while waiting for subscription response");
e.printStackTrace();
LOGGER.error("Timeout while waiting for subscription response", e);
plcSubscriber.onDisconnect(context);
};

BiConsumer<OpcuaAPU, Throwable> error = (message, e) -> {
LOGGER.error("Error while waiting for subscription response");
e.printStackTrace();
LOGGER.error("Error while waiting for subscription response", e);
plcSubscriber.onDisconnect(context);
};

outstandingRequests.add(requestHandle);
channel.submit(context, timeout, error, consumer, buffer);

} catch (SerializationException e) {
LOGGER.warn("Unable to serialize subscription request");
e.printStackTrace();
LOGGER.warn("Unable to serialize subscription request", e);
}
}
/* Put the subscriber loop to sleep for the rest of the cycle. */
Expand All @@ -338,8 +331,7 @@ public void startSubscriber() {
//sleep(this.revisedCycleTime * 10);
complete = true;
} catch (Exception e) {
LOGGER.error("Failed :(");
e.printStackTrace();
LOGGER.error("Failed to start subscription", e);
}
return null;
});
Expand Down Expand Up @@ -400,27 +392,23 @@ public void stopSubscriber() {
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();
LOGGER.error("Unable to parse the returned Delete Subscriptions Response", e);
}
};

Consumer<TimeoutException> timeout = e -> {
LOGGER.error("Timeout while waiting for delete subscription response");
e.printStackTrace();
LOGGER.error("Timeout while waiting for delete subscription response", e);
plcSubscriber.onDisconnect(context);
};

BiConsumer<OpcuaAPU, Throwable> error = (message, e) -> {
LOGGER.error("Error while waiting for delete subscription response");
e.printStackTrace();
LOGGER.error("Error while waiting for delete subscription response", e);
plcSubscriber.onDisconnect(context);
};

channel.submit(context, timeout, error, consumer, buffer);
} catch (SerializationException e) {
LOGGER.warn("Unable to serialize subscription request");
e.printStackTrace();
LOGGER.warn("Unable to serialize subscription request", e);
}

sleep(500);
Expand Down

0 comments on commit 066c395

Please sign in to comment.