<groupId>ch.qos.logback</groupId>
<artifactId>logback-access</artifactId>
</dependency>
- <dependency>
- <groupId>org.apache.activemq</groupId>
- <artifactId>activemq-broker</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.activemq</groupId>
- <artifactId>activemq-openwire-legacy</artifactId>
- </dependency>
<dependency>
<groupId>org.freemarker</groupId>
<artifactId>freemarker</artifactId>
</exclusion>
</exclusions>
</dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-jms</artifactId>
- </dependency>
- <dependency>
- <groupId>javax.jms</groupId>
- <artifactId>javax.jms-api</artifactId>
- <version>2.0.1</version>
- </dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
* ============LICENSE_END=========================================================
*/
-package org.onap.aai.web;
+package org.onap.aai.config;
import java.util.Map;
import java.util.HashMap;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
+import org.onap.aai.domain.deltaEvent.DeltaEvent;
import org.onap.aai.domain.notificationEvent.NotificationEvent;
+import org.onap.aai.kafka.DeltaProducer;
+import org.onap.aai.kafka.DeltaProducerService;
import org.onap.aai.kafka.NotificationProducer;
import org.onap.aai.kafka.NotificationProducerService;
import org.slf4j.Logger;
import org.springframework.kafka.support.serializer.JsonSerializer;
@Configuration
-public class KafkaNotificationEventConfig {
+public class KafkaConfig {
- private static final Logger logger = LoggerFactory.getLogger(KafkaNotificationEventConfig.class);
-
- @Value("${jms.bind.address}")
- private String bindAddress;
+ private static final Logger logger = LoggerFactory.getLogger(KafkaConfig.class);
@Value("${spring.kafka.producer.bootstrap-servers}")
private String bootstrapServers;
return props;
}
+ @Bean
+ public KafkaTemplate<String, NotificationEvent> kafkaNotificationEventTemplate(ProducerFactory<String, NotificationEvent> producerFactory) throws Exception {
+ return new KafkaTemplate<>(producerFactory);
+ }
+
+ @Bean
+ public NotificationProducer notificationProducer(KafkaTemplate<String,NotificationEvent> kafkaTemplate) {
+ return new NotificationProducerService(kafkaTemplate);
+ }
+
@Bean
public ProducerFactory<String, NotificationEvent> notificationEventProducerFactory() throws Exception {
Map<String, Object> props = buildKafkaProperties();
-
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
- public KafkaTemplate<String, NotificationEvent> kafkaNotificationEventTemplate(ProducerFactory<String, NotificationEvent> producerFactory) throws Exception {
- try {
+ public ProducerFactory<String, DeltaEvent> deltaEventProducerFactory() throws Exception {
+ Map<String, Object> props = buildKafkaProperties();
+ return new DefaultKafkaProducerFactory<>(props);
+ }
+ @Bean
+ public KafkaTemplate<String, DeltaEvent> kafkaDeltaEventTemplate(ProducerFactory<String, DeltaEvent> producerFactory) throws Exception {
return new KafkaTemplate<>(producerFactory);
- } catch (Exception e) {
- String smth = "";
- return null;
- }
}
@Bean
- public NotificationProducer notificationProducer(KafkaTemplate<String,NotificationEvent> kafkaTemplate) {
- return new NotificationProducerService(kafkaTemplate);
+ public DeltaProducer deltaProducer(KafkaTemplate<String,DeltaEvent> kafkaTemplate) {
+ return new DeltaProducerService(kafkaTemplate);
}
}
--- /dev/null
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2024 Deutsche Telekom. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.aai.domain.deltaEvent;
+
+import java.util.Collection;
+
+import org.onap.aai.domain.notificationEvent.NotificationEvent.EventHeader;
+import org.onap.aai.util.delta.ObjectDelta;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import lombok.Data;
+
+@Data
+public class DeltaEvent {
+ @JsonProperty("cambria.partition")
+ protected String cambriaPartition;
+ @JsonProperty("event-header")
+ protected EventHeader eventHeader;
+ Collection<ObjectDelta> entities;
+}
@XmlType(
name = "",
propOrder = {"id", "timestamp", "sourceName", "domain", "sequenceNumber", "severity", "eventType",
- "version", "action", "entityType", "topEntityType", "entityLink"})
+ "version", "action", "entityType", "topEntityType", "entityLink","entityUuid"})
public static class EventHeader {
@XmlElement(required = true)
@XmlElement(name = "entity-link", required = true)
@JsonProperty("entity-link")
protected String entityLink;
+ @JsonProperty("entity-uuid")
+ protected String entityUuid;
}
}
+++ /dev/null
-/**
- * ============LICENSE_START=======================================================
- * org.onap.aai
- * ================================================================================
- * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Modifications Copyright © 2018 IBM.
- * ================================================================================
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.aai.kafka;
-
-import java.util.Map;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.TextMessage;
-
-import org.json.JSONException;
-import org.json.JSONObject;
-import org.onap.aai.aailog.logs.AaiDmaapMetricLog;
-import org.onap.aai.exceptions.AAIException;
-import org.onap.aai.logging.AaiElsErrorCode;
-import org.onap.aai.logging.ErrorLogHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
-import org.springframework.kafka.core.KafkaTemplate;
-
-
-public class AAIKafkaEventJMSConsumer implements MessageListener {
-
- private static final String EVENT_TOPIC = "event-topic";
-
- private static final Logger LOGGER = LoggerFactory.getLogger(AAIKafkaEventJMSConsumer.class);
-
- private Map<String, String> mdcCopy;
- private final KafkaTemplate<String, String> kafkaTemplate;
-
- public AAIKafkaEventJMSConsumer(KafkaTemplate<String, String> kafkaTemplate) {
- super();
- mdcCopy = MDC.getCopyOfContextMap();
- this.kafkaTemplate = kafkaTemplate;
- }
-
- @Override
- public void onMessage(Message message) {
-
- if (kafkaTemplate == null) {
- return;
- }
-
- String jmsMessageText = "";
- String aaiEvent = "";
- JSONObject aaiEventHeader;
- JSONObject aaiEventPayload;
- String transactionId = "";
- String serviceName = "";
- String topicName = "";
- String aaiElsErrorCode = AaiElsErrorCode.SUCCESS;
- String errorDescription = "";
-
- if (mdcCopy != null) {
- MDC.setContextMap(mdcCopy);
- }
-
- if (message instanceof TextMessage) {
- AaiDmaapMetricLog metricLog = new AaiDmaapMetricLog();
- try {
- jmsMessageText = ((TextMessage) message).getText();
- JSONObject jsonObject = new JSONObject(jmsMessageText);
- if (jsonObject.has("aaiEventPayload")) {
- aaiEventPayload = jsonObject.getJSONObject("aaiEventPayload");
- aaiEvent = aaiEventPayload.toString();
- } else {
- return;
- }
- if (jsonObject.getString(EVENT_TOPIC) != null) {
- topicName = jsonObject.getString(EVENT_TOPIC);
- }
- if (aaiEventPayload.has("event-header")) {
- try {
- aaiEventHeader = aaiEventPayload.getJSONObject("event-header");
- if (aaiEventHeader.has("id")) {
- transactionId = aaiEventHeader.get("id").toString();
- }
- if (aaiEventHeader.has("entity-link")) {
- serviceName = aaiEventHeader.get("entity-link").toString();
- }
- } catch (JSONException jexc) {
- // ignore, this is just used for logging
- }
- }
- metricLog.pre(topicName, aaiEvent, transactionId, serviceName);
-
- if ("AAI-EVENT".equals(topicName)) {
-
- kafkaTemplate.send(topicName, aaiEvent);
-
- } else {
- LOGGER.error(String.format("%s|Event Topic invalid.", topicName));
- }
- } catch (JMSException | JSONException e) {
- aaiElsErrorCode = AaiElsErrorCode.DATA_ERROR;
- errorDescription = e.getMessage();
- ErrorLogHelper.logException(new AAIException("AAI_7350"));
- } catch (Exception e) {
- e.printStackTrace();
- // LOGGER.error();
- LOGGER.error(e.getMessage());
- aaiElsErrorCode = AaiElsErrorCode.AVAILABILITY_TIMEOUT_ERROR;
- errorDescription = e.getMessage();
- String errorMessage = String.format("Error processing message: %s, message payload: %s", e.getMessage(), jmsMessageText);
- ErrorLogHelper.logException(new AAIException("AAI_7304", errorMessage));
- } finally {
- metricLog.post(aaiElsErrorCode, errorDescription);
- }
- }
- }
-}
+++ /dev/null
-/**
- * ============LICENSE_START=======================================================
- * org.onap.aai
- * ================================================================================
- * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Modifications Copyright © 2018 IBM.
- * ================================================================================
- * 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.
- * ============LICENSE_END=========================================================
- */
-
- package org.onap.aai.kafka;
-
- import org.json.JSONObject;
- import org.onap.aai.config.SpringContextAware;
- import org.onap.aai.domain.notificationEvent.NotificationEvent;
- import org.onap.aai.util.AAIConfig;
- import org.springframework.jms.core.JmsTemplate;
- import org.springframework.stereotype.Service;
-
- import com.fasterxml.jackson.core.JsonProcessingException;
- import com.fasterxml.jackson.databind.ObjectMapper;
- import com.fasterxml.jackson.databind.json.JsonMapper;
-
- import lombok.RequiredArgsConstructor;
- import lombok.extern.slf4j.Slf4j;
-
- @Slf4j
- @Service
- @RequiredArgsConstructor
- public class AAIKafkaEventJMSProducer implements MessageProducer {
-
- private boolean eventsEnabled = "true".equals(AAIConfig.get("aai.jms.enable", "true"));
- private JmsTemplate jmsTemplate;
- private static final ObjectMapper mapper = new JsonMapper();
-
- public AAIKafkaEventJMSProducer(JmsTemplate jmsTemplate) {
- this.jmsTemplate = jmsTemplate;
- }
-
- public void sendMessageToDefaultDestination(String msg) {
- if (eventsEnabled) {
- if(jmsTemplate == null) {
- this.jmsTemplate = SpringContextAware.getBean(JmsTemplate.class);
- }
- jmsTemplate.convertAndSend(msg);
- }
- }
-
- public void sendMessageToDefaultDestination(JSONObject finalJson) {
- sendMessageToDefaultDestination(finalJson.toString());
- }
- }
--- /dev/null
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2024 Deutsche Telekom. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.aai.kafka;
+
+import org.onap.aai.domain.deltaEvent.DeltaEvent;
+import org.springframework.stereotype.Service;
+
+@Service
+public interface DeltaProducer {
+ public void sendNotification(DeltaEvent notificationEvent);
+}
--- /dev/null
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2024 Deutsche Telekom. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.aai.kafka;
+
+import org.onap.aai.domain.deltaEvent.DeltaEvent;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+
+@Service
+@RequiredArgsConstructor
+public class DeltaProducerService implements DeltaProducer {
+
+ private final KafkaTemplate<String,DeltaEvent> kafkaTemplate;
+ @Value("${aai.notifications.enabled:true}")
+ boolean notificationsEnabled;
+
+ @Override
+ public void sendNotification(DeltaEvent deltaEvent) {
+ if(notificationsEnabled) {
+ kafkaTemplate.send("DELTA", deltaEvent);
+ }
+ }
+}
+++ /dev/null
-/**
- * ============LICENSE_START=======================================================
- * org.onap.aai
- * ================================================================================
- * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.aai.rest.notification;
-
-import org.onap.aai.exceptions.AAIException;
-import org.onap.aai.introspection.Introspector;
-import org.onap.aai.introspection.Loader;
-import org.onap.aai.setup.SchemaVersion;
-import org.onap.aai.util.StoreNotificationEvent;
-
-/**
- * The Class NotificationEvent.
- */
-public class NotificationEvent {
-
- private final Loader loader;
-
- private final Introspector eventHeader;
-
- private final Introspector obj;
- private final String transactionId;
- private final String sourceOfTruth;
-
- /**
- * Instantiates a new notification event.
- *
- * @param eventHeader the event header
- * @param obj the obj
- */
- public NotificationEvent(Loader loader, Introspector eventHeader, Introspector obj, String transactionId,
- String sourceOfTruth) {
- this.loader = loader;
- this.eventHeader = eventHeader;
- this.obj = obj;
- this.transactionId = transactionId;
- this.sourceOfTruth = sourceOfTruth;
- }
-
- /**
- * Trigger.
- *
- * @throws AAIException the AAI exception
- */
- public void trigger() throws AAIException {
-
- StoreNotificationEvent sne = new StoreNotificationEvent(transactionId, sourceOfTruth);
-
- sne.storeEventAndSendToJms(loader, eventHeader, obj);
-
- }
-
- public String getNotificationEvent() throws AAIException {
- return new StoreNotificationEvent(transactionId, sourceOfTruth).storeEventOnly(loader, eventHeader, obj);
- }
-
- /**
- * Gets the notification version.
- *
- * @return the notification version
- */
- public SchemaVersion getNotificationVersion() {
- return loader.getVersion();
- }
-
- /**
- * Gets the event header.
- *
- * @return the event header
- */
- public Introspector getEventHeader() {
- return eventHeader;
- }
-
- /**
- * Gets the obj.
- *
- * @return the obj
- */
- public Introspector getObj() {
- return obj;
- }
-
-}
import javax.ws.rs.core.Response.Status;
+import org.onap.aai.domain.notificationEvent.NotificationEvent;
import org.onap.aai.domain.notificationEvent.NotificationEvent.EventHeader;
import org.onap.aai.exceptions.AAIException;
import org.onap.aai.introspection.Introspector;
private final String domain = AAIConfig.get("aai.notificationEvent.default.domain", "UNK");
private final String sequenceNumber = AAIConfig.get("aai.notificationEvent.default.sequenceNumber", "UNK");
private final String severity = AAIConfig.get("aai.notificationEvent.default.severity", "UNK");
- private final Map<String, org.onap.aai.domain.notificationEvent.NotificationEvent> events;
+ private final Map<String, NotificationEvent> events;
private final Loader currentVersionLoader;
private final SchemaVersion notificationVersion;
Introspector entity = entityConverter.convert(obj);
- final org.onap.aai.domain.notificationEvent.NotificationEvent event =
- new org.onap.aai.domain.notificationEvent.NotificationEvent();
+ final NotificationEvent event = new NotificationEvent();
event.setEventHeader(eventHeader);
event.setCambriaPartition(AAIConstants.UEB_PUB_PARTITION_AAI);
event.setEntity(entity);
return action;
}
- public List<org.onap.aai.domain.notificationEvent.NotificationEvent> getEvents() {
+ public List<NotificationEvent> getEvents() {
return new ArrayList<>(this.events.values());
}
+++ /dev/null
-/**
- * ============LICENSE_START=======================================================
- * org.onap.aai
- * ================================================================================
- * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.aai.util;
-
-import java.io.StringWriter;
-import java.util.Iterator;
-import java.util.UUID;
-
-import javax.xml.bind.Marshaller;
-
-import org.eclipse.persistence.dynamic.DynamicEntity;
-import org.eclipse.persistence.jaxb.dynamic.DynamicJAXBContext;
-import org.json.JSONException;
-import org.json.JSONObject;
-import org.onap.aai.domain.notificationEvent.NotificationEvent;
-import org.onap.aai.exceptions.AAIException;
-import org.onap.aai.introspection.Introspector;
-import org.onap.aai.introspection.Loader;
-import org.onap.aai.introspection.exceptions.AAIUnknownObjectException;
-import org.onap.aai.kafka.AAIKafkaEventJMSProducer;
-import org.onap.aai.kafka.MessageProducer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationContext;
-import org.springframework.core.env.Environment;
-import org.springframework.jms.core.JmsTemplate;
-
-public class StoreNotificationEvent {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(StoreNotificationEvent.class);
-
- @Autowired JmsTemplate jmsTemplate;
-
- private final MessageProducer messageProducer;
- private String fromAppId = "";
- private String transId = "";
- private final String transactionId;
- private final String sourceOfTruth;
-
- private ApplicationContext context;
- private Environment env;
-
- /**
- * Instantiates a new store notification event.
- */
- public StoreNotificationEvent(String transactionId, String sourceOfTruth) {
- this.messageProducer = new AAIKafkaEventJMSProducer(jmsTemplate);
- this.transactionId = transactionId;
- this.sourceOfTruth = sourceOfTruth;
- }
-
- public StoreNotificationEvent(MessageProducer producer, String transactionId, String sourceOfTruth) {
- this.messageProducer = producer;
- this.transactionId = transactionId;
- this.sourceOfTruth = sourceOfTruth;
- }
-
- /**
- * Store event.
- *
- * @param eh
- * the eh
- * @param obj
- * the obj
- * @throws AAIException
- * the AAI exception
- */
- public String storeEventAndSendToJms(NotificationEvent.EventHeader eh, Object obj) throws AAIException {
-
- if (obj == null) {
- throw new AAIException("AAI_7350");
- }
-
- org.onap.aai.domain.notificationEvent.ObjectFactory factory =
- new org.onap.aai.domain.notificationEvent.ObjectFactory();
-
- org.onap.aai.domain.notificationEvent.NotificationEvent ne = factory.createNotificationEvent();
-
- if (eh.getId() == null) {
- eh.setId(genDate2() + "-" + UUID.randomUUID().toString());
- }
- if (eh.getTimestamp() == null) {
- eh.setTimestamp(genDate());
- }
-
- // there's no default, but i think we want to put this in hbase?
-
- if (eh.getEntityLink() == null) {
- eh.setEntityLink("UNK");
- }
-
- if (eh.getAction() == null) {
- eh.setAction("UNK");
- }
-
- if (eh.getEventType() == null) {
- eh.setEventType(AAIConfig.get("aai.notificationEvent.default.eventType", "UNK"));
- }
-
- if (eh.getDomain() == null) {
- eh.setDomain(AAIConfig.get("aai.notificationEvent.default.domain", "UNK"));
- }
-
- if (eh.getSourceName() == null) {
- eh.setSourceName(AAIConfig.get("aai.notificationEvent.default.sourceName", "UNK"));
- }
-
- if (eh.getSequenceNumber() == null) {
- eh.setSequenceNumber(AAIConfig.get("aai.notificationEvent.default.sequenceNumber", "UNK"));
- }
-
- if (eh.getSeverity() == null) {
- eh.setSeverity(AAIConfig.get("aai.notificationEvent.default.severity", "UNK"));
- }
-
- if (eh.getVersion() == null) {
- eh.setVersion(AAIConfig.get("aai.notificationEvent.default.version", "UNK"));
- }
-
- ne.setCambriaPartition(AAIConstants.UEB_PUB_PARTITION_AAI);
- ne.setEventHeader(eh);
- ne.setEntity(obj);
-
- try {
- PojoUtils pu = new PojoUtils();
- String entityJson = pu.getJsonFromObject(ne);
- sendToKafkaJmsQueue(entityJson);
- return entityJson;
- } catch (Exception e) {
- throw new AAIException("AAI_7350", e);
- }
- }
-
- /**
- * Store dynamic event.
- *
- * @param notificationJaxbContext
- * the notification jaxb context
- * @param notificationVersion
- * the notification version
- * @param eventHeader
- * the event header
- * @param obj
- * the obj
- * @throws AAIException
- * the AAI exception
- */
- public void storeDynamicEvent(DynamicJAXBContext notificationJaxbContext, String notificationVersion,
- DynamicEntity eventHeader, DynamicEntity obj) throws AAIException {
-
- if (obj == null) {
- throw new AAIException("AAI_7350");
- }
-
- DynamicEntity notificationEvent = notificationJaxbContext
- .getDynamicType("inventory.aai.onap.org." + notificationVersion + ".NotificationEvent")
- .newDynamicEntity();
-
- if (eventHeader.get("id") == null) {
- eventHeader.set("id", genDate2() + "-" + UUID.randomUUID().toString());
- }
-
- if (eventHeader.get("timestamp") == null) {
- eventHeader.set("timestamp", genDate());
- }
-
- if (eventHeader.get("entityLink") == null) {
- eventHeader.set("entityLink", "UNK");
- }
-
- if (eventHeader.get("action") == null) {
- eventHeader.set("action", "UNK");
- }
-
- if (eventHeader.get("eventType") == null) {
- eventHeader.set("eventType", AAIConfig.get("aai.notificationEvent.default.eventType", "UNK"));
- }
-
- if (eventHeader.get("domain") == null) {
- eventHeader.set("domain", AAIConfig.get("aai.notificationEvent.default.domain", "UNK"));
- }
-
- if (eventHeader.get("sourceName") == null) {
- eventHeader.set("sourceName", AAIConfig.get("aai.notificationEvent.default.sourceName", "UNK"));
- }
-
- if (eventHeader.get("sequenceNumber") == null) {
- eventHeader.set("sequenceNumber", AAIConfig.get("aai.notificationEvent.default.sequenceNumber", "UNK"));
- }
-
- if (eventHeader.get("severity") == null) {
- eventHeader.set("severity", AAIConfig.get("aai.notificationEvent.default.severity", "UNK"));
- }
-
- if (eventHeader.get("version") == null) {
- eventHeader.set("version", AAIConfig.get("aai.notificationEvent.default.version", "UNK"));
- }
-
- if (notificationEvent.get("cambriaPartition") == null) {
- notificationEvent.set("cambriaPartition", AAIConstants.UEB_PUB_PARTITION_AAI);
- }
-
- notificationEvent.set("eventHeader", eventHeader);
- notificationEvent.set("entity", obj);
-
- try {
- StringWriter result = new StringWriter();
-
- Marshaller marshaller = notificationJaxbContext.createMarshaller();
- marshaller.setProperty(org.eclipse.persistence.jaxb.MarshallerProperties.MEDIA_TYPE, "application/json");
- marshaller.setProperty(org.eclipse.persistence.jaxb.MarshallerProperties.JSON_INCLUDE_ROOT, false);
- marshaller.setProperty(org.eclipse.persistence.jaxb.MarshallerProperties.JSON_WRAPPER_AS_ARRAY_NAME, false);
- marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, false);
- marshaller.marshal(notificationEvent, result);
- this.sendToKafkaJmsQueue(result.toString());
-
- } catch (Exception e) {
- throw new AAIException("AAI_7350", e);
- }
- }
-
- public String storeEventOnly(Loader loader, Introspector eventHeader, Introspector obj) throws AAIException {
- if (obj == null) {
- throw new AAIException("AAI_7350");
- }
-
- try {
- final Introspector notificationEvent = loader.introspectorFromName("notification-event");
-
- if (eventHeader.getValue("id") == null) {
- eventHeader.setValue("id", genDate2() + "-" + UUID.randomUUID().toString());
- }
-
- if (eventHeader.getValue("timestamp") == null) {
- eventHeader.setValue("timestamp", genDate());
- }
-
- if (eventHeader.getValue("entity-link") == null) {
- eventHeader.setValue("entity-link", "UNK");
- }
-
- if (eventHeader.getValue("action") == null) {
- eventHeader.setValue("action", "UNK");
- }
-
- if (eventHeader.getValue("event-type") == null) {
- eventHeader.setValue("event-type", AAIConfig.get("aai.notificationEvent.default.eventType", "UNK"));
- }
-
- if (eventHeader.getValue("domain") == null) {
- eventHeader.setValue("domain", AAIConfig.get("aai.notificationEvent.default.domain", "UNK"));
- }
-
- if (eventHeader.getValue("source-name") == null) {
- eventHeader.setValue("source-name", AAIConfig.get("aai.notificationEvent.default.sourceName", "UNK"));
- }
-
- if (eventHeader.getValue("sequence-number") == null) {
- eventHeader.setValue("sequence-number",
- AAIConfig.get("aai.notificationEvent.default.sequenceNumber", "UNK"));
- }
-
- if (eventHeader.getValue("severity") == null) {
- eventHeader.setValue("severity", AAIConfig.get("aai.notificationEvent.default.severity", "UNK"));
- }
-
- if (eventHeader.getValue("version") == null) {
- eventHeader.setValue("version", AAIConfig.get("aai.notificationEvent.default.version", "UNK"));
- }
-
- if (notificationEvent.getValue("cambria-partition") == null) {
- notificationEvent.setValue("cambria-partition",
- AAIConfig.get("aai.notificationEvent.default.partition", AAIConstants.UEB_PUB_PARTITION_AAI));
- }
-
- notificationEvent.setValue("event-header", eventHeader.getUnderlyingObject());
- notificationEvent.setValue("entity", obj.getUnderlyingObject());
-
- String entityJson = notificationEvent.marshal(false);
- JSONObject entityJsonObject = new JSONObject(entityJson);
-
- JSONObject entityJsonObjectUpdated = new JSONObject();
-
- JSONObject entityHeader = entityJsonObject.getJSONObject("event-header");
- String cambriaPartition = entityJsonObject.getString("cambria.partition");
-
- entityJsonObject.remove("event-header");
- entityJsonObject.remove("cambria.partition");
-
- entityJsonObjectUpdated.put("event-header", entityHeader);
- entityJsonObjectUpdated.put("cambria.partition", cambriaPartition);
-
- Iterator<String> iter = entityJsonObject.keys();
- JSONObject entity = new JSONObject();
- if (iter.hasNext()) {
- entity = entityJsonObject.getJSONObject(iter.next());
- }
-
- entityJsonObjectUpdated.put("entity", entity);
-
- return entityJsonObjectUpdated.toString();
- } catch (JSONException e) {
- throw new AAIException("AAI_7350", e);
- } catch (AAIUnknownObjectException e) {
- throw new AAIException("AAI_7350", e);
- }
- }
-
- public String storeEventAndSendToJms(Loader loader, Introspector eventHeader, Introspector obj)
- throws AAIException {
- if (obj == null) {
- throw new AAIException("AAI_7350");
- }
-
- try {
- final Introspector notificationEvent = loader.introspectorFromName("notification-event");
-
- if (eventHeader.getValue("id") == null) {
- eventHeader.setValue("id", genDate2() + "-" + UUID.randomUUID().toString());
- }
-
- if (eventHeader.getValue("timestamp") == null) {
- eventHeader.setValue("timestamp", genDate());
- }
-
- if (eventHeader.getValue("entity-link") == null) {
- eventHeader.setValue("entity-link", "UNK");
- }
-
- if (eventHeader.getValue("action") == null) {
- eventHeader.setValue("action", "UNK");
- }
-
- if (eventHeader.getValue("event-type") == null) {
- eventHeader.setValue("event-type", AAIConfig.get("aai.notificationEvent.default.eventType", "UNK"));
- }
-
- if (eventHeader.getValue("domain") == null) {
- eventHeader.setValue("domain", AAIConfig.get("aai.notificationEvent.default.domain", "UNK"));
- }
-
- if (eventHeader.getValue("source-name") == null) {
- eventHeader.setValue("source-name", AAIConfig.get("aai.notificationEvent.default.sourceName", "UNK"));
- }
-
- if (eventHeader.getValue("sequence-number") == null) {
- eventHeader.setValue("sequence-number",
- AAIConfig.get("aai.notificationEvent.default.sequenceNumber", "UNK"));
- }
-
- if (eventHeader.getValue("severity") == null) {
- eventHeader.setValue("severity", AAIConfig.get("aai.notificationEvent.default.severity", "UNK"));
- }
-
- if (eventHeader.getValue("version") == null) {
- eventHeader.setValue("version", AAIConfig.get("aai.notificationEvent.default.version", "UNK"));
- }
-
- if (notificationEvent.getValue("cambria-partition") == null) {
- notificationEvent.setValue("cambria-partition",
- AAIConfig.get("aai.notificationEvent.default.partition", AAIConstants.UEB_PUB_PARTITION_AAI));
- }
-
- notificationEvent.setValue("event-header", eventHeader.getUnderlyingObject());
- notificationEvent.setValue("entity", obj.getUnderlyingObject());
-
- String entityJson = notificationEvent.marshal(false);
- sendToKafkaJmsQueue(entityJson);
- return entityJson;
- } catch (JSONException e) {
- throw new AAIException("AAI_7350", e);
- } catch (AAIUnknownObjectException e) {
- throw new AAIException("AAI_7350", e);
- }
- }
-
- private void sendToKafkaJmsQueue(String entityString) throws JSONException {
-
- JSONObject entityJsonObject = new JSONObject(entityString);
-
- JSONObject entityJsonObjectUpdated = new JSONObject();
- JSONObject finalJson = new JSONObject();
-
- JSONObject entityHeader = entityJsonObject.getJSONObject("event-header");
- String cambriaPartition = entityJsonObject.getString("cambria.partition");
-
- entityJsonObject.remove("event-header");
- entityJsonObject.remove("cambria.partition");
-
- entityJsonObjectUpdated.put("event-header", entityHeader);
- entityJsonObjectUpdated.put("cambria.partition", cambriaPartition);
-
- String transId = entityHeader.getString("id");
- String fromAppId = entityHeader.getString("source-name");
-
- Iterator<String> iter = entityJsonObject.keys();
- JSONObject entity = new JSONObject();
- if (iter.hasNext()) {
- entity = entityJsonObject.getJSONObject(iter.next());
- }
-
- entityJsonObjectUpdated.put("entity", entity);
-
- finalJson.put("event-topic", "AAI-EVENT");
- finalJson.put("transId", transId);
- finalJson.put("fromAppId", fromAppId);
- finalJson.put("fullId", "");
- finalJson.put("aaiEventPayload", entityJsonObjectUpdated);
-
- messageProducer.sendMessageToDefaultDestination(finalJson);
- }
-
- /**
- * Gen date.
- *
- * @return the string
- */
- public static String genDate() {
- FormatDate fd = new FormatDate("YYYYMMdd-HH:mm:ss:SSS");
- return fd.getDateTime();
- }
-
- /**
- * Gen date 2.
- *
- * @return the string
- */
- public static String genDate2() {
- FormatDate fd = new FormatDate("YYYYMMddHHmmss");
- return fd.getDateTime();
- }
-
-}
import java.util.Map;
import org.onap.aai.db.props.AAIProperties;
-import org.onap.aai.kafka.MessageProducer;
+import org.onap.aai.domain.deltaEvent.DeltaEvent;
+import org.onap.aai.domain.notificationEvent.NotificationEvent.EventHeader;
+import org.onap.aai.kafka.DeltaProducer;
import org.onap.aai.util.AAIConfig;
import org.springframework.beans.factory.annotation.Autowired;
-import com.google.gson.FieldNamingPolicy;
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import com.google.gson.JsonObject;
-
public class DeltaEvents {
-
- private static final Gson gson =
- new GsonBuilder().setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_DASHES).create();
- private static final String eventVersion = "v1";
-
private final String transId;
private final String sourceName;
private final String schemaVersion;
private final Map<String, ObjectDelta> objectDeltas;
- @Autowired private MessageProducer messageProducer;
+ @Autowired private DeltaProducer deltaProducer;
public DeltaEvents(String transId, String sourceName, String schemaVersion, Map<String, ObjectDelta> objectDeltas) {
- this.transId = transId;
- this.sourceName = sourceName;
- this.schemaVersion = schemaVersion;
- this.objectDeltas = objectDeltas;
+ this.transId = transId;
+ this.sourceName = sourceName;
+ this.schemaVersion = schemaVersion;
+ this.objectDeltas = objectDeltas;
}
public boolean triggerEvents() {
return false;
}
- JsonObject finalJson = new JsonObject();
- finalJson.addProperty("event-topic", "DELTA");
- finalJson.addProperty("transId", transId);
- finalJson.addProperty("fromAppId", sourceName);
- finalJson.addProperty("fullId", "");
- finalJson.add("aaiEventPayload", buildEvent());
-
- this.messageProducer.sendMessageToDefaultDestination(finalJson.toString());
+ deltaProducer.sendNotification(buildEvent());
return true;
}
- private JsonObject buildEvent() {
- JsonObject event = new JsonObject();
- event.addProperty("cambria.partition", this.getPartition());
- event.add("event-header", getHeader());
- event.add("entities", gson.toJsonTree(objectDeltas.values()));
- return event;
+ private DeltaEvent buildEvent() {
+ DeltaEvent deltaEvent = new DeltaEvent();
+ deltaEvent.setCambriaPartition(getPartition());
+ deltaEvent.setEventHeader(getHeader());
+ deltaEvent.setEntities(objectDeltas.values());
+ return deltaEvent;
}
private String getPartition() {
return "DELTA";
}
- private JsonObject getHeader() {
+ private EventHeader getHeader() {
ObjectDelta first = objectDeltas.values().iterator().next();
- JsonObject header = new JsonObject();
- header.addProperty("id", this.transId);
- header.addProperty("timestamp", this.getTimeStamp(first.getTimestamp()));
- header.addProperty("source-name", this.sourceName);
- header.addProperty("domain", this.getDomain());
- header.addProperty("event-type", this.getEventType());
- header.addProperty("event-version", eventVersion);
- header.addProperty("schema-version", this.schemaVersion);
- header.addProperty("action", first.getAction().toString());
- header.addProperty("entity-type", this.getEntityType(first));
- header.addProperty("entity-link", first.getUri());
- header.addProperty("entity-uuid", this.getUUID(first));
-
+ EventHeader header = new EventHeader();
+ header.setId(this.transId);
+ header.setTimestamp(this.getTimeStamp(first.getTimestamp()));
+ header.setSourceName(this.sourceName);
+ header.setDomain(this.getDomain());
+ header.setEventType(this.getEventType());
+ header.setVersion(this.schemaVersion);
+ header.setAction(first.getAction().toString());
+ header.setEntityType(this.getEntityType(first));
+ header.setEntityLink(first.getUri());
+ header.setEntityUuid(this.getUUID(first));
return header;
}
package org.onap.aai.util.delta;
-import com.google.gson.annotations.SerializedName;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import lombok.Data;
import java.util.ArrayList;
import java.util.HashMap;
import org.apache.commons.lang3.builder.ToStringBuilder;
+@Data
public class ObjectDelta {
- @SerializedName("uri")
private String uri;
-
- @SerializedName("action")
private DeltaAction action;
-
- @SerializedName("source-of-truth")
+ @JsonProperty("source-of-truth")
private String sourceOfTruth;
-
- @SerializedName("timestamp")
private long timestamp;
-
- @SerializedName("property-deltas")
+ @JsonProperty("property-deltas")
private Map<String, PropertyDelta> propertyDeltas = new HashMap<>();
-
- @SerializedName("relationship-deltas")
+ @JsonProperty("relationship-deltas")
private List<RelationshipDelta> relationshipDeltas = new ArrayList<>();
public ObjectDelta(String uri, DeltaAction action, String sourceOfTruth, long timestamp) {
relationshipDeltas.add(relationshipDelta);
}
- public String getUri() {
- return uri;
- }
-
- public void setUri(String uri) {
- this.uri = uri;
- }
-
- public DeltaAction getAction() {
- return action;
- }
-
- public void setAction(DeltaAction action) {
- this.action = action;
- }
-
- public String getSourceOfTruth() {
- return sourceOfTruth;
- }
-
- public void setSourceOfTruth(String sourceOfTruth) {
- this.sourceOfTruth = sourceOfTruth;
- }
-
- public long getTimestamp() {
- return timestamp;
- }
-
- public void setTimestamp(long timestamp) {
- this.timestamp = timestamp;
- }
-
- public void setPropertyDeltas(Map<String, PropertyDelta> propertyDeltas) {
- this.propertyDeltas = propertyDeltas;
- }
-
- public void setRelationshipDeltas(List<RelationshipDelta> relationshipDeltas) {
- this.relationshipDeltas = relationshipDeltas;
- }
-
- public Map<String, PropertyDelta> getPropertyDeltas() {
- return propertyDeltas;
- }
-
- public List<RelationshipDelta> getRelationshipDeltas() {
- return relationshipDeltas;
- }
-
@Override
public String toString() {
return new ToStringBuilder(this).append("uri", uri).append("action", action)
package org.onap.aai.util.delta;
-import com.google.gson.annotations.SerializedName;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import lombok.Data;
import org.apache.commons.lang3.builder.ToStringBuilder;
+@Data
public class PropertyDelta {
- @SerializedName("action")
protected DeltaAction action;
-
- @SerializedName("value")
protected Object value;
-
- @SerializedName("old-value")
+ @JsonProperty("old-value")
private Object oldValue;
public PropertyDelta(DeltaAction action, Object value) {
this.oldValue = oldValue;
}
- public DeltaAction getAction() {
- return action;
- }
-
- public void setAction(DeltaAction action) {
- this.action = action;
- }
-
- public Object getValue() {
- return value;
- }
-
- public void setValue(Object value) {
- this.value = value;
- }
-
- public Object getOldValue() {
- return oldValue;
- }
-
- public void setOldValue(Object oldValue) {
- this.oldValue = oldValue;
- }
-
@Override
public String toString() {
return new ToStringBuilder(this).append("action", action).append("value", value).append("oldValue", oldValue)
+++ /dev/null
-/**
- * ============LICENSE_START=======================================================
- * org.onap.aai
- * ================================================================================
- * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.aai.web;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import javax.annotation.PostConstruct;
-import javax.jms.ConnectionFactory;
-import javax.jms.MessageListener;
-import javax.jms.Queue;
-
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.command.ActiveMQQueue;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.onap.aai.introspection.LoaderFactory;
-import org.onap.aai.kafka.AAIKafkaEventJMSConsumer;
-import org.onap.aai.kafka.AAIKafkaEventJMSProducer;
-import org.onap.aai.kafka.MessageProducer;
-import org.onap.aai.kafka.NotificationProducer;
-import org.onap.aai.rest.notification.NotificationService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Profile;
-import org.springframework.jms.connection.CachingConnectionFactory;
-import org.springframework.jms.core.JmsTemplate;
-import org.springframework.jms.listener.DefaultMessageListenerContainer;
-import org.springframework.kafka.core.DefaultKafkaProducerFactory;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.core.ProducerFactory;
-
-@Profile("kafka")
-@Configuration
-public class KafkaConfig {
-
- @Value("${jms.bind.address}")
- private String bindAddress;
-
- @Value("${spring.kafka.producer.bootstrap-servers}")
- private String bootstrapServers;
-
- @Value("${spring.kafka.producer.properties.security.protocol}")
- private String securityProtocol;
-
- @Value("${spring.kafka.producer.properties.sasl.mechanism}")
- private String saslMechanism;
-
- @Value("${spring.kafka.producer.properties.sasl.jaas.config}")
- private String saslJaasConfig;
-
- @Value("${spring.kafka.producer.retries}")
- private String retries;
-
- private static final Logger logger = LoggerFactory.getLogger(KafkaConfig.class);
-
- @PostConstruct
- public void init() {
- System.setProperty("activemq.tcp.url", bindAddress);
- }
-
- @Bean(destroyMethod = "stop")
- public BrokerService brokerService() throws Exception {
-
- BrokerService broker = new BrokerService();
- broker.addConnector(bindAddress);
- broker.setPersistent(false);
- broker.setUseJmx(false);
- broker.setSchedulerSupport(false);
- broker.start();
-
- return broker;
- }
-
- @ConditionalOnMissingBean
- @Bean(name = "connectionFactory")
- public ConnectionFactory activeMQConnectionFactory() {
- return new ActiveMQConnectionFactory(bindAddress);
- }
-
- @Bean
- @ConditionalOnMissingBean
- public CachingConnectionFactory cachingConnectionFactory(ConnectionFactory targetConnectionFactory) {
- return new CachingConnectionFactory(targetConnectionFactory);
- }
-
- @Bean(name = "destinationQueue")
- public Queue activeMQQueue() {
- return new ActiveMQQueue("IN_QUEUE");
- }
-
- @Bean
- public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory, Queue queue) {
- JmsTemplate jmsTemplate = new JmsTemplate();
-
- jmsTemplate.setConnectionFactory(connectionFactory);
- jmsTemplate.setDefaultDestination(queue);
-
- return jmsTemplate;
- }
-
- @Bean(name = "jmsConsumer")
- public MessageListener jmsConsumer(KafkaTemplate<String, String> kafkaTemplate) throws Exception {
- return new AAIKafkaEventJMSConsumer(kafkaTemplate);
- }
-
- @Bean
- public DefaultMessageListenerContainer defaultMessageListenerContainer(ConnectionFactory connectionFactory, MessageListener messageListener)
- throws Exception {
-
- DefaultMessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer();
-
- messageListenerContainer.setConnectionFactory(connectionFactory);
- messageListenerContainer.setDestinationName("IN_QUEUE");
- messageListenerContainer.setMessageListener(messageListener);
-
- return messageListenerContainer;
- }
-
- @Bean
- public ProducerFactory<String, String> producerFactory() throws Exception {
- Map<String, Object> props = new HashMap<>();
- if (bootstrapServers == null) {
- logger.error("Environment Variable " + bootstrapServers + " is missing");
- throw new Exception("Environment Variable " + bootstrapServers + " is missing");
- } else {
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
- }
- props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
- props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
- props.put(ProducerConfig.RETRIES_CONFIG, retries);
- props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
-
- if (saslJaasConfig == null) {
- logger.info("Not using any authentication for kafka interaction");
- } else {
- logger.info("Using authentication provided by kafka interaction");
- // Strimzi Kafka security properties
- props.put("security.protocol", securityProtocol);
- props.put("sasl.mechanism", saslMechanism);
- props.put("sasl.jaas.config", saslJaasConfig);
- }
-
- return new DefaultKafkaProducerFactory<>(props);
- }
-
- @Bean
- public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) throws Exception {
- return new KafkaTemplate<>(producerFactory);
- }
-
- @Bean
- public MessageProducer messageProducer(JmsTemplate jmsTemplate) {
- return new AAIKafkaEventJMSProducer(jmsTemplate);
- }
-
- @Bean
- @ConditionalOnMissingBean
- public NotificationService notificationService(LoaderFactory loaderFactory,
- @Value("${schema.uri.base.path}") String basePath,
- @Value("${delta.events.enabled:false}") boolean isDeltaEventsEnabled, NotificationProducer notificationProducer) {
- return new NotificationService(null, loaderFactory, basePath, isDeltaEventsEnabled, notificationProducer);
- }
-}
<logger name="org.springframework" level="WARN" />
<logger name="org.springframework.beans" level="WARN" />
<logger name="org.springframework.web" level="WARN" />
- <logger name="com.blog.spring.jms" level="WARN" />
<logger name="com.jayway.jsonpath" level="WARN" />
<!-- AJSC Services (bootstrap services) -->
import org.onap.aai.setup.SchemaVersion;
import org.onap.aai.setup.SchemaVersions;
import org.onap.aai.util.AAIConstants;
-import org.onap.aai.web.KafkaConfig;
-import org.onap.aai.web.KafkaNotificationEventConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.test.context.ContextConfiguration;
classes = {ConfigConfiguration.class, AAIConfigTranslator.class, EdgeIngestor.class, EdgeSerializer.class,
NodeIngestor.class, SpringContextAware.class, IntrospectionConfig.class, RestBeanConfig.class,
XmlFormatTransformerConfiguration.class, ValidationService.class, ValidationConfiguration.class,
- KafkaConfig.class, LoaderFactory.class, NotificationService.class, KafkaNotificationEventConfig.class})
+ KafkaConfig.class, LoaderFactory.class, NotificationService.class, KafkaConfig.class})
@TestPropertySource(
properties = {"schema.uri.base.path = /aai", "schema.xsd.maxoccurs = 5000", "schema.translator.list=config",
"schema.nodes.location=src/test/resources/onap/oxm",
import org.onap.aai.setup.SchemaVersions;
import org.onap.aai.testutils.TestUtilConfigTranslatorforDataLink;
import org.onap.aai.util.AAIConstants;
-import org.onap.aai.web.KafkaNotificationEventConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.test.annotation.DirtiesContext;
@ContextConfiguration(
classes = {ConfigConfiguration.class, TestUtilConfigTranslatorforDataLink.class, EdgeIngestor.class,
EdgeSerializer.class, NodeIngestor.class, SpringContextAware.class, IntrospectionConfig.class,
- RestBeanConfig.class, XmlFormatTransformerConfiguration.class, LoaderFactory.class, NotificationService.class, KafkaNotificationEventConfig.class})
+ RestBeanConfig.class, XmlFormatTransformerConfiguration.class, LoaderFactory.class, NotificationService.class, KafkaConfig.class})
@TestPropertySource(
properties = {"schema.uri.base.path = /aai", "schema.xsd.maxoccurs = 5000", "schema.version.api.default = v4",
"schema.version.edge.label.start = v4", "schema.version.depth.start = v3",
@EmbeddedKafka(partitions = 1, topics = { "AAI-EVENT" })
@TestPropertySource(
properties = {
- "jms.bind.address=tcp://localhost:61647",
"aai.events.enabled=true",
"spring.kafka.producer.retries=0",
"spring.kafka.producer.properties.sasl.jaas.config=#{null}",
import org.junit.runner.RunWith;
import org.onap.aai.config.ConfigConfiguration;
import org.onap.aai.config.IntrospectionConfig;
+import org.onap.aai.config.KafkaConfig;
import org.onap.aai.config.SpringContextAware;
import org.onap.aai.config.XmlFormatTransformerConfiguration;
import org.onap.aai.db.props.AAIProperties;
import org.onap.aai.serialization.queryformats.QueryFormatTestHelper;
import org.onap.aai.setup.SchemaVersions;
import org.onap.aai.util.AAIConstants;
-import org.onap.aai.web.KafkaNotificationEventConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
@ContextConfiguration(
classes = {ConfigConfiguration.class, QueryTestsConfigTranslator.class, NodeIngestor.class, EdgeIngestor.class,
EdgeSerializer.class, SpringContextAware.class, IntrospectionConfig.class,
- XmlFormatTransformerConfiguration.class, LoaderFactory.class, NotificationService.class, KafkaNotificationEventConfig.class})
+ XmlFormatTransformerConfiguration.class, LoaderFactory.class, NotificationService.class, KafkaConfig.class})
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
@TestPropertySource(
properties = {"schema.translator.list = config", "schema.nodes.location=src/test/resources/onap/oxm",
import org.junit.runner.RunWith;
import org.onap.aai.config.ConfigConfiguration;
import org.onap.aai.config.IntrospectionConfig;
+import org.onap.aai.config.KafkaConfig;
import org.onap.aai.config.SpringContextAware;
import org.onap.aai.config.XmlFormatTransformerConfiguration;
import org.onap.aai.db.props.AAIProperties;
import org.onap.aai.setup.SchemaVersion;
import org.onap.aai.setup.SchemaVersions;
import org.onap.aai.util.AAIConstants;
-import org.onap.aai.web.KafkaNotificationEventConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.TestPropertySource;
classes = {ConfigConfiguration.class, AAICoreFakeEdgesConfigTranslator.class, NodeIngestor.class,
EdgeIngestor.class, EdgeSerializer.class, SpringContextAware.class, IntrospectionConfig.class,
XmlFormatTransformerConfiguration.class, LoaderFactory.class, NotificationService.class,
- KafkaNotificationEventConfig.class})
+ KafkaConfig.class})
@TestPropertySource(
properties = {"schema.translator.list = config", "schema.nodes.location=src/test/resources/onap/oxm",
"schema.edges.location=src/test/resources/onap/dbedgerules","aai.notifications.enabled=false"})
+++ /dev/null
-/**
- * ============LICENSE_START=======================================================
- * org.onap.aai
- * ================================================================================
- * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.aai.util;
-
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import com.fasterxml.jackson.core.JsonGenerationException;
-import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-
-import java.io.IOException;
-
-import org.eclipse.persistence.dynamic.DynamicEntity;
-import org.eclipse.persistence.jaxb.dynamic.DynamicJAXBContext;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.mockito.Mockito;
-import org.onap.aai.AAISetup;
-import org.onap.aai.domain.notificationEvent.NotificationEvent.EventHeader;
-import org.onap.aai.exceptions.AAIException;
-import org.onap.aai.introspection.Introspector;
-import org.onap.aai.introspection.Loader;
-import org.onap.aai.introspection.ModelType;
-import org.onap.aai.kafka.AAIKafkaEventJMSProducer;
-
-public class StoreNotificationEventTest extends AAISetup {
-
- private static AAIKafkaEventJMSProducer producer;
- private static StoreNotificationEvent sne;
-
- private final ObjectMapper mapper = new ObjectMapper();
-
- @BeforeClass
- public static void setUp() {
- producer = Mockito.mock(AAIKafkaEventJMSProducer.class);
- // sne = new StoreNotificationEvent(producer, "transiationId", "sourceOfTruth");
- }
-
- @Before
- public void setUpBefore() {
- producer = Mockito.mock(AAIKafkaEventJMSProducer.class);
- sne = new StoreNotificationEvent(producer, "transiationId", "sourceOfTruth");
-
- }
-
- @Test(expected = AAIException.class)
- public void testStoreEventNullObj() throws AAIException {
- sne.storeEventAndSendToJms(new EventHeader(), null);
- }
-
- @Test(expected = AAIException.class)
- public void testStoreEventInvalidObjForPojoUtils() throws AAIException {
- sne.storeEventAndSendToJms(new EventHeader(), new Object());
- }
-
- @Test
- public void testStoreEventEmptyEventHeader()
- throws AAIException, JsonGenerationException, JsonMappingException, IOException {
-
- ObjectNode object = mapper.createObjectNode().put("hello", "world");
- String res = sne.storeEventAndSendToJms(new EventHeader(), object);
-
- assertNotNull(res);
- assertTrue(res.contains("\"cambria.partition\" : \"" + AAIConstants.UEB_PUB_PARTITION_AAI + "\""));
- assertTrue(res.contains("\"event-header\""));
- assertTrue(res.contains("\"id\""));
- assertTrue(res.contains("\"timestamp\""));
- assertTrue(res
- .contains("\"source-name\" : \"" + AAIConfig.get("aai.notificationEvent.default.sourceName") + "\""));
- assertTrue(res.contains("\"domain\" : \"" + AAIConfig.get("aai.notificationEvent.default.domain") + "\""));
- assertTrue(res.contains(
- "\"sequence-number\" : \"" + AAIConfig.get("aai.notificationEvent.default.sequenceNumber") + "\""));
- assertTrue(res.contains("\"severity\" : \"" + AAIConfig.get("aai.notificationEvent.default.severity") + "\""));
- assertTrue(
- res.contains("\"event-type\" : \"" + AAIConfig.get("aai.notificationEvent.default.eventType") + "\""));
- assertTrue(res.contains("\"version\" : \"" + AAIConfig.get("aai.notificationEvent.default.version") + "\""));
- assertTrue(res.contains("\"action\" : \"UNK\""));
- assertTrue(res.contains("\"entity-link\" : \"UNK\""));
- assertTrue(res.contains("\"entity\""));
- assertTrue(res.contains("\"hello\""));
- assertTrue(res.contains("\"world\""));
- }
-
- @Test
- public void testStoreEvent() throws AAIException, JsonGenerationException, JsonMappingException, IOException {
- ObjectNode object = mapper.createObjectNode().put("hello", "world");
- EventHeader eh = new EventHeader();
- eh.setId("123");
- eh.setTimestamp("current-time");
- eh.setEntityLink("entity-link");
- eh.setAction("action!");
- eh.setEventType("surprise");
- eh.setDomain("PROD");
- eh.setSourceName("source");
- eh.setSequenceNumber("23");
- eh.setSeverity("ALERT");
- eh.setVersion("v12");
-
- String res = sne.storeEventAndSendToJms(eh, object);
-
- assertNotNull(res);
- assertTrue(res.contains("\"cambria.partition\" : \"" + AAIConstants.UEB_PUB_PARTITION_AAI + "\""));
- assertTrue(res.contains("\"event-header\""));
- assertTrue(res.contains("\"id\" : \"123\""));
- assertTrue(res.contains("\"timestamp\" : \"current-time\""));
- assertTrue(res.contains("\"source-name\" : \"source\""));
- assertTrue(res.contains("\"domain\" : \"PROD\""));
- assertTrue(res.contains("\"sequence-number\" : \"23\""));
- assertTrue(res.contains("\"severity\" : \"ALERT\""));
- assertTrue(res.contains("\"event-type\" : \"surprise\""));
- assertTrue(res.contains("\"version\" : \"v12\""));
- assertTrue(res.contains("\"action\" : \"action!\""));
- assertTrue(res.contains("\"entity-link\" : \"entity-link\""));
- assertTrue(res.contains("\"entity\""));
- assertTrue(res.contains("\"hello\""));
- assertTrue(res.contains("\"world\""));
- }
-
- @Test(expected = AAIException.class)
- public void testStoreDynamicEventNullObj() throws AAIException {
- DynamicEntity eventHeader = Mockito.mock(DynamicEntity.class);
- DynamicJAXBContext notificationJaxbContext =
- nodeIngestor.getContextForVersion(schemaVersions.getEdgeLabelVersion());
- sne.storeDynamicEvent(notificationJaxbContext, "v12", eventHeader, null);
- }
-
- @Test(expected = Exception.class)
- public void testStoreDynamicEventAAIException() throws Exception {
-
- DynamicJAXBContext notificationJaxbContext =
- nodeIngestor.getContextForVersion(schemaVersions.getEdgeLabelVersion());
- DynamicEntity obj = Mockito.mock(DynamicEntity.class);
- DynamicEntity eventHeader = Mockito.mock(DynamicEntity.class);
- sne.storeDynamicEvent(notificationJaxbContext, "v12", eventHeader, obj);
- }
-
- @Test(expected = AAIException.class)
- public void testStoreEventIntrospectorNullObj() throws Exception {
- Loader loader = Mockito.mock(Loader.class);
- sne.storeEventAndSendToJms(loader, null, null);
- }
-
- @Ignore("Stopped working since the model driven story")
- @Test
- public void testStoreEvent1Introspector() throws Exception {
- Loader loader = loaderFactory.createLoaderForVersion(ModelType.MOXY, schemaVersions.getEdgeLabelVersion());
- Introspector eventHeader = loader.introspectorFromName("notification-event-header");
- eventHeader.setValue("id", "123");
- eventHeader.setValue("timestamp", "current-time");
- eventHeader.setValue("entity-link", "entity-link");
- eventHeader.setValue("action", "action!");
- eventHeader.setValue("event-type", "surprise");
- eventHeader.setValue("domain", "PROD");
- eventHeader.setValue("source-name", "source");
- eventHeader.setValue("sequence-number", "23");
- eventHeader.setValue("severity", "ALERT");
- eventHeader.setValue("version", "v12");
- Introspector obj = loader.introspectorFromName("notification-event");
- String res = sne.storeEventAndSendToJms(loader, eventHeader, obj);
-
- assertNotNull(res);
- assertTrue(res.contains("\"cambria.partition\":\"" + AAIConstants.UEB_PUB_PARTITION_AAI + "\""));
- assertTrue(res.contains("\"event-header\""));
- assertTrue(res.contains("\"id\":\"123\""));
- assertTrue(res.contains("\"timestamp\":\"current-time\""));
- assertTrue(res.contains("\"source-name\":\"source\""));
- assertTrue(res.contains("\"domain\":\"PROD\""));
- assertTrue(res.contains("\"sequence-number\":\"23\""));
- assertTrue(res.contains("\"severity\":\"ALERT\""));
- assertTrue(res.contains("\"event-type\":\"surprise\""));
- assertTrue(res.contains("\"version\":\"v12\""));
- assertTrue(res.contains("\"action\":\"action!\""));
- assertTrue(res.contains("\"entity-link\":\"entity-link\""));
- assertTrue(res.contains("\"notification-event\""));
- }
-
- @Ignore("Stopped working since the model driven story")
- @Test
- public void testStoreEventIntrospectorEmptyEventHeader() throws Exception {
- Loader loader = loaderFactory.createLoaderForVersion(ModelType.MOXY, schemaVersions.getEdgeLabelVersion());
- Introspector eventHeader = loader.introspectorFromName("notification-event-header");
- Introspector obj = loader.introspectorFromName("notification-event");
-
- String res = sne.storeEventAndSendToJms(loader, eventHeader, obj);
-
- assertNotNull(res);
- assertTrue(res.contains("\"cambria.partition\":\"" + AAIConstants.UEB_PUB_PARTITION_AAI + "\""));
- assertTrue(res.contains("\"event-header\""));
- assertTrue(res.contains("\"id\""));
- assertTrue(res.contains("\"timestamp\""));
- assertTrue(
- res.contains("\"source-name\":\"" + AAIConfig.get("aai.notificationEvent.default.sourceName") + "\""));
- assertTrue(res.contains("\"domain\":\"" + AAIConfig.get("aai.notificationEvent.default.domain") + "\""));
- assertTrue(res.contains(
- "\"sequence-number\":\"" + AAIConfig.get("aai.notificationEvent.default.sequenceNumber") + "\""));
- assertTrue(res.contains("\"severity\":\"" + AAIConfig.get("aai.notificationEvent.default.severity") + "\""));
- assertTrue(res.contains("\"event-type\":\"" + AAIConfig.get("aai.notificationEvent.default.eventType") + "\""));
- assertTrue(res.contains("\"version\":\"" + AAIConfig.get("aai.notificationEvent.default.version") + "\""));
- assertTrue(res.contains("\"action\":\"UNK\""));
- assertTrue(res.contains("\"entity-link\":\"UNK\""));
- assertTrue(res.contains("\"notification-event\""));
- }
-}
aai.realtime.clients=RO,SDNC,SO
-aai.jms.enable=false
-
aai.rest.getall.depthparam=someuuid
aaf.valid.issuer.wildcard=aaf wild card issuer|aafWildCardIssuer|OU=another
-->
<spring.boot.version>2.5.15</spring.boot.version>
<spring.version>5.3.39</spring.version>
- <spring.jms.version>5.2.25.RELEASE</spring.jms.version>
<spring.test.version>${spring.version}</spring.test.version>
<json.path.version>2.2.0</json.path.version>
<json.version>20190722</json.version>
<artifactId>spring-security-rsa</artifactId>
<version>${spring.security.rsa.version}</version>
</dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-jms</artifactId>
- <version>5.3.39</version>
- </dependency>
-
-
<dependency>
<groupId>com.googlecode.json-simple</groupId>
Code Review / aai / aai-common.git / commitdiff