+ private Collection<DataNode> buildDataNodes(final String dataspaceName,
+ final String anchorName,
+ final String parentNodeXpath,
+ final String jsonData) {
+
+ final var anchor = cpsAdminService.getAnchor(dataspaceName, anchorName);
+ final var schemaContext = getSchemaContext(dataspaceName, anchor.getSchemaSetName());
+
+ final NormalizedNode<?, ?> normalizedNode = YangUtils.parseJsonData(jsonData, schemaContext, parentNodeXpath);
+ final Collection<DataNode> dataNodes = new DataNodeBuilder()
+ .withParentNodeXpath(parentNodeXpath)
+ .withNormalizedNodeTree(normalizedNode)
+ .buildCollection();
+ if (dataNodes.isEmpty()) {
+ throw new DataValidationException("Invalid data.", "No data nodes provided");
+ }
+ return dataNodes;
+
+ }
+
+ private void processDataUpdatedEventAsync(final String dataspaceName, final String anchorName,
+ final OffsetDateTime observedTimestamp, final String xpath,
+ final Operation operation) {
+ final var anchor = cpsAdminService.getAnchor(dataspaceName, anchorName);
+ this.processDataUpdatedEventAsync(anchor, xpath, operation, observedTimestamp);
+ }
+
+ private void processDataUpdatedEventAsync(final Anchor anchor, final String xpath, final Operation operation,
+ final OffsetDateTime observedTimestamp) {
+ try {
+ notificationService.processDataUpdatedEvent(anchor, observedTimestamp, xpath, operation);
+ } catch (final Exception exception) {
+ //If async message can't be queued for notification service, the initial request should not failed.
+ log.error("Failed to send message to notification service", exception);
+ }
+ }
+