2 * ============LICENSE_START=======================================================
3 * Copyright (c) 2021-2022 Bell Canada.
4 * ================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.onap.cps.notification;
23 import java.time.OffsetDateTime;
24 import java.util.Arrays;
25 import java.util.Collections;
26 import java.util.List;
27 import java.util.concurrent.CompletableFuture;
28 import java.util.concurrent.Future;
29 import java.util.regex.Pattern;
30 import java.util.stream.Collectors;
31 import lombok.extern.slf4j.Slf4j;
32 import org.springframework.scheduling.annotation.Async;
33 import org.springframework.stereotype.Service;
37 public class NotificationService {
39 private static final String ROOT_NODE_XPATH = "/";
41 private NotificationProperties notificationProperties;
42 private NotificationPublisher notificationPublisher;
43 private CpsDataUpdatedEventFactory cpsDataUpdatedEventFactory;
44 private NotificationErrorHandler notificationErrorHandler;
45 private List<Pattern> dataspacePatterns;
48 * Create an instance of Notification Subscriber.
50 * @param notificationProperties properties for notification
51 * @param notificationPublisher notification Publisher
52 * @param cpsDataUpdatedEventFactory to create CPSDataUpdatedEvent
53 * @param notificationErrorHandler error handler
55 public NotificationService(
56 final NotificationProperties notificationProperties,
57 final NotificationPublisher notificationPublisher,
58 final CpsDataUpdatedEventFactory cpsDataUpdatedEventFactory,
59 final NotificationErrorHandler notificationErrorHandler) {
60 log.info("Notification Properties {}", notificationProperties);
61 this.notificationProperties = notificationProperties;
62 this.notificationPublisher = notificationPublisher;
63 this.cpsDataUpdatedEventFactory = cpsDataUpdatedEventFactory;
64 this.notificationErrorHandler = notificationErrorHandler;
65 this.dataspacePatterns = getDataspaceFilterPatterns(notificationProperties);
68 private List<Pattern> getDataspaceFilterPatterns(final NotificationProperties notificationProperties) {
69 if (notificationProperties.isEnabled()) {
70 return Arrays.stream(notificationProperties.getFilters()
71 .getOrDefault("enabled-dataspaces", "")
73 .map(filterPattern -> Pattern.compile(filterPattern, Pattern.CASE_INSENSITIVE))
74 .collect(Collectors.toList());
76 return Collections.emptyList();
81 * Process Data Updated Event and publishes the notification.
83 * @param dataspaceName dataspace name
84 * @param anchorName anchor name
85 * @param observedTimestamp observedTimestamp
86 * @param xpath xpath of changed data node
87 * @param operation operation
90 @Async("notificationExecutor")
91 public Future<Void> processDataUpdatedEvent(final String dataspaceName, final String anchorName,
92 final OffsetDateTime observedTimestamp,
93 final String xpath, final Operation operation) {
94 log.debug("process data updated event for dataspace '{}' & anchor '{}'", dataspaceName, anchorName);
96 if (shouldSendNotification(dataspaceName)) {
97 final var cpsDataUpdatedEvent =
98 cpsDataUpdatedEventFactory.createCpsDataUpdatedEvent(dataspaceName, anchorName,
99 observedTimestamp, getRootNodeOperation(xpath, operation));
100 log.debug("data updated event to be published {}", cpsDataUpdatedEvent);
101 notificationPublisher.sendNotification(cpsDataUpdatedEvent);
103 } catch (final Exception exception) {
104 /* All the exceptions are handled to not to propagate it to caller.
105 CPS operation should not fail if sending event fails for any reason.
107 notificationErrorHandler.onException("Failed to process cps-data-updated-event.",
108 exception, dataspaceName, anchorName);
110 return CompletableFuture.completedFuture(null);
114 Add more complex rules based on dataspace and anchor later
116 private boolean shouldSendNotification(final String dataspaceName) {
118 return notificationProperties.isEnabled()
119 && dataspacePatterns.stream()
120 .anyMatch(pattern -> pattern.matcher(dataspaceName).find());
123 private Operation getRootNodeOperation(final String xpath, final Operation operation) {
124 return ROOT_NODE_XPATH.equals(xpath) ? operation : Operation.UPDATE;