summaryrefslogtreecommitdiffstats
path: root/cps-ncmp-service/src/main
diff options
context:
space:
mode:
authorhalil.cakal <halil.cakal@est.tech>2023-07-03 15:24:02 +0100
committerhalil.cakal <halil.cakal@est.tech>2023-07-07 17:38:00 +0100
commitadfb0693ec18025abcd8b9036bafe20a25f2e496 (patch)
treebf8e1b1f2918ce5298774a379121b34e37b65742 /cps-ncmp-service/src/main
parentfb7005ffda3f7d2194061192c5c8e4574a72027b (diff)
Subscription Creation: NCMP to DMI CloudEvent transformation
- Add mapper to convert client event into ncmp event - Add sample json object of ncmp version - Change subscription event consumer to consume CloudEvents - Change subscription event forwarder to publish CloudEvents - Change test producer config to support CloudEvents - Change sample subscription event json to comply with new schema - Add more test for missing branches - Change packages of the mappers into relevant directory Issue-ID: CPS-1737 Change-Id: I8c9e6e7bf713a8fb530a0586dfb2bce796a462f5 Signed-off-by: halil.cakal <halil.cakal@est.tech>
Diffstat (limited to 'cps-ncmp-service/src/main')
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java34
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java28
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java72
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java15
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java79
5 files changed, 174 insertions, 54 deletions
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java
new file mode 100644
index 0000000000..59b1d09c7b
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java
@@ -0,0 +1,34 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * Licensed 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.events.avcsubscription;
+
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
+
+@Mapper(componentModel = "spring")
+public interface ClientSubscriptionEventMapper {
+
+ @Mapping(target = "data.predicates.targets", ignore = true)
+ org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent toNcmpSubscriptionEvent(
+ SubscriptionEvent subscriptionEvent);
+
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java
index 88b41d0075..f511965c77 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java
@@ -20,13 +20,14 @@
package org.onap.cps.ncmp.api.impl.events.avcsubscription;
+import io.cloudevents.CloudEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence;
+import org.onap.cps.ncmp.api.impl.utils.SubscriptionEventCloudMapper;
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
-import org.onap.cps.ncmp.event.model.InnerSubscriptionEvent;
-import org.onap.cps.ncmp.event.model.SubscriptionEvent;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
import org.onap.cps.spi.exceptions.OperationNotYetSupportedException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
@@ -54,28 +55,25 @@ public class SubscriptionEventConsumer {
* @param subscriptionEventConsumerRecord the event to be consumed
*/
@KafkaListener(topics = "${app.ncmp.avc.subscription-topic}",
- properties = {"spring.json.value.default.type=org.onap.cps.ncmp.event.model.SubscriptionEvent"})
- public void consumeSubscriptionEvent(
- final ConsumerRecord<String, SubscriptionEvent> subscriptionEventConsumerRecord) {
- final SubscriptionEvent subscriptionEvent = subscriptionEventConsumerRecord.value();
- final InnerSubscriptionEvent event = subscriptionEvent.getEvent();
- final String eventDatastore = event.getPredicates().getDatastore();
+ containerFactory = "cloudEventConcurrentKafkaListenerContainerFactory")
+ public void consumeSubscriptionEvent(final ConsumerRecord<String, CloudEvent> subscriptionEventConsumerRecord) {
+ final CloudEvent cloudEvent = subscriptionEventConsumerRecord.value();
+ final SubscriptionEvent subscriptionEvent = SubscriptionEventCloudMapper.toSubscriptionEvent(cloudEvent);
+ final String eventDatastore = subscriptionEvent.getData().getPredicates().getDatastore();
if (!(eventDatastore.equals("passthrough-running") || eventDatastore.equals("passthrough-operational"))) {
throw new OperationNotYetSupportedException(
"passthrough datastores are currently only supported for event subscriptions");
}
- if ("CM".equals(event.getDataType().getDataCategory())) {
- log.debug("Consuming event {} ...", subscriptionEvent);
+ if ("CM".equals(subscriptionEvent.getData().getDataType().getDataCategory())) {
if (subscriptionModelLoaderEnabled) {
persistSubscriptionEvent(subscriptionEvent);
}
- if ("CREATE".equals(subscriptionEvent.getEventType().value())) {
+ if ("CREATE".equals(cloudEvent.getType())) {
log.info("Subscription for ClientID {} with name {} ...",
- event.getSubscription().getClientID(),
- event.getSubscription().getName());
+ subscriptionEvent.getData().getSubscription().getClientID(),
+ subscriptionEvent.getData().getSubscription().getName());
if (notificationFeatureEnabled) {
- subscriptionEventForwarder.forwardCreateSubscriptionEvent(subscriptionEvent,
- subscriptionEventConsumerRecord.headers());
+ subscriptionEventForwarder.forwardCreateSubscriptionEvent(subscriptionEvent);
}
}
} else {
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
index 1d87a057a7..1fe963a279 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
@@ -21,13 +21,12 @@
package org.onap.cps.ncmp.api.impl.events.avcsubscription;
import com.hazelcast.map.IMap;
+import io.cloudevents.CloudEvent;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -35,16 +34,17 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.common.header.Headers;
import org.onap.cps.ncmp.api.impl.config.embeddedcache.ForwardedSubscriptionEventCacheConfig;
import org.onap.cps.ncmp.api.impl.events.EventsPublisher;
import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence;
import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus;
import org.onap.cps.ncmp.api.impl.utils.DmiServiceNameOrganizer;
+import org.onap.cps.ncmp.api.impl.utils.SubscriptionEventCloudMapper;
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
import org.onap.cps.ncmp.api.inventory.InventoryPersistence;
-import org.onap.cps.ncmp.event.model.SubscriptionEvent;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.CmHandle;
import org.onap.cps.spi.exceptions.OperationNotYetSupportedException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@@ -56,10 +56,11 @@ import org.springframework.stereotype.Component;
public class SubscriptionEventForwarder {
private final InventoryPersistence inventoryPersistence;
- private final EventsPublisher<SubscriptionEvent> eventsPublisher;
+ private final EventsPublisher<CloudEvent> eventsPublisher;
private final IMap<String, Set<String>> forwardedSubscriptionEventCache;
private final SubscriptionEventResponseOutcome subscriptionEventResponseOutcome;
private final SubscriptionEventMapper subscriptionEventMapper;
+ private final ClientSubscriptionEventMapper clientSubscriptionEventMapper;
private final SubscriptionPersistence subscriptionPersistence;
private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
@Value("${app.ncmp.avc.subscription-forward-topic-prefix}")
@@ -73,28 +74,22 @@ public class SubscriptionEventForwarder {
*
* @param subscriptionEvent the event to be forwarded
*/
- public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent,
- final Headers eventHeaders) {
- final List<Object> cmHandleTargets = subscriptionEvent.getEvent().getPredicates().getTargets();
+ public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent) {
+ final List<String> cmHandleTargets = subscriptionEvent.getData().getPredicates().getTargets();
if (cmHandleTargets == null || cmHandleTargets.isEmpty()
- || cmHandleTargets.stream().anyMatch(id -> ((String) id).contains("*"))) {
+ || cmHandleTargets.stream().anyMatch(id -> (id).contains("*"))) {
throw new OperationNotYetSupportedException(
"CMHandle targets are required. \"Wildcard\" operations are not yet supported");
}
- final List<String> cmHandleTargetsAsStrings = cmHandleTargets.stream().map(
- Objects::toString).collect(Collectors.toList());
final Collection<YangModelCmHandle> yangModelCmHandles =
- inventoryPersistence.getYangModelCmHandles(cmHandleTargetsAsStrings);
-
+ inventoryPersistence.getYangModelCmHandles(cmHandleTargets);
final Map<String, Map<String, Map<String, String>>> dmiPropertiesPerCmHandleIdPerServiceName
= DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
-
- findDmisAndRespond(subscriptionEvent, eventHeaders, cmHandleTargetsAsStrings,
- dmiPropertiesPerCmHandleIdPerServiceName);
+ findDmisAndRespond(subscriptionEvent, cmHandleTargets, dmiPropertiesPerCmHandleIdPerServiceName);
}
- private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent, final Headers eventHeaders,
- final List<String> cmHandleTargetsAsStrings,
+ private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent,
+ final List<String> cmHandleTargetsAsStrings,
final Map<String, Map<String, Map<String, String>>>
dmiPropertiesPerCmHandleIdPerServiceName) {
final List<String> cmHandlesThatExistsInDb = dmiPropertiesPerCmHandleIdPerServiceName.entrySet().stream()
@@ -109,18 +104,20 @@ public class SubscriptionEventForwarder {
updatesCmHandlesToRejectedAndPersistSubscriptionEvent(subscriptionEvent, targetCmHandlesDoesNotExistInDb);
}
if (dmisToRespond.isEmpty()) {
- final String clientID = subscriptionEvent.getEvent().getSubscription().getClientID();
- final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName();
+ final String clientID = subscriptionEvent.getData().getSubscription().getClientID();
+ final String subscriptionName = subscriptionEvent.getData().getSubscription().getName();
subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName);
} else {
startResponseTimeout(subscriptionEvent, dmisToRespond);
- forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent, eventHeaders);
+ final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent ncmpSubscriptionEvent =
+ clientSubscriptionEventMapper.toNcmpSubscriptionEvent(subscriptionEvent);
+ forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, ncmpSubscriptionEvent);
}
}
private void startResponseTimeout(final SubscriptionEvent subscriptionEvent, final Set<String> dmisToRespond) {
- final String subscriptionClientId = subscriptionEvent.getEvent().getSubscription().getClientID();
- final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName();
+ final String subscriptionClientId = subscriptionEvent.getData().getSubscription().getClientID();
+ final String subscriptionName = subscriptionEvent.getData().getSubscription().getName();
final String subscriptionEventId = subscriptionClientId + subscriptionName;
forwardedSubscriptionEventCache.put(subscriptionEventId, dmisToRespond,
@@ -137,20 +134,33 @@ public class SubscriptionEventForwarder {
}
private void forwardEventToDmis(final Map<String, Map<String, Map<String, String>>> dmiNameCmHandleMap,
- final SubscriptionEvent subscriptionEvent,
- final Headers eventHeaders) {
+ final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent
+ ncmpSubscriptionEvent) {
dmiNameCmHandleMap.forEach((dmiName, cmHandlePropertiesMap) -> {
- subscriptionEvent.getEvent().getPredicates().setTargets(Collections.singletonList(cmHandlePropertiesMap));
- final String eventKey = createEventKey(subscriptionEvent, dmiName);
+ final List<CmHandle> cmHandleTargets = cmHandlePropertiesMap.entrySet().stream().map(
+ cmHandleAndProperties -> {
+ final CmHandle cmHandle = new CmHandle();
+ cmHandle.setId(cmHandleAndProperties.getKey());
+ cmHandle.setAdditionalProperties(cmHandleAndProperties.getValue());
+ return cmHandle;
+ }).collect(Collectors.toList());
+
+ ncmpSubscriptionEvent.getData().getPredicates().setTargets(cmHandleTargets);
+ final String eventKey = createEventKey(ncmpSubscriptionEvent, dmiName);
final String dmiAvcSubscriptionTopic = dmiAvcSubscriptionTopicPrefix + dmiName;
- eventsPublisher.publishEvent(dmiAvcSubscriptionTopic, eventKey, eventHeaders, subscriptionEvent);
+
+ final CloudEvent ncmpSubscriptionCloudEvent =
+ SubscriptionEventCloudMapper.toCloudEvent(ncmpSubscriptionEvent, eventKey);
+ eventsPublisher.publishCloudEvent(dmiAvcSubscriptionTopic, eventKey, ncmpSubscriptionCloudEvent);
});
}
- private String createEventKey(final SubscriptionEvent subscriptionEvent, final String dmiName) {
- return subscriptionEvent.getEvent().getSubscription().getClientID()
+ private String createEventKey(
+ final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent subscriptionEvent,
+ final String dmiName) {
+ return subscriptionEvent.getData().getSubscription().getClientID()
+ "-"
- + subscriptionEvent.getEvent().getSubscription().getName()
+ + subscriptionEvent.getData().getSubscription().getName()
+ "-"
+ dmiName;
}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java
index dcbdcf339d..bf9ceb1c3d 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java
@@ -27,17 +27,16 @@ import org.mapstruct.Mapping;
import org.mapstruct.Named;
import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus;
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
-import org.onap.cps.ncmp.event.model.SubscriptionEvent;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
@Mapper(componentModel = "spring")
public interface SubscriptionEventMapper {
- @Mapping(source = "event.subscription.clientID", target = "clientId")
- @Mapping(source = "event.subscription.name", target = "subscriptionName")
- @Mapping(source = "event.subscription.isTagged", target = "tagged")
- @Mapping(source = "event.predicates.targets", target = "predicates.targetCmHandles",
+ @Mapping(source = "data.subscription.clientID", target = "clientId")
+ @Mapping(source = "data.subscription.name", target = "subscriptionName")
+ @Mapping(source = "data.predicates.targets", target = "predicates.targetCmHandles",
qualifiedByName = "mapTargetsToCmHandleTargets")
- @Mapping(source = "event.predicates.datastore", target = "predicates.datastore")
+ @Mapping(source = "data.predicates.datastore", target = "predicates.datastore")
YangModelSubscriptionEvent toYangModelSubscriptionEvent(SubscriptionEvent subscriptionEvent);
/**
@@ -47,8 +46,8 @@ public interface SubscriptionEventMapper {
* @return TargetCmHandle list
*/
@Named("mapTargetsToCmHandleTargets")
- default List<YangModelSubscriptionEvent.TargetCmHandle> mapTargetsToCmHandleTargets(List<Object> targets) {
- return targets.stream().map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target.toString(),
+ default List<YangModelSubscriptionEvent.TargetCmHandle> mapTargetsToCmHandleTargets(List<String> targets) {
+ return targets.stream().map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target,
SubscriptionStatus.PENDING))
.collect(Collectors.toList());
}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java
new file mode 100644
index 0000000000..a7de479046
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java
@@ -0,0 +1,79 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * Licensed 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.utils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.CloudEventUtils;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.cloudevents.core.data.PojoCloudEventData;
+import io.cloudevents.jackson.PojoCloudEventDataMapper;
+import java.net.URI;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Slf4j
+public class SubscriptionEventCloudMapper {
+
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ /**
+ * Maps CloudEvent object to SubscriptionEvent.
+ *
+ * @param cloudEvent object.
+ * @return SubscriptionEvent deserialized.
+ */
+ public static SubscriptionEvent toSubscriptionEvent(final CloudEvent cloudEvent) {
+ final PojoCloudEventData<SubscriptionEvent> deserializedCloudEvent = CloudEventUtils
+ .mapData(cloudEvent, PojoCloudEventDataMapper.from(objectMapper, SubscriptionEvent.class));
+ if (deserializedCloudEvent == null) {
+ log.debug("No data found in the consumed event");
+ return null;
+ } else {
+ final SubscriptionEvent subscriptionEvent = deserializedCloudEvent.getValue();
+ log.debug("Consuming event {}", subscriptionEvent);
+ return subscriptionEvent;
+ }
+ }
+
+ /**
+ * Maps SubscriptionEvent to a CloudEvent.
+ *
+ * @param ncmpSubscriptionEvent object.
+ * @param eventKey as String.
+ * @return CloudEvent builded.
+ */
+ public static CloudEvent toCloudEvent(
+ final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent ncmpSubscriptionEvent,
+ final String eventKey) {
+ try {
+ return CloudEventBuilder.v1()
+ .withData(objectMapper.writeValueAsBytes(ncmpSubscriptionEvent))
+ .withId(eventKey).withType("CREATE").withSource(
+ URI.create(ncmpSubscriptionEvent.getData().getSubscription().getClientID())).build();
+ } catch (final Exception ex) {
+ throw new RuntimeException("The Cloud Event could not be constructed.", ex);
+ }
+ }
+}