summaryrefslogtreecommitdiffstats
path: root/appc-event-listener/appc-event-listener-bundle
diff options
context:
space:
mode:
Diffstat (limited to 'appc-event-listener/appc-event-listener-bundle')
-rw-r--r--appc-event-listener/appc-event-listener-bundle/.gitignore7
-rw-r--r--appc-event-listener/appc-event-listener-bundle/pom.xml110
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AbstractListener.java159
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AppcEventListenerActivator.java185
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ListenerImpl.java128
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ProviderOperations.java296
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/WorkerImpl.java79
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Action.java55
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/CommonMessage.java181
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/IncomingMessage.java140
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/OutgoingMessage.java150
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Status.java74
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ListenerImpl.java130
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ProviderOperations.java291
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/WorkerImpl.java82
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Action.java55
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/CommonMessage.java340
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/IncomingMessage.java88
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/OutgoingMessage.java192
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Status.java69
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Controller.java55
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/EventHandler.java185
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/conv/Converter.java85
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/ListenerImpl.java159
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/WorkerImpl.java120
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ActionIdentifiers.java81
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/CommonHeader.java112
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapIncomingMessage.java50
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapMessage.java79
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapOutgoingMessage.java50
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/InputBody.java102
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/OutputBody.java98
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ResponseStatus.java70
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/GenericProviderOperationRequestFormatter.java115
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/OperationStatus.java66
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperationRequestFormatter.java42
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperations.java260
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Listener.java72
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/ListenerProperties.java319
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/ControllerImpl.java126
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java326
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/util/Mapper.java134
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/main/resources/org/openecomp/appc/default.properties99
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestListenerImpl.java43
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestProviderOperations.java122
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestWorker.java72
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/model/TestEnums.java58
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/model/TestMessages.java151
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL1607/model/TestEnums.java59
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL1607/model/TestMessages.java160
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/LCM/TestConverter.java91
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/LCM1607/model/TestJsonGenericMessages.java99
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestAbstractListener.java98
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestAppcDmaapListenerActivator.java46
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestListenerProperties.java154
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestController.java26
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestEventHandler.java171
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestListener.java112
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/util/TestMapper.java104
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/resources/IncomingMessage.txt24
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/resources/IncomingMessage1607.txt24
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/resources/OutgoingMessage.txt14
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/resources/OutgoingMessage1607.txt27
-rw-r--r--appc-event-listener/appc-event-listener-bundle/src/test/resources/org/openecomp/appc/default.properties89
64 files changed, 7360 insertions, 0 deletions
diff --git a/appc-event-listener/appc-event-listener-bundle/.gitignore b/appc-event-listener/appc-event-listener-bundle/.gitignore
new file mode 100644
index 000000000..76a83c1cf
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/.gitignore
@@ -0,0 +1,7 @@
+/bin/*
+/target/
+/.settings/
+/bin/
+/target/
+/logs/
+/debug-logs/ \ No newline at end of file
diff --git a/appc-event-listener/appc-event-listener-bundle/pom.xml b/appc-event-listener/appc-event-listener-bundle/pom.xml
new file mode 100644
index 000000000..1ebec4a74
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/pom.xml
@@ -0,0 +1,110 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.openecomp.appc</groupId>
+ <artifactId>appc-event-listener</artifactId>
+ <version>1.0.0</version>
+ </parent>
+
+ <artifactId>appc-event-listener-bundle</artifactId>
+ <packaging>bundle</packaging>
+ <name>Event Listener - bundle</name>
+
+ <dependencies>
+<!--
+ <dependency>
+ <groupId>org.openecomp.appc</groupId>
+ <artifactId>appc-rest-bundle</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+-->
+ <dependency>
+ <groupId>com.att.eelf</groupId>
+ <artifactId>eelf-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.openecomp.appc</groupId>
+ <artifactId>appc-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.openecomp.appc</groupId>
+ <artifactId>appc-dmaap-adapter-bundle</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+<!--
+ <dependency>
+ <groupId>org.openecomp.appc</groupId>
+ <artifactId>appc-rest-model</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+-->
+
+ <dependency>
+ <groupId>equinoxSDK381</groupId>
+ <artifactId>org.eclipse.osgi</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+ <extensions>true</extensions>
+ <configuration>
+ <instructions>
+ <Bundle-SymbolicName>org.openecomp.appc.listener</Bundle-SymbolicName>
+ <Bundle-Activator>org.openecomp.appc.listener.AppcEventListenerActivator</Bundle-Activator>
+ <Export-Package>org.openecomp.appc.listener</Export-Package>
+ <Import-Package>
+ !com.datastax.*,
+ !com.ibm.icu.text,
+ !com.jcraft.jzlib,
+ !groovy.lang,
+ !javax.jms,
+ !net.sf.ehcache,
+ !net.spy.memcached,
+ !org.apache.commons.lang3,
+ !org.apache.log,
+ !org.apache.log4j.*,
+ !org.apache.zookeeper.*,
+ !org.jasypt.*,
+ !org.codehaus.commons.compiler,
+ !org.codehaus.groovy.*,
+ !org.codehaus.jackson.*,
+ !org.codehaus.jackson.map.*,
+ !org.codehaus.janino,
+ !org.I0Itec.*,
+ !com.sun.faces.*,
+ *;resolution:=optional</Import-Package>
+ <Embed-Dependency>*;scope=compile|runtime;artifactId=!appc-metric-bundle|sli-common|org.eclipse.osgi|slf4j-api|jcl-over-slf4j|mysql-connector-java|xml-apis|pax-*</Embed-Dependency>
+ <Embed-Transitive>true</Embed-Transitive>
+ </instructions>
+ </configuration>
+ </plugin>
+
+ </plugins>
+ </build>
+</project>
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AbstractListener.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AbstractListener.java
new file mode 100644
index 000000000..b36f4d7e6
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AbstractListener.java
@@ -0,0 +1,159 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+import java.security.SecureRandom;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.openecomp.appc.listener.impl.EventHandlerImpl;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+public abstract class AbstractListener implements Listener {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(AbstractListener.class);
+
+ protected AtomicBoolean run = new AtomicBoolean(false);
+
+ protected int QUEUED_MIN = 1;
+ protected int QUEUED_MAX = 10;
+
+ protected int THREAD_MIN = 4;
+ protected int THREAD_MAX = THREAD_MIN; // Fixed thread pool
+ protected int THREAD_SCALE_DOWN_SEC = 10; // Number of seconds to wait until we remove idle threads
+
+ protected ThreadPoolExecutor executor;
+
+ protected EventHandler dmaap;
+
+ protected ListenerProperties props;
+
+ private String listenerId;
+
+ public AbstractListener(ListenerProperties props) {
+ updateProperties(props);
+
+ dmaap = new EventHandlerImpl(props);
+ if (dmaap.getClientId().equals("0")) {
+ dmaap.setClientId(String.valueOf(new SecureRandom().nextInt(1000)));
+ }
+
+ BlockingQueue<Runnable> threadQueue = new ArrayBlockingQueue<Runnable>(QUEUED_MAX + QUEUED_MIN + 1);
+ executor = new ThreadPoolExecutor(THREAD_MIN, THREAD_MAX, THREAD_SCALE_DOWN_SEC, TimeUnit.SECONDS, threadQueue,
+ new JobRejectionHandler());
+
+ // Custom Named thread factory
+ BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("DMaaP-Worker-%d").build();
+ executor.setThreadFactory(threadFactory);
+
+ run.set(true);
+ }
+
+ /**
+ * Starts a loop that will only end after stop() or stopNow() are called. The loop will read messages off the DMaaP
+ * topic and perform some action on them while writing messages back to DMaaP at critical points in the execution.
+ * Inherited from Runnable.
+ *
+ * @see java.lang.Runnable#run()
+ */
+ @Override
+ public void run() {
+ LOG.error("Listener.run() has not been implemented");
+ }
+
+ @Override
+ public void stop() {
+ run.set(false);
+ LOG.info(String.format("Stopping with %d messages in queue", executor.getQueue().size()));
+ executor.shutdown();
+ try {
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.error("Listener graceful stop() failed", e);
+ }
+
+ // close DMaaP clients
+ if (dmaap != null) {
+ dmaap.closeClients();
+ }
+ LOG.info("Listener Thread Pool Finished");
+ }
+
+ @Override
+ public void stopNow() {
+ run.set(false);
+ LOG.info(String.format("StopNow called. Orphaning %d messages in the queue", executor.getQueue().size()));
+ executor.getQueue().clear();
+ stop();
+ }
+
+ @Override
+ public String getBenchmark() {
+ return String.format("%s - No benchmarking implemented.", getListenerId());
+ }
+
+ @Override
+ public String getListenerId() {
+ return listenerId;
+ }
+
+ // Sets the id of the listener in
+ @Override
+ public void setListenerId(String id) {
+ listenerId = id;
+ }
+
+ private void updateProperties(ListenerProperties properties) {
+ this.props = properties;
+ QUEUED_MIN =
+ Integer.valueOf(props.getProperty(ListenerProperties.KEYS.THREADS_MIN_QUEUE, String.valueOf(QUEUED_MIN)));
+ QUEUED_MAX =
+ Integer.valueOf(props.getProperty(ListenerProperties.KEYS.THREADS_MAX_QUEUE, String.valueOf(QUEUED_MAX)));
+ THREAD_MIN =
+ Integer.valueOf(props.getProperty(ListenerProperties.KEYS.THREADS_MIN_POOL, String.valueOf(THREAD_MIN)));
+ THREAD_MAX =
+ Integer.valueOf(props.getProperty(ListenerProperties.KEYS.THREADS_MAX_POOL, String.valueOf(THREAD_MAX)));
+
+ listenerId = props.getPrefix();
+ }
+
+ /**
+ * This class will be used to handle what happens when we cannot add a job because of a ThreadPool issue. It does
+ * not get invoked if there is any fault with the job. NOTE: So far, this has only been seen when doing a
+ * {@link Listener#stopNow}
+ *
+ */
+ class JobRejectionHandler implements RejectedExecutionHandler {
+ @Override
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+ LOG.error(String.format("A job was rejected. [%s]", r));
+ }
+
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AppcEventListenerActivator.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AppcEventListenerActivator.java
new file mode 100644
index 000000000..4c2e28196
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AppcEventListenerActivator.java
@@ -0,0 +1,185 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+import org.openecomp.appc.configuration.Configuration;
+import org.openecomp.appc.configuration.ConfigurationFactory;
+import org.openecomp.appc.listener.impl.ControllerImpl;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import org.osgi.framework.BundleActivator;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceRegistration;
+
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * This activator is used to initialize and terminate the dmaap listener controller and pool(s)
+ * <p>
+ * The DMaaP listener is responsible for listening to a topic on the Universal Event Bus and reading in messages that
+ * conform to the DCAE message format for APPC. These messages will then be parsed and passed along to the APPC Provider
+ * to take action on. The listener will also send messages out on DMaaP during critical phases. The messages sent out will
+ * have a status of:
+ * <ul>
+ * <li><i>PENDING</i> - The listener has read the message off of DMaaP and has put it in the queue to be processed</li>
+ * <li><i>ACTIVE</i> - The listener has begun actually processing the request and is waiting on the appc provider to
+ * complete the request</li>
+ * <li><i>SUCCESS</i> or <i>FAILURE</i> - The listener has gotten a response back from the appc provider. If it is a
+ * FAILURE, a message should also be included</li>
+ * </ul>
+ * </p>
+ * <p>
+ * Activation of the bundle will provision 1 controller that in turn will provision 1 (or in the future more) listener
+ * to interact with DMaaP. Each listener will have a queue of messages read off of DMaaP and a thread pool of workers to
+ * process them. This worker is responsible for contacting appc provider to perform the action
+ * </p>
+ * <p>
+ * When the bundle is deactivated, the stopNow() method is called and the thread pool is emptied and all remaining jobs
+ * are orphaned. Alternatively stop() could be called which would allow all remaining jobs in the queue to complete at
+ * the cost of longer run time.
+ * </p>
+ *
+ * @since Aug 30, 2015
+ * @version $Id$
+ */
+public class AppcEventListenerActivator implements BundleActivator {
+
+ /**
+ * The bundle registration
+ */
+ private ServiceRegistration registration = null;
+
+ /**
+ * The configuration object
+ */
+ private Configuration configuration;
+
+ /**
+ * The bundle context
+ */
+ private static BundleContext context;
+
+ /**
+ * The reference to the actual implementation object that implements the services
+ */
+ private Controller adapter;
+
+ /**
+ * The logger to be used
+ */
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(AppcEventListenerActivator.class);
+
+ /**
+ * Called when this bundle is started so the Framework can perform the bundle-specific activities necessary to start
+ * this bundle. This method can be used to register services or to allocate any resources that this bundle needs.
+ * <p>
+ * This method must complete and return to its caller in a timely manner.
+ * </p>
+ *
+ * @param ctx
+ * The execution context of the bundle being started.
+ * @throws java.lang.Exception
+ * If this method throws an exception, this bundle is marked as stopped and the Framework will remove
+ * this bundle's listeners, unregister all services registered by this bundle, and release all services
+ * used by this bundle.
+ * @see org.osgi.framework.BundleActivator#start(org.osgi.framework.BundleContext)
+ */
+ @Override
+ public void start(BundleContext ctx) throws Exception {
+ LOG.info("Starting Bundle " + getName());
+
+ context = ctx;
+
+ configuration = ConfigurationFactory.getConfiguration();
+
+ Properties props = configuration.getProperties();
+
+ Set<ListenerProperties> listeners = new HashSet<ListenerProperties>();
+
+ // Configure App-C Closed Loop Listener
+/* ListenerProperties clProps = new ListenerProperties("appc.ClosedLoop", props);
+ clProps.setListenerClass(org.openecomp.appc.listener.CL.impl.ListenerImpl.class);
+ listeners.add(clProps);*/
+
+ // Configure App-C 1607 Closed Loop Listener
+ ListenerProperties cl1607Props = new ListenerProperties("appc.ClosedLoop1607", props);
+ cl1607Props.setListenerClass(org.openecomp.appc.listener.CL1607.impl.ListenerImpl.class);
+ listeners.add(cl1607Props);
+
+
+/* ListenerProperties clLCMProps = new ListenerProperties("appc.LCM", props);
+ clLCMProps.setListenerClass(org.openecomp.appc.listener.LCM.impl.ListenerImpl.class);
+ listeners.add(clLCMProps);*/
+
+/*
+ ListenerProperties clLCMProps1607 = new ListenerProperties("appc.LCM1607", props);
+ clLCMProps1607.setListenerClass(org.openecomp.appc.listener.LCM1607.impl.ListenerImpl.class);
+ listeners.add(clLCMProps1607);
+*/
+
+ adapter = new ControllerImpl(listeners);
+ if (ctx != null && registration == null) {
+ LOG.info("Registering service DMaaP Controller");
+ registration = ctx.registerService(Controller.class, adapter, null);
+ }
+ adapter.start();
+
+ LOG.info("DMaaP Listener started successfully");
+ }
+
+ /**
+ * Called when this bundle is stopped so the Framework can perform the bundle-specific activities necessary to stop
+ * the bundle. In general, this method should undo the work that the BundleActivator.start method started. There
+ * should be no active threads that were started by this bundle when this bundle returns. A stopped bundle must not
+ * call any Framework objects.
+ * <p>
+ * This method must complete and return to its caller in a timely manner.
+ * </p>
+ *
+ * @param ctx
+ * The execution context of the bundle being stopped.
+ * @throws java.lang.Exception
+ * If this method throws an exception, the bundle is still marked as stopped, and the Framework will
+ * remove the bundle's listeners, unregister all services registered by the bundle, and release all
+ * services used by the bundle. *
+ * @see org.osgi.framework.BundleActivator#stop(org.osgi.framework.BundleContext)
+ */
+ @Override
+ public void stop(BundleContext ctx) throws Exception {
+ boolean stopNow = true;
+ LOG.info("Stopping DMaaP Listener. StopNow=" + stopNow);
+ adapter.stop(stopNow);
+ if (registration != null) {
+ registration.unregister();
+ registration = null;
+ }
+ LOG.info("DMaaP Listener stopped successfully");
+ }
+
+ public String getName() {
+ return "DMaaP Listener";
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ListenerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ListenerImpl.java
new file mode 100644
index 000000000..3f0f8a112
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ListenerImpl.java
@@ -0,0 +1,128 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.impl;
+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.concurrent.RejectedExecutionException;
+
+import org.openecomp.appc.listener.AbstractListener;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.CL.model.IncomingMessage;
+import org.openecomp.appc.listener.CL.model.Status;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import com.att.eelf.i18n.EELFResourceManager;
+
+public class ListenerImpl extends AbstractListener {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(ListenerImpl.class);
+
+ private long startTime = 0;
+
+ public ListenerImpl(ListenerProperties props) {
+ super(props);
+ String url = props.getProperty("provider.url");
+ LOG.info("DMaaP Provider Endpoint: " + url);
+ ProviderOperations.setUrl(url);
+
+ // Set Basic Auth
+ String user = props.getProperty("provider.user");
+ String pass = props.getProperty("provider.pass");
+ ProviderOperations.setAuthentication(user, pass);
+ }
+
+ @Override
+ public void run() {
+ // Some vars for benchmarking
+ startTime = System.currentTimeMillis();
+
+ LOG.info("Running DMaaP Listener");
+
+ while (run.get()) {
+ // Only update if the queue is low. otherwise we read in more
+ // messages than we need
+ try {
+ if (executor.getQueue().size() <= QUEUED_MIN) {
+ LOG.debug("DMaaP queue running low. Querying for more jobs");
+ List<IncomingMessage> messages = dmaap.getIncomingEvents(IncomingMessage.class, QUEUED_MAX);
+ LOG.debug(String.format("Read %d messages from dmaap", messages.size()));
+ for (IncomingMessage incoming : messages) {
+ // Acknowledge that we read the event
+ LOG.info("Acknowledging Message: " + incoming.getId());
+ dmaap.postStatus(incoming.toOutgoing(Status.PENDING, null).toString());
+ }
+ for (IncomingMessage incoming : messages) {
+ // Add to pool if still running
+ if (run.get()) {
+ LOG.info(String.format("Adding DMaaP message to pool queue [%s]", incoming.getId()));
+ if (incoming.isValid()) {
+ try {
+ executor.execute(new WorkerImpl(incoming, dmaap));
+ } catch (RejectedExecutionException rejectEx) {
+ LOG.error("Task Rejected: ", rejectEx);
+ }
+ } else {
+ // Badly formed message
+ LOG.error("Message was not valid. Rejecting");
+ }
+ } else {
+ LOG.info("Run stopped. Orphaning Message: " + incoming.getId());
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOG.error("Exception " + e.getClass().getSimpleName() + " caught in DMaaP listener");
+ LOG.error(EELFResourceManager.format(e));
+ LOG.error("DMaaP Listener logging and ignoring the exception, continue...");
+ }
+ }
+
+ LOG.info("Stopping DMaaP Listener thread");
+
+ // We've told the listener to stop
+ // TODO - Should we:
+ // 1) Put a message back on the queue indicating that APP-C never got to
+ // the message
+ // or
+ // 2) Let downstream figure it out after timeout between PENDING and
+ // ACTIVE messages
+ }
+
+ @Override
+ public String getBenchmark() {
+ long time = System.currentTimeMillis();
+ DateFormat df = new SimpleDateFormat("HH:mm:ss");
+ df.setTimeZone(TimeZone.getTimeZone("UTC"));
+ String runningTime = df.format(new Date(time - startTime));
+
+ String out = String.format("Running for %s and completed %d jobs using %d threads.", runningTime,
+ executor.getCompletedTaskCount(), executor.getPoolSize());
+ LOG.info("***BENCHMARK*** " + out);
+ return out;
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ProviderOperations.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ProviderOperations.java
new file mode 100644
index 000000000..7a53bcdd9
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ProviderOperations.java
@@ -0,0 +1,296 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.impl;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.MalformedURLException;
+import java.net.Socket;
+import java.net.URL;
+import java.net.UnknownHostException;
+import java.security.KeyManagementException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.io.IOUtils;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpVersion;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.conn.ClientConnectionManager;
+import org.apache.http.conn.scheme.PlainSocketFactory;
+import org.apache.http.conn.scheme.Scheme;
+import org.apache.http.conn.scheme.SchemeRegistry;
+import org.apache.http.conn.ssl.SSLSocketFactory;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
+import org.apache.http.params.BasicHttpParams;
+import org.apache.http.params.HttpParams;
+import org.apache.http.params.HttpProtocolParams;
+import org.apache.http.protocol.HTTP;
+import org.json.JSONObject;
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.CL.model.IncomingMessage;
+import org.openecomp.appc.listener.util.Mapper;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+public class ProviderOperations {
+
+ private static final EELFLogger LOG = EELFManager.getInstance().getLogger(ProviderOperations.class);
+
+ private static URL url;
+
+ private static String basic_auth;
+
+ //@formatter:off
+ @SuppressWarnings("nls")
+ private final static String TEMPLATE = "{\"input\": {\"common-request-header\": {\"service-request-id\": \"%s\"},\"vnf-resource\": {\"vm-id\": \"%s\"%s}}}";
+ //@formatter:on
+
+ /**
+ * Calls the AppcProvider to run a topology directed graph
+ *
+ * @param msg
+ * The incoming message to be run
+ * @return True if the result is success. Never returns false and throws an exception instead.
+ * @throws UnsupportedEncodingException
+ * @throws Exception
+ * if there was a failure processing the request. The exception message is the failure reason.
+ */
+ @SuppressWarnings("nls")
+ public static boolean topologyDG(IncomingMessage msg) throws APPCException {
+ if (msg == null) {
+ throw new APPCException("Provided message was null");
+ }
+
+ HttpPost post = null;
+ try {
+ // Concatenate the "action" on the end of the URL
+ String path = url.getPath() + ":" + msg.getAction().getValue().toLowerCase();
+ URL serviceUrl = new URL(url.getProtocol(), url.getHost(), url.getPort(), path);
+
+ post = new HttpPost(serviceUrl.toExternalForm());
+ post.setHeader("Content-Type", "application/json");
+ post.setHeader("Accept", "application/json");
+
+ // Set Auth
+ if (basic_auth != null) {
+ post.setHeader("Authorization", "Basic " + basic_auth);
+ }
+
+ String body = buildReqest(msg.getId(), msg.getUrl(), msg.getIdentityUrl());
+
+
+ LOG.info(String.format("DMaaP ACTION PATH : %s", path));
+ LOG.info(String.format("DMaaP ACTION BODY : %s", body));
+
+ StringEntity entity = new StringEntity(body);
+ entity.setContentType("application/json");
+ post.setEntity(new StringEntity(body));
+ } catch (UnsupportedEncodingException | MalformedURLException e) {
+ throw new APPCException(e);
+ }
+
+ HttpClient client = getHttpClient();
+
+ int httpCode = 0;
+ String respBody = null;
+ try {
+ HttpResponse response = client.execute(post);
+ httpCode = response.getStatusLine().getStatusCode();
+ respBody = IOUtils.toString(response.getEntity().getContent());
+ } catch (IOException e) {
+ throw new APPCException(e);
+ }
+
+ if (httpCode >= 200 && httpCode < 300 && respBody != null) {
+ JSONObject json;
+ try {
+ json = Mapper.toJsonObject(respBody);
+ } catch (Exception e) {
+ LOG.error("Error processing response from provider. Could not map response to json", e);
+ throw new APPCException("APPC has an unknown RPC error");
+ }
+ boolean success;
+ String reason;
+ try {
+ JSONObject header = json.getJSONObject("output").getJSONObject("common-response-header");
+ success = header.getBoolean("success");
+ reason = header.getString("reason");
+ } catch (Exception e) {
+ LOG.error("Unknown error prcoessing failed response from provider. Json not in expected format", e);
+ throw new APPCException("APPC has an unknown RPC error");
+ }
+ if (success) {
+ return true;
+ }
+ String reasonStr = reason == null ? "Unknown" : reason;
+ LOG.warn(String.format("Topology Operation [%s] failed. Reason: %s", msg.getId(), reasonStr));
+ throw new APPCException(reasonStr);
+
+ }
+ throw new APPCException(String.format("Unexpected response from endpoint: [%d] - %s ", httpCode, respBody));
+ }
+
+ /**
+ * Updates the static var URL and returns the value;
+ *
+ * @return The new value of URL
+ */
+ public static String getUrl() {
+ return url.toExternalForm();
+ }
+
+ public static void setUrl(String newUrl) {
+ try {
+ url = new URL(newUrl);
+ } catch (MalformedURLException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * Sets the basic authentication header for the given user and password. If either entry is null then set basic auth
+ * to null
+ *
+ * @param user
+ * The user with optional domain name
+ * @param password
+ * The password for the user
+ * @return The new value of the basic auth string that will be used in the request headers
+ */
+ public static String setAuthentication(String user, String password) {
+ if (user != null && password != null) {
+ String authStr = user + ":" + password;
+ basic_auth = new String(Base64.encodeBase64(authStr.getBytes()));
+ } else {
+ basic_auth = null;
+ }
+ return basic_auth;
+ }
+
+ /**
+ * Builds the request body for a topology operation
+ *
+ * @param id
+ * The request id
+ * @param action
+ * The action in lowercase
+ * @param url
+ * The vm's url
+ * @return A String containing the request body
+ */
+ private static String buildReqest(String id, String url, String ident) {
+ String extraVmResource = "";
+ if (ident != null) {
+ extraVmResource = String.format(", \"identity-url\": \"%s\"", ident);
+ }
+ return String.format(TEMPLATE, id, url, extraVmResource);
+ }
+
+ @SuppressWarnings("deprecation")
+ private static HttpClient getHttpClient() throws APPCException {
+ HttpClient client;
+ if (url.getProtocol().equals("https")) {
+ try {
+ KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
+ trustStore.load(null, null);
+ MySSLSocketFactory sf = new MySSLSocketFactory(trustStore);
+ sf.setHostnameVerifier(MySSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
+
+ HttpParams params = new BasicHttpParams();
+ HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1);
+ HttpProtocolParams.setContentCharset(params, HTTP.UTF_8);
+
+ SchemeRegistry registry = new SchemeRegistry();
+ registry.register(new Scheme("http", PlainSocketFactory.getSocketFactory(), 80));
+ registry.register(new Scheme("https", sf, 443));
+ registry.register(new Scheme("https", sf, 8443));
+ registry.register(new Scheme("http", sf, 8181));
+
+ ClientConnectionManager ccm = new ThreadSafeClientConnManager(params, registry);
+ client = new DefaultHttpClient(ccm, params);
+ } catch (Exception e) {
+ client = new DefaultHttpClient();
+ }
+ } else if (url.getProtocol().equals("http")) {
+ client = new DefaultHttpClient();
+ } else {
+ throw new APPCException(
+ "The provider.topology.url property is invalid. The url did not start with http[s]");
+ }
+ return client;
+ }
+
+ @SuppressWarnings("deprecation")
+ public static class MySSLSocketFactory extends SSLSocketFactory {
+ private SSLContext sslContext = SSLContext.getInstance("TLS");
+
+ public MySSLSocketFactory(KeyStore truststore) throws NoSuchAlgorithmException, KeyManagementException,
+ KeyStoreException, UnrecoverableKeyException {
+ super(truststore);
+
+ TrustManager tm = new X509TrustManager() {
+ @Override
+ public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+ }
+
+ @Override
+ public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+ }
+
+ @Override
+ public X509Certificate[] getAcceptedIssuers() {
+ return null;
+ }
+ };
+
+ sslContext.init(null, new TrustManager[] {
+ tm
+ }, null);
+ }
+
+ @Override
+ public Socket createSocket(Socket socket, String host, int port, boolean autoClose)
+ throws IOException, UnknownHostException {
+ return sslContext.getSocketFactory().createSocket(socket, host, port, autoClose);
+ }
+
+ @Override
+ public Socket createSocket() throws IOException {
+ return sslContext.getSocketFactory().createSocket();
+ }
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/WorkerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/WorkerImpl.java
new file mode 100644
index 000000000..1eb13f736
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/WorkerImpl.java
@@ -0,0 +1,79 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.impl;
+
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.EventHandler;
+import org.openecomp.appc.listener.CL.model.IncomingMessage;
+import org.openecomp.appc.listener.CL.model.Status;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+public class WorkerImpl implements Runnable {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(WorkerImpl.class);
+
+ // Should have all of the data we need for processing
+ private IncomingMessage event;
+
+ // So we can post messages from inside the worker.
+ private EventHandler dmaap;
+
+ public WorkerImpl(IncomingMessage message, EventHandler dmaap) {
+ this.event = message;
+ this.dmaap = dmaap;
+ }
+
+ @Override
+ public void run() {
+ LOG.debug(String.format("Started working on %s", event.getId()));
+ dmaap.postStatus(event.toOutgoing(Status.ACTIVE, null));
+ // Run the dg in a try catch to handle all exceptions and update the
+ // message at the end
+ try {
+ if (doDG(event)) {
+ dmaap.postStatus(event.toOutgoing(Status.SUCCESS, null));
+ LOG.debug(String.format("Event %s finished successfully", event.getId()));
+ } else {
+ // Should never happen. Exception with message should be thrown instead.
+ LOG.error(String.format(
+ "We somehow returned false from doDG() instead of throwing exception. Incoming event [%s]",
+ event.toJson().toString()));
+ dmaap.postStatus(event.toOutgoing(Status.FAILURE));
+ }
+
+ } catch (Exception e) {
+ // Unknown exception from DG method. Fail and pass the exception
+ // along
+ String msg = "Exception: " + e.getMessage();
+ LOG.warn(String.format("Event %s finished with failure. %s", event.getId(), msg));
+ dmaap.postStatus(event.toOutgoing(Status.FAILURE, msg));
+ }
+
+ LOG.debug("Done working on " + event.getId());
+ }
+
+ private boolean doDG(IncomingMessage msg) throws APPCException {
+ return ProviderOperations.topologyDG(msg);
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Action.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Action.java
new file mode 100644
index 000000000..a8fe8035b
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Action.java
@@ -0,0 +1,55 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.model;
+
+public enum Action {
+ Restart("Restart"), Rebuild("Rebuild"), Migrate("Migrate"), Evacuate("Evacuate");
+
+ /**
+ * Converts the string to an Action
+ *
+ * @param value
+ * The string to try and convert. Is case insensitive
+ * @return The action matching the string or null if no match was found.
+ */
+ public static Action toAction(String value) {
+ if (value != null) {
+ for (Action e : values()) {
+ if (e.getValue().toUpperCase().equals(value.toUpperCase())) {
+ return e;
+ }
+ }
+ }
+
+ return null;
+ }
+
+ private String value;
+
+ private Action(String valueToUse) {
+ value = valueToUse;
+ }
+
+ public final String getValue() {
+ return value;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/CommonMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/CommonMessage.java
new file mode 100644
index 000000000..76a112618
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/CommonMessage.java
@@ -0,0 +1,181 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.model;
+
+import java.io.Serializable;
+
+import org.json.JSONObject;
+import org.openecomp.appc.listener.util.Mapper;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion;
+
+/**
+ * This class holds attributes that are common to DMaaP messages both coming in from DCAE and being sent out by APPC
+ *
+ */
+@JsonSerialize(include = Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CommonMessage implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /*
+ * The unique id of the event as of 1602
+ */
+ @JsonProperty("eventID")
+ private String id;
+
+ /*
+ * The time that the request was sent out.
+ */
+ @JsonProperty("requestTime")
+ private String requestTime;
+
+ /*
+ * The originator of the event
+ */
+ @JsonProperty("requestClient")
+ private String requestClient;
+
+ /*
+ * The system that sent the message
+ */
+ @JsonProperty("from")
+ private String fromSystem;
+
+ /*
+ * The actual trap message
+ */
+ @JsonProperty("message")
+ private String message;
+
+ /*
+ * The vm name associated with the event
+ */
+ @JsonProperty("VMName")
+ private String vmName;
+
+ /*
+ * The policy name on the incoming event
+ */
+ @JsonProperty("policyName")
+ private String policyName;
+
+ /*
+ * The policy version on the incoming event
+ */
+ @JsonProperty("policyVersion")
+ private String policyVersion;
+
+ @JsonIgnore
+ private long startTime = System.currentTimeMillis();
+
+ /*
+ * Getters and Setters
+ */
+
+ public String getId() {
+ return id;
+ }
+
+ public String getRequestTime() {
+ return requestTime;
+ }
+
+ public String getRequestClient() {
+ return requestClient;
+ }
+
+ public String getFromSystem() {
+ return fromSystem;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+ public String getPolicyName() {
+ return policyName;
+ }
+
+ public String getPolicyVersion() {
+ return policyVersion;
+ }
+
+ public String getVmName() {
+ return vmName;
+ }
+
+ public long getStartTime() {
+ return startTime;
+ }
+
+ public void setId(String eventId) {
+ id = eventId;
+ }
+
+ public void setRequestTime(String requestTime) {
+ this.requestTime = requestTime;
+ }
+
+ public void setRequestClient(String requestClient) {
+ this.requestClient = requestClient;
+ }
+
+ public void setFromSystem(String fromSystem) {
+ this.fromSystem = fromSystem;
+ }
+
+ public void setMessage(String message) {
+ this.message = message;
+ }
+
+ public void setPolicyName(String name) {
+ policyName = name;
+ }
+
+ public void setPolicyVersion(String version) {
+ policyVersion = version;
+ }
+
+ public void setVmName(String vmName) {
+ this.vmName = vmName;
+ }
+
+ public void setStartTime(long startTime) {
+ this.startTime = startTime;
+ }
+
+ /**
+ * Convenience method to return a json representation of this object.
+ *
+ * @return The json representation of this object
+ */
+ public JSONObject toJson() {
+ return Mapper.toJsonObject(this);
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/IncomingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/IncomingMessage.java
new file mode 100644
index 000000000..60d1d004b
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/IncomingMessage.java
@@ -0,0 +1,140 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion;
+
+/**
+ * This class reperesnts a message coming in from DCAE.
+ *
+ */
+@JsonSerialize(include = Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class IncomingMessage extends CommonMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ /*
+ * The action being requested. Its presence signals that it is an incoming message and it is not present on outgoing
+ * messages
+ */
+ @JsonProperty("request")
+ private String request;
+
+ /*
+ * The url for the server used for auth. http://<compute>:<port>/<tenentId>/server/<serverID>
+ */
+ @JsonProperty("VServerSelfLink")
+ private String Url;
+
+ /*
+ * The tenant Id in OpenStack that the server belongs to
+ */
+ @JsonProperty("TenantID")
+ private String TenantId;
+
+ /*
+ * The VM's UUID in
+ */
+ @JsonProperty("VMID")
+ private String VmId;
+
+ @JsonProperty("Identity")
+ private String identityUrl;
+
+ public String getRequest() {
+ return request;
+ }
+
+ @JsonIgnore
+ public Action getAction() {
+ return Action.toAction(request);
+ }
+
+ public String getUrl() {
+ return Url;
+ }
+
+ public String getTenantId() {
+ return TenantId;
+ }
+
+ public String getVmId() {
+ return VmId;
+ }
+
+ public String getIdentityUrl() {
+ return identityUrl;
+ }
+
+ public void setUrl(String Url) {
+ this.Url = Url;
+ }
+
+ public void setTenantId(String TenantId) {
+ this.TenantId = TenantId;
+ }
+
+ public void setVmId(String VmId) {
+ this.VmId = VmId;
+ }
+
+ public void setRequest(String request) {
+ this.request = request;
+ }
+
+ public void setIdentityUrl(String identityUrl) {
+ this.identityUrl = identityUrl;
+ }
+
+ @Override
+ public String toString() {
+ String time = getRequestTime() != null ? getRequestTime() : "N/A";
+ // String req = request != null ? request : "N/A";
+ return String.format("[%s - %s]", time, getId());
+ }
+
+ public String toOutgoing(Status status) {
+ return toOutgoing(status, getMessage());
+ }
+
+ public String toOutgoing(Status status, String msg) {
+ OutgoingMessage out = new OutgoingMessage(this);
+ out.setResponse(status);
+ out.setMessage(msg);
+ return out.toResponse().toString();
+ }
+
+ /**
+ * Determines if this message should be parsed parsed. Will eventually check that the message is well formed, has
+ * all required fields, and had not exceeded any timing restrictions.
+ *
+ * @return True if the message should be parsed. False otherwise
+ */
+ public boolean isValid() {
+ return true;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/OutgoingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/OutgoingMessage.java
new file mode 100644
index 000000000..fa8560ed2
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/OutgoingMessage.java
@@ -0,0 +1,150 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.model;
+
+import java.net.InetAddress;
+import java.security.SecureRandom;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.TimeZone;
+
+import org.json.JSONObject;
+import org.openecomp.appc.listener.util.Mapper;
+import org.openecomp.appc.util.Time;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion;
+
+/**
+ * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request
+ *
+ */
+@JsonSerialize(include = Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class OutgoingMessage extends CommonMessage {
+
+ private static final long serialVersionUID = -5447940920271469613L;
+
+ @JsonProperty("response")
+ private Status response;
+
+ @JsonProperty("responseTime")
+ private String responseTime;
+
+ @JsonProperty("originalRequest")
+ private String originalRequest;
+
+ public OutgoingMessage() {
+
+ }
+
+ public OutgoingMessage(IncomingMessage msg) {
+ setId(msg.getId());
+ setOriginalRequest(msg.getRequest());
+ setRequestClient(msg.getRequestClient());
+ setRequestTime(msg.getRequestTime());
+ setVmName(msg.getVmName());
+ setFromSystem(generateFrom());
+ setResponse(Status.PENDING);
+ setPolicyName(msg.getPolicyName());
+ setPolicyVersion(msg.getPolicyVersion());
+ setStartTime(msg.getStartTime());
+ }
+
+ @JsonProperty("duration")
+ public long getDuration() {
+ return System.currentTimeMillis() - getStartTime();
+ }
+
+ public Status getResponse() {
+ return response;
+ }
+
+ public String getResponseTime() {
+ return responseTime;
+ }
+
+ public String getOriginalRequest() {
+ return originalRequest;
+ }
+
+ @JsonIgnore
+ public void setResponse(Status response) {
+ this.response = response;
+ }
+
+ public void setResponse(String responseString) {
+ this.response = Status.valueOf(responseString);
+ }
+
+ public void setResponseTime(String responseTime) {
+ this.responseTime = responseTime;
+ }
+
+ public void setOriginalRequest(String originalRequest) {
+ this.originalRequest = originalRequest;
+ }
+
+ public void updateResponseTime() {
+ SecureRandom rand = new SecureRandom();
+ SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd kk:mm:ss.SSS");
+ df.setTimeZone(TimeZone.getTimeZone("UTC"));
+ String date = df.format(new Date(Time.utcTime()));
+ this.responseTime = String.format("%s%03d", date, rand.nextInt(1000));
+ }
+
+ public String generateFrom() {
+ String name;
+ try {
+ InetAddress iAddress = InetAddress.getLocalHost();
+ name = iAddress.getCanonicalHostName();
+ } catch (Exception e) {
+ // Could not get anything from the InetAddress
+ name = "UnknownHost";
+ }
+ return "appc@" + name;
+ }
+
+ public JSONObject toResponse() {
+ updateResponseTime();
+ JSONObject json = Mapper.toJsonObject(this);
+
+ if (!json.has("message")) {
+ // If there is no message, parrot the status (response field)
+ // TODO - Can this be removed for 1602 making message truely optional?
+ json.put("message", this.getResponse().toString());
+ }
+
+ // Removed duplication of status from message for 1602
+ // json.put("message", String.format("%s: %s", request, json.get("message")));
+
+ return json;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s - %s", getId(), getResponse());
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Status.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Status.java
new file mode 100644
index 000000000..d8e8b5c64
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Status.java
@@ -0,0 +1,74 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.model;
+
+public enum Status {
+ /*
+ * APP-C acknowledges that it has read the event off of the wire. This is the initial status of an OutgoingEvent
+ */
+ PENDING("PENDING"),
+
+ /*
+ * APP-C has started processing the event
+ */
+ ACTIVE("ACTIVE"),
+
+ /*
+ * APP-C has finished processing the event without errors
+ */
+ SUCCESS("SUCCESS"),
+
+ /*
+ * APP-C has finished processing the event with errors
+ */
+ FAILURE("FAILURE");
+
+ /**
+ * Converts the string to an Status
+ *
+ * @param value
+ * The string to try and convert. Is case insensitive
+ * @return The status matching the string or null if no match was found.
+ */
+ public static Status toStatus(String value) {
+ if (value != null) {
+ for (Status e : values()) {
+ if (e.getValue().toUpperCase().equals(value.toUpperCase())) {
+ return e;
+ }
+ }
+ }
+
+ return null;
+ }
+
+ private String value;
+
+ private Status(String valueToUse) {
+ value = valueToUse;
+ }
+
+ public final String getValue() {
+ return value;
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ListenerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ListenerImpl.java
new file mode 100644
index 000000000..5d3b741ff
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ListenerImpl.java
@@ -0,0 +1,130 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.impl;
+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.concurrent.RejectedExecutionException;
+
+import org.openecomp.appc.listener.AbstractListener;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.CL1607.model.IncomingMessage;
+import org.openecomp.appc.listener.CL1607.model.Status;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import com.att.eelf.i18n.EELFResourceManager;
+
+public class ListenerImpl extends AbstractListener {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(ListenerImpl.class);
+
+ private long startTime = 0;
+
+ public ListenerImpl(ListenerProperties props) {
+ super(props);
+ String url = props.getProperty("provider.url");
+ LOG.info("DMaaP Provider Endpoint: " + url);
+ ProviderOperations.setUrl(url);
+
+ // Set Basic Auth
+ String user = props.getProperty("provider.user");
+ String pass = props.getProperty("provider.pass");
+ ProviderOperations.setAuthentication(user, pass);
+ }
+
+ @Override
+ public void run() {
+ // Some vars for benchmarking
+ startTime = System.currentTimeMillis();
+
+ LOG.info("Running DMaaP Listener");
+
+ while (run.get()) {
+ // Only update if the queue is low. otherwise we read in more
+ // messages than we need
+ try {
+ if (executor.getQueue().size() <= QUEUED_MIN) {
+ LOG.debug("DMaaP queue running low. Querying for more jobs");
+ List<IncomingMessage> messages = dmaap.getIncomingEvents(IncomingMessage.class, QUEUED_MAX);
+ LOG.debug(String.format("Read %d messages from dmaap", messages.size()));
+ for (IncomingMessage incoming : messages) {
+ // Acknowledge that we read the event
+ LOG.info("Acknowledging Message: " + incoming.getHeader().getRequestID());
+
+ //TODO: Should we post a pending status for 1607
+ //dmaap.postStatus(incoming.toOutgoing(Status.PENDING, null).toString());
+ }
+ for (IncomingMessage incoming : messages) {
+ // Add to pool if still running
+ if (run.get()) {
+ LOG.info(String.format("Adding DMaaP message to pool queue [%s]", incoming.getHeader().getRequestID()));
+ if (incoming.isValid()) {
+ try {
+ executor.execute(new WorkerImpl(incoming, dmaap));
+ } catch (RejectedExecutionException rejectEx) {
+ LOG.error("Task Rejected: ", rejectEx);
+ }
+ } else {
+ // Badly formed message
+ LOG.error("Message was not valid. Rejecting");
+ }
+ } else {
+ LOG.info("Run stopped. Orphaning Message: " + incoming.getHeader().getRequestID());
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOG.error("Exception " + e.getClass().getSimpleName() + " caught in DMaaP listener");
+ LOG.error(EELFResourceManager.format(e));
+ LOG.error("DMaaP Listener logging and ignoring the exception, continue...");
+ }
+ }
+
+ LOG.info("Stopping DMaaP Listener thread");
+
+ // We've told the listener to stop
+ // TODO - Should we:
+ // 1) Put a message back on the queue indicating that APP-C never got to
+ // the message
+ // or
+ // 2) Let downstream figure it out after timeout between PENDING and
+ // ACTIVE messages
+ }
+
+ @Override
+ public String getBenchmark() {
+ long time = System.currentTimeMillis();
+ DateFormat df = new SimpleDateFormat("HH:mm:ss");
+ df.setTimeZone(TimeZone.getTimeZone("UTC"));
+ String runningTime = df.format(new Date(time - startTime));
+
+ String out = String.format("Running for %s and completed %d jobs using %d threads.", runningTime,
+ executor.getCompletedTaskCount(), executor.getPoolSize());
+ LOG.info("***BENCHMARK*** " + out);
+ return out;
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ProviderOperations.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ProviderOperations.java
new file mode 100644
index 000000000..c300df154
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ProviderOperations.java
@@ -0,0 +1,291 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.impl;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.MalformedURLException;
+import java.net.Socket;
+import java.net.URL;
+import java.net.UnknownHostException;
+import java.security.KeyManagementException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.io.IOUtils;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpVersion;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.conn.ClientConnectionManager;
+import org.apache.http.conn.scheme.PlainSocketFactory;
+import org.apache.http.conn.scheme.Scheme;
+import org.apache.http.conn.scheme.SchemeRegistry;
+import org.apache.http.conn.ssl.SSLSocketFactory;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
+import org.apache.http.params.BasicHttpParams;
+import org.apache.http.params.HttpParams;
+import org.apache.http.params.HttpProtocolParams;
+import org.apache.http.protocol.HTTP;
+import org.json.JSONObject;
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.CL1607.model.IncomingMessage;
+import org.openecomp.appc.listener.util.Mapper;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+public class ProviderOperations {
+
+ private static final EELFLogger LOG = EELFManager.getInstance().getLogger(ProviderOperations.class);
+
+ private static URL url;
+
+ private static String basic_auth;
+
+ //@formatter:off
+ @SuppressWarnings("nls")
+ private final static String TEMPLATE = "{\"input\": {\"common-request-header\": {\"service-request-id\": \"%s\"},\"config-payload\": {\"config-url\": \"%s\",\"config-json\":\"%s\"}}}";
+ //@formatter:on
+
+ /**
+ * Calls the AppcProvider to run a topology directed graph
+ *
+ * @param msg
+ * The incoming message to be run
+ * @return True if the result is success. Never returns false and throws an exception instead.
+ * @throws UnsupportedEncodingException
+ * @throws Exception
+ * if there was a failure processing the request. The exception message is the failure reason.
+ */
+ @SuppressWarnings("nls")
+ public static boolean topologyDG(IncomingMessage msg) throws APPCException {
+ if (msg == null) {
+ throw new APPCException("Provided message was null");
+ }
+
+ HttpPost post = null;
+ try {
+ // Concatenate the "action" on the end of the URL
+ String path = url.getPath() + ":" + msg.getAction().getValue();
+ URL serviceUrl = new URL(url.getProtocol(), url.getHost(), url.getPort(), path);
+
+ post = new HttpPost(serviceUrl.toExternalForm());
+ post.setHeader("Content-Type", "application/json");
+ post.setHeader("Accept", "application/json");
+
+ // Set Auth
+ if (basic_auth != null) {
+ post.setHeader("Authorization", "Basic " + basic_auth);
+ }
+
+ //String body = buildReqest(msg.getId(), msg.getUrl(), msg.getIdentityUrl());
+ String body = buildReqest(msg.getHeader().getRequestID(), msg.getPayload().getGenericVnfId(), msg.getPayload().getPgStreams());
+ StringEntity entity = new StringEntity(body);
+ entity.setContentType("application/json");
+ post.setEntity(new StringEntity(body));
+ } catch (UnsupportedEncodingException | MalformedURLException e) {
+ throw new APPCException(e);
+ }
+
+ HttpClient client = getHttpClient();
+
+ int httpCode = 0;
+ String respBody = null;
+ try {
+ HttpResponse response = client.execute(post);
+ httpCode = response.getStatusLine().getStatusCode();
+ respBody = IOUtils.toString(response.getEntity().getContent());
+ } catch (IOException e) {
+ throw new APPCException(e);
+ }
+
+ if (httpCode == 200 && respBody != null) {
+ JSONObject json;
+ try {
+ json = Mapper.toJsonObject(respBody);
+ } catch (Exception e) {
+ LOG.error("Error prcoessing response from provider. Could not map response to json", e);
+ throw new APPCException("APPC has an unknown RPC error");
+ }
+ boolean success;
+ String reason;
+ try {
+ JSONObject header = json.getJSONObject("output").getJSONObject("common-response-header");
+ success = header.getBoolean("success");
+ reason = header.getString("reason");
+ } catch (Exception e) {
+ LOG.error("Unknown error prcoessing failed response from provider. Json not in expected format", e);
+ throw new APPCException("APPC has an unknown RPC error");
+ }
+ if (success) {
+ return true;
+ }
+ String reasonStr = reason == null ? "Unknown" : reason;
+ LOG.warn(String.format("Topology Operation [%s] failed. Reason: %s", msg.getHeader().getRequestID(), reasonStr));
+ throw new APPCException(reasonStr);
+
+ }
+ throw new APPCException(String.format("Unexpected response from endpoint: [%d] - %s ", httpCode, respBody));
+ }
+
+ /**
+ * Updates the static var URL and returns the value;
+ *
+ * @return The new value of URL
+ */
+ public static String getUrl() {
+ return url.toExternalForm();
+ }
+
+ public static void setUrl(String newUrl) {
+ try {
+ url = new URL(newUrl);
+ } catch (MalformedURLException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * Sets the basic authentication header for the given user and password. If either entry is null then set basic auth
+ * to null
+ *
+ * @param user
+ * The user with optional domain name
+ * @param password
+ * The password for the user
+ * @return The new value of the basic auth string that will be used in the request headers
+ */
+ public static String setAuthentication(String user, String password) {
+ if (user != null && password != null) {
+ String authStr = user + ":" + password;
+ basic_auth = new String(Base64.encodeBase64(authStr.getBytes()));
+ } else {
+ basic_auth = null;
+ }
+ return basic_auth;
+ }
+
+ /**
+ * Builds the request body for a topology operation
+ *
+ * @param id
+ * The request id
+ * @param url
+ * The vm's url
+ *
+ * @param pgstreams
+ * The streams to send to the traffic generator
+ *
+ * @return A String containing the request body
+ */
+ private static String buildReqest(String id, String url, String pgstreams) {
+
+ return String.format(TEMPLATE, id, url, pgstreams);
+ }
+
+ @SuppressWarnings("deprecation")
+ private static HttpClient getHttpClient() throws APPCException {
+ HttpClient client;
+ if (url.getProtocol().equals("https")) {
+ try {
+ KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
+ trustStore.load(null, null);
+ MySSLSocketFactory sf = new MySSLSocketFactory(trustStore);
+ sf.setHostnameVerifier(MySSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
+
+ HttpParams params = new BasicHttpParams();
+ HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1);
+ HttpProtocolParams.setContentCharset(params, HTTP.UTF_8);
+
+ SchemeRegistry registry = new SchemeRegistry();
+ registry.register(new Scheme("http", PlainSocketFactory.getSocketFactory(), 80));
+ registry.register(new Scheme("https", sf, 443));
+ registry.register(new Scheme("https", sf, 8443));
+ registry.register(new Scheme("http", sf, 8181));
+
+ ClientConnectionManager ccm = new ThreadSafeClientConnManager(params, registry);
+ client = new DefaultHttpClient(ccm, params);
+ } catch (Exception e) {
+ client = new DefaultHttpClient();
+ }
+ } else if (url.getProtocol().equals("http")) {
+ client = new DefaultHttpClient();
+ } else {
+ throw new APPCException(
+ "The provider.topology.url property is invalid. The url did not start with http[s]");
+ }
+ return client;
+ }
+
+ @SuppressWarnings("deprecation")
+ public static class MySSLSocketFactory extends SSLSocketFactory {
+ private SSLContext sslContext = SSLContext.getInstance("TLS");
+
+ public MySSLSocketFactory(KeyStore truststore) throws NoSuchAlgorithmException, KeyManagementException,
+ KeyStoreException, UnrecoverableKeyException {
+ super(truststore);
+
+ TrustManager tm = new X509TrustManager() {
+ @Override
+ public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+ }
+
+ @Override
+ public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+ }
+
+ @Override
+ public X509Certificate[] getAcceptedIssuers() {
+ return null;
+ }
+ };
+
+ sslContext.init(null, new TrustManager[] {
+ tm
+ }, null);
+ }
+
+ @Override
+ public Socket createSocket(Socket socket, String host, int port, boolean autoClose)
+ throws IOException, UnknownHostException {
+ return sslContext.getSocketFactory().createSocket(socket, host, port, autoClose);
+ }
+
+ @Override
+ public Socket createSocket() throws IOException {
+ return sslContext.getSocketFactory().createSocket();
+ }
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/WorkerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/WorkerImpl.java
new file mode 100644
index 000000000..52bcd2f5e
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/WorkerImpl.java
@@ -0,0 +1,82 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.impl;
+
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.EventHandler;
+import org.openecomp.appc.listener.CL1607.model.IncomingMessage;
+import org.openecomp.appc.listener.CL1607.model.Status;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+public class WorkerImpl implements Runnable {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(WorkerImpl.class);
+
+ // Should have all of the data we need for processing
+ private IncomingMessage event;
+
+ // So we can post messages from inside the worker.
+ private EventHandler dmaap;
+
+ public WorkerImpl(IncomingMessage message, EventHandler dmaap) {
+ this.event = message;
+ this.dmaap = dmaap;
+ }
+
+ @Override
+ public void run() {
+ LOG.debug(String.format("Started working on %s", event.getHeader().getRequestID()));
+
+ dmaap.postStatus(event.toOutgoing(Status.ACCEPTED));
+ // Run the dg in a try catch to handle all exceptions and update the
+ // message at the end
+ try {
+ if (doDG(event)) {
+ dmaap.postStatus(event.toOutgoing(Status.SUCCESS));
+ LOG.debug(String.format("Event %s finished successfully", event.getHeader().getRequestID()));
+ } else {
+ // Should never happen. Exception with message should be thrown instead.
+ LOG.error(String.format(
+ "We somehow returned false from doDG() instead of throwing exception. Incoming event [%s]",
+ event.toJson().toString()));
+ dmaap.postStatus(event.toOutgoing(Status.FAILURE));
+ }
+
+ } catch (Exception e) {
+ // Unknown exception from DG method. Fail and pass the exception
+ // along
+ String msg = "Exception: " + e.getMessage();
+ LOG.warn(String.format("Event %s finished with failure. %s", event.getHeader().getRequestID(), msg));
+ //TODO: should a message be included? there is nothing in the API spec for a msg?
+ //dmaap.postStatus(event.toOutgoing(Status.FAILURE, msg));
+ dmaap.postStatus(event.toOutgoing(Status.FAILURE));
+ }
+
+ LOG.debug("Done working on " + event.getHeader().getRequestID());
+ }
+
+ private boolean doDG(IncomingMessage msg) throws APPCException {
+ return ProviderOperations.topologyDG(msg);
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Action.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Action.java
new file mode 100644
index 000000000..9e7e923be
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Action.java
@@ -0,0 +1,55 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.model;
+
+public enum Action {
+ Restart("Restart"), Rebuild("Rebuild"), Migrate("Migrate"), Evacuate("Evacuate"), Snapshot("Snapshot"),modifyconfig("ModifyConfig");
+
+ /**
+ * Converts the string to an Action
+ *
+ * @param value
+ * The string to try and convert. Is case insensitive
+ * @return The action matching the string or null if no match was found.
+ */
+ public static Action toAction(String value) {
+ if (value != null) {
+ for (Action e : values()) {
+ if (e.getValue().toUpperCase().equals(value.toUpperCase())) {
+ return e;
+ }
+ }
+ }
+
+ return null;
+ }
+
+ private String value;
+
+ private Action(String valueToUse) {
+ value = valueToUse;
+ }
+
+ public final String getValue() {
+ return value;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/CommonMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/CommonMessage.java
new file mode 100644
index 000000000..80a78095b
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/CommonMessage.java
@@ -0,0 +1,340 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.model;
+
+import java.io.Serializable;
+import java.util.Collection;
+
+import org.json.JSONObject;
+import org.openecomp.appc.listener.util.Mapper;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+/**
+ * This class holds attributes that are common to DMaaP messages both coming in from DCAE and being sent out by APPC
+ *
+ */
+@JsonSerialize(include = Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CommonMessage implements Serializable {
+ /*
+ * { "CommonHeader": { "TimeStamp": "0000-00-00T00:00:00.000Z", "APIver": "1.01", "OriginatorID": "policy.pdp01",
+ * "RequestID": "b74d13c5-bb26-4b04-992c-4679dfc8280e", "SubrequestID": "1" }, "Action": "RESTART", "Payload": {
+ * "VServerSelfLink":
+ * "http://192.168.1.2:8774/v2/abcde12345fghijk6789lmnopq123rst/servers/abc12345-1234-5678-890a-abcdefg12345",
+ * "VNF_NAME": "test", "VMID": "abc12345-1234-5678-890a-abcdefg12345", "TenantID":
+ * "abcde12345fghijk6789lmnopq123rst", "LOC_ID": "Test", "in-maint": "false", "Identity":
+ * "http://example.com:5000/v2.0", "Prov_status": "ACTIVE", "OAM_IPV4": "192.168.1.2",
+ * "is-closed-loop-disabled": "false", "VM_NAME": "basx0001vm034", "OAM_IPV6": "aaaa::bbbb:cccc:dddd:eeee/64" } }
+ */
+
+ private static final long serialVersionUID = 1L;
+
+ /*
+ * The common header
+ */
+ @JsonProperty("CommonHeader")
+ private CommonHeader header;
+
+ /*
+ * The payload
+ */
+ @JsonProperty("Payload")
+ private Payload payload;
+
+ @JsonIgnore
+ private long startTime = System.currentTimeMillis();
+
+ /*
+ * Getters and Setters
+ */
+
+ public long getStartTime() {
+ return startTime;
+ }
+
+ public void setStartTime(long startTime) {
+ this.startTime = startTime;
+ }
+
+ /**
+ * @return the header
+ */
+ public CommonHeader getHeader() {
+ return header;
+ }
+
+ /**
+ * @param header
+ * the header to set
+ */
+ public void setHeader(CommonHeader header) {
+ this.header = header;
+ }
+
+ /**
+ * @return the payload
+ */
+ public Payload getPayload() {
+ return payload;
+ }
+
+ /**
+ * @param payload
+ * the payload to set
+ */
+ public void setPayload(Payload payload) {
+ this.payload = payload;
+ }
+
+ /**
+ * Convenience method to return a json representation of this object.
+ *
+ * @return The json representation of this object
+ */
+ public JSONObject toJson() {
+ return Mapper.toJsonObject(this);
+ }
+
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class CommonHeader {
+ /*
+ * "CommonHeader": { "TimeStamp": "2016-05-11T13:53:53.146Z", "APIver": "1.01", "OriginatorID": "policy.pdp01",
+ * "RequestID": "b74d13c5-bb26-4b04-992c-4679dfc8280e", "SubrequestID": "1" }
+ */
+
+ /*
+ * The timestamp of the message
+ */
+ @JsonProperty("TimeStamp")
+ private String timeStamp;
+
+ /*
+ * The API version of the message
+ */
+ @JsonProperty("APIver")
+ private String apiVer;
+
+ /*
+ * The Originator ID of the message
+ */
+ @JsonProperty("OriginatorID")
+ private String originatorId;
+
+ /*
+ * The Request Id of the message
+ */
+ @JsonProperty("RequestID")
+ private String requestID;
+
+ /*
+ * The Subrequest Id of the message
+ */
+ @JsonProperty("SubrequestID")
+ private String subRequestId;
+
+ /**
+ * @return the timeStamp
+ */
+ public String getTimeStamp() {
+ return timeStamp;
+ }
+
+ /**
+ * @param timeStamp
+ * the timeStamp to set
+ */
+ public void setTimeStamp(String timeStamp) {
+ this.timeStamp = timeStamp;
+ }
+
+ /**
+ * @return the apiVer
+ */
+ public String getApiVer() {
+ return apiVer;
+ }
+
+ /**
+ * @param apiVer
+ * the apiVer to set
+ */
+ public void setApiVer(String apiVer) {
+ this.apiVer = apiVer;
+ }
+
+ /**
+ * @return the originatorId
+ */
+ public String getOriginatorId() {
+ return originatorId;
+ }
+
+ /**
+ * @param originatorId
+ * the originatorId to set
+ */
+ public void setOriginatorId(String originatorId) {
+ this.originatorId = originatorId;
+ }
+
+ /**
+ * @return the requestID
+ */
+ public String getRequestID() {
+ return requestID;
+ }
+
+ /**
+ * @param requestID
+ * the requestID to set
+ */
+ public void setRequestID(String requestID) {
+ this.requestID = requestID;
+ }
+
+ /**
+ * @return the subRequestId
+ */
+ public String getSubRequestId() {
+ return subRequestId;
+ }
+
+ /**
+ * @param subRequestId
+ * the subRequestId to set
+ */
+ public void setSubRequestId(String subRequestId) {
+ this.subRequestId = subRequestId;
+ }
+ };
+
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class Payload {
+ /*
+ * "Payload": { "VServerSelfLink":
+ * "http://192.168.1.2:8774/v2/abcde12345fghijk6789lmnopq123rst/servers/abc12345-1234-5678-890a-abcdefg12345",
+ * "VNF_NAME": "test", "VMID": "abc12345-1234-5678-890a-abcdefg12345", "TenantID":
+ * "abcde12345fghijk6789lmnopq123rst", "LOC_ID": "Test", "in-maint": "false", "Identity":
+ * "http://example.com:5000/v2.0", "Prov_status": "ACTIVE", "OAM_IPV4": "192.168.1.2",
+ * "is-closed-loop-disabled": "false", "VM_NAME": "test", "OAM_IPV6": "aaaa::bbbb:cccc:dddd:eeee/64" }
+ */
+
+ /*
+ * The TenantID of the message
+ */
+ @JsonProperty("generic-vnf.vnf-id")
+ private String genericVnfId;
+
+ /**
+ * @return the TenantID
+ */
+ public String getGenericVnfId() {
+ return genericVnfId;
+ }
+
+ /**
+ * @param TenantID
+ * the TenantID to set
+ */
+ public void setGenericVnfId(String genericVnfId) {
+ this.genericVnfId = genericVnfId;
+ }
+
+ @JsonProperty("pg-streams")
+ private pgStreams pgStreams;
+
+ /**
+ * @return the TenantID
+ */
+
+ public String getPgStreams() {
+ String r = "{\\\"pg-streams\\\": {\\\"pg-stream\\\":[";
+ boolean first = true;
+ for(pgStream p : this.pgStreams.streams){
+ String n = "{\\\"id\\\":\\\""+p.getId()+"\\\", \\\"is-enabled\\\":\\\""+p.getIsEnabled()+"\\\"}";
+ if(!first){
+ r = r.concat(",");
+ }
+ first = false;
+ r = r.concat(n);
+ }
+ r= r.concat("]}}");
+ return r;
+ }
+
+ /**
+ * @param TenantID
+ * the TenantID to set
+ */
+ public void setPgStreams(pgStreams pgStreams) {
+ this.pgStreams = pgStreams;
+ }
+
+
+
+
+
+ };
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class pgStreams {
+ @JsonProperty("pg-stream")
+ private Collection<pgStream> streams;
+
+ public Collection<pgStream> getStreams() {
+ return streams;
+ }
+
+ public void setStreams(Collection<pgStream> streams) {
+ this.streams = streams;
+ }
+
+
+ };
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class pgStream{
+ public String getId() {
+ return id;
+ }
+ public void setId(String id) {
+ this.id = id;
+ }
+ public String getIsEnabled() {
+ return isEnabled;
+ }
+ public void setIsEnabled(String isEnabled) {
+ this.isEnabled = isEnabled;
+ }
+ @JsonProperty("id")
+ private String id;
+ @JsonProperty("is-enabled")
+ private String isEnabled;
+ };
+
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/IncomingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/IncomingMessage.java
new file mode 100644
index 000000000..62d4d1b7d
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/IncomingMessage.java
@@ -0,0 +1,88 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion;
+
+/**
+ * This class represents a message coming in from DCAE.
+ *
+ */
+@JsonSerialize(include = Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class IncomingMessage extends CommonMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ /*
+ * The action being requested. Its presence signals that it is an incoming message and it is not present on outgoing
+ * messages
+ */
+ //TODO; use enum
+ @JsonProperty("Action")
+ private String action;
+
+
+ public String getRequest() {
+ return action;
+ }
+
+ @JsonIgnore
+ public Action getAction() {
+ return Action.toAction(action);
+ }
+
+ public void setRequest(String request) {
+ this.action = request;
+ }
+
+// @Override
+// public String toString() {
+// String time = getRequestTime() != null ? getRequestTime() : "N/A";
+// // String req = request != null ? request : "N/A";
+// return String.format("[%s - %s]", time, getId());
+// }
+
+// public String toOutgoing(Status status) {
+// return toOutgoing(status);
+// }
+
+ public String toOutgoing(Status status) {
+ OutgoingMessage out = new OutgoingMessage(this);
+ out.setResponse(status);
+ return out.toResponse().toString();
+ }
+
+ /**
+ * Determines if this message should be parsed parsed. Will eventually check that the message is well formed, has
+ * all required fields, and had not exceeded any timing restrictions.
+ *
+ * @return True if the message should be parsed. False otherwise
+ */
+ public boolean isValid() {
+ return true;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/OutgoingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/OutgoingMessage.java
new file mode 100644
index 000000000..7b766212d
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/OutgoingMessage.java
@@ -0,0 +1,192 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.model;
+
+import java.net.InetAddress;
+import java.security.SecureRandom;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.TimeZone;
+
+import org.json.JSONObject;
+import org.openecomp.appc.listener.util.Mapper;
+import org.openecomp.appc.util.Time;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion;
+
+/**
+ * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request
+ *
+ */
+@JsonSerialize(include = Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class OutgoingMessage extends CommonMessage {
+
+ public OutgoingMessage() {
+
+ }
+
+ public OutgoingMessage(IncomingMessage msg) {
+ setHeader(msg.getHeader());
+ setPayload(msg.getPayload());
+// setId(msg.getId());
+// setOriginalRequest(msg.getRequest());
+// setRequestClient(msg.getRequestClient());
+// setRequestTime(msg.getRequestTime());
+// setVmName(msg.getVmName());
+// setFromSystem(generateFrom());
+// setResponse(Status.PENDING);
+// setPolicyName(msg.getPolicyName());
+// setPolicyVersion(msg.getPolicyVersion());
+// setStartTime(msg.getStartTime());
+ }
+
+ private static final long serialVersionUID = -5447940920271469613L;
+ /*
+ * The status of the response
+ */
+ @JsonProperty("Status")
+ private OutStatus status;
+
+ /**
+ * @return the status
+ */
+ public OutStatus getStatus() {
+ return status;
+ }
+
+ /**
+ * @param status the status to set
+ */
+ public void setStatus(OutStatus status) {
+ this.status = status;
+ }
+
+ public void updateResponseTime() {
+ SecureRandom rand = new SecureRandom();
+ SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd kk:mm:ss.SSS");
+ df.setTimeZone(TimeZone.getTimeZone("UTC"));
+ String date = df.format(new Date(Time.utcTime()));
+ //this.responseTime = String.format("%s%03d", date, rand.nextInt(1000));
+ }
+
+ public String generateFrom() {
+ String name;
+ try {
+ InetAddress iAddress = InetAddress.getLocalHost();
+ name = iAddress.getCanonicalHostName();
+ } catch (Exception e) {
+ // Could not get anything from the InetAddress
+ name = "UnknownHost";
+ }
+ return "appc@" + name;
+ }
+
+ public JSONObject toResponse() {
+ updateResponseTime();
+ JSONObject json = Mapper.toJsonObject(this);
+
+ if (!json.has("message")) {
+ // If there is no message, parrot the status (response field)
+ // TODO - Can this be removed for 1602 making message truely optional?
+ //json.put("message", this.getResponse().toString());
+ }
+
+ // Removed duplication of status from message for 1602
+ // json.put("message", String.format("%s: %s", request, json.get("message")));
+
+ return json;
+ }
+
+// @Override
+// public String toString() {
+// return String.format("%s - %s", getId(), getResponse());
+// }
+
+ public static class OutStatus{
+ @JsonProperty("Code")
+ private String code;
+
+ @JsonProperty("Value")
+ private String value;
+
+ /**
+ * @return the code
+ */
+ public String getCode() {
+ return code;
+ }
+
+ /**
+ * @param code the code to set
+ */
+ public void setCode(String code) {
+ this.code = code;
+ }
+
+ /**
+ * @return the value
+ */
+ public String getValue() {
+ return value;
+ }
+
+ /**
+ * @param value the value to set
+ */
+ public void setValue(String value) {
+ this.value = value;
+ }
+
+ }
+
+ public void setResponse(Status newStatus) {
+ if(this.status == null){
+ this.status = new OutStatus();
+ }
+
+ switch (newStatus){
+ case ACCEPTED:
+ this.status.setValue(newStatus.getValue());
+ this.status.setCode("100");
+ break;
+
+ case FAILURE:
+ this.status.setValue(newStatus.getValue());
+ this.status.setCode("500");
+ break;
+
+ case SUCCESS:
+ this.status.setValue(newStatus.getValue());
+ this.status.setCode("400");
+ break;
+ default:
+ break;
+
+ }
+
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Status.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Status.java
new file mode 100644
index 000000000..f7ec8febd
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Status.java
@@ -0,0 +1,69 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.model;
+
+public enum Status {
+ /*
+ * APP-C acknowledges that it has read the event off of the wire. This is the initial status of an OutgoingEvent
+ */
+ ACCEPTED("ACCEPTED"),
+
+ /*
+ * APP-C has finished processing the event without errors
+ */
+ SUCCESS("SUCCESS"),
+
+ /*
+ * APP-C has finished processing the event with errors
+ */
+ FAILURE("FAILURE");
+
+ /**
+ * Converts the string to an Status
+ *
+ * @param value
+ * The string to try and convert. Is case insensitive
+ * @return The status matching the string or null if no match was found.
+ */
+ public static Status toStatus(String value) {
+ if (value != null) {
+ for (Status e : values()) {
+ if (e.getValue().toUpperCase().equals(value.toUpperCase())) {
+ return e;
+ }
+ }
+ }
+
+ return null;
+ }
+
+ private String value;
+
+ private Status(String valueToUse) {
+ value = valueToUse;
+ }
+
+ public final String getValue() {
+ return value;
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Controller.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Controller.java
new file mode 100644
index 000000000..66793e248
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Controller.java
@@ -0,0 +1,55 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+import java.util.Map;
+
+/**
+ * A controller is responsible for creating a listener for each ListenerProperties object that is passed in to it on
+ * instantiation. The controller will create a thread pool that will contain all of the listener threads so no listener
+ * can crash the controller.
+ *
+ */
+public interface Controller {
+
+ /**
+ * Creates a new thread in the thread pool for an implementation of the {@see #Listener} class set in the
+ * ListenerProperties. This thread is run immediately after it is created.
+ */
+ public void start();
+
+ /**
+ * Stops each of the listeners known by this controller. Takes an optional parameter that indicates the the listener
+ * should stop immediately rather than waiting for all threads to complete.
+ *
+ * @param stopNow
+ * Determines what method the listeners should use to shutdown. If true, listeners will use the stopNow()
+ * method. Otherwise they will use the stop() method.
+ */
+ public void stop(boolean stopNow);
+
+ /**
+ * @return A Map of ListenerProperties and the Listener object that is running in the controllers thread pool.
+ */
+ public Map<ListenerProperties, Listener> getListeners();
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/EventHandler.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/EventHandler.java
new file mode 100644
index 000000000..5b2906f3c
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/EventHandler.java
@@ -0,0 +1,185 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * EventHandler defines a class that wraps DMaaP operations (most notably Get Message and Post Message) to make them
+ * easier to use.
+ *
+ */
+public interface EventHandler {
+
+ /**
+ * Gets a list of messages as Strings on the read topic.
+ *
+ * @return A list of String messages. Never returns null.
+ */
+ public List<String> getIncomingEvents();
+
+ /**
+ * Gets a list of messages as String on the read topic.
+ *
+ * @param limit
+ * The maximum amount of entries to return
+ * @return A list of String messages. Never returns null.
+ */
+ public List<String> getIncomingEvents(int limit);
+
+ /**
+ * Gets a list of messages Mapped to the given Class. If a message cannot be mapped to that class, it is discarded.
+ *
+ * @param cls
+ * The class to map the message to.
+ * @return A list of objects of the provided class. Never returns null.
+ */
+ public <T> List<T> getIncomingEvents(Class<T> cls);
+
+ /**
+ * Gets a list of messages Mapped to the given Class. If a message cannot be mapped to that class, it is discarded.
+ *
+ * @param cls
+ * The class to map the message to.
+ * @param limit
+ * The maximum amount of entries to return
+ * @return A list of objects of the provided class. Never returns null.
+ */
+ public <T> List<T> getIncomingEvents(Class<T> cls, int limit);
+
+ /**
+ * Posts the String message to the write topic(s).
+ *
+ * @param event
+ * The String to post.
+ */
+ public void postStatus(String event);
+
+ /**
+ * Posts the String message to the write topic(s) on the specified partition. Partitions are only used to guarantee
+ * ordering and do not impact if data is retreived.
+ *
+ * @param partition
+ * The partition to post to or null if no partition should be used.
+ * @param event
+ * The String to post.
+ */
+ public void postStatus(String partition, String event);
+
+ /**
+ * @return The client/group id used to read messages
+ */
+ public String getClientId();
+
+ /**
+ * Set the client/group id used to read messages
+ *
+ * @param clientId
+ * The new clientId to use
+ */
+ public void setClientId(String clientId);
+
+ /**
+ * @return The client/group name to use.
+ */
+ public String getClientName();
+
+ /**
+ * Set the client/group name used to read messages.
+ *
+ * @param clientName
+ * The new clientName to use
+ */
+ public void setClientName(String clientName);
+
+ /**
+ * @return The name of the topic to read from
+ */
+ public String getReadTopic();
+
+ /**
+ * Set the name of the topic to read from.
+ *
+ * @param topic
+ * The new topic to read from
+ */
+ public void setReadTopic(String topic);
+
+ /**
+ * @return The name of the topic to write to
+ */
+ public Set<String> getWriteTopics();
+
+ /**
+ * Set the name of the topic to write to
+ *
+ * @param topic
+ * The new topic to write to
+ */
+ public void setWriteTopics(Set<String> topic);
+
+ /**
+ * Adds a DMaaP host to the host pool
+ *
+ * @param host
+ * The host to add to the pool in &lt;host&gt;:&lt;port&gt; format
+ */
+ public void addToPool(String host);
+
+ /**
+ * Remove the host name from the pool if it exists
+ *
+ * @param host
+ * The host to add to the pool in &lt;host&gt;:&lt;port&gt; format
+ */
+ public void removeFromPool(String host);
+
+ /**
+ * Get all of the hosts in the DMaaP pool
+ *
+ * @return A collection of host in &lt;host&gt;:&lt;port&gt; format
+ */
+ public Collection<String> getPool();
+
+ /**
+ * Clear any provided api credentials and make future requests as an unauthenticated user
+ */
+ public void clearCredentials();
+
+ /**
+ * Set the api credentials and make future requests as an authenticated user
+ *
+ * @param access
+ * The access portion of the credentials (either user name or api key)
+ * @param secret
+ * The secret portion of the credentials (either password or api secret)
+ */
+ public void setCredentials(String access, String secret);
+
+ /**
+ * Close consumer/producer DMaaP clients
+ */
+ public void closeClients();
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/conv/Converter.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/conv/Converter.java
new file mode 100644
index 000000000..021dfefc8
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/conv/Converter.java
@@ -0,0 +1,85 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.conv;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.*;
+import org.apache.commons.lang3.StringUtils;
+import org.json.JSONObject;
+import org.openecomp.appc.listener.LCM.model.DmaapMessage;
+import org.openecomp.appc.listener.LCM.model.DmaapOutgoingMessage;
+import org.openecomp.appc.listener.util.Mapper;
+
+
+public class Converter {
+
+
+ public static DmaapOutgoingMessage convJsonNodeToDmaapOutgoingMessage(JsonNode inObj, String rpcName) {
+ DmaapOutgoingMessage outObj = new DmaapOutgoingMessage();
+ outObj.setBody(inObj);
+ outObj.setRpcName(rpcName);
+ return outObj;
+ }
+
+ public static String convDmaapOutgoingMessageToJsonString(DmaapMessage inObj) throws JsonProcessingException {
+// return Mapper.toJsonString(inObj);
+ ObjectMapper objectMapper = new ObjectMapper();
+ ObjectWriter writer = objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL).configure(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY,true)
+ .writer().withFeatures(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS);
+ return writer.writeValueAsString(inObj);
+
+ }
+
+ public static DmaapOutgoingMessage buildDmaapOutgoingMessageWithUnexpectedError(JsonNode dmaapInputBody, String rpcName,Exception inputException) {
+ DmaapOutgoingMessage dmaapOutgoingMessage = null;
+ String errMsg = StringUtils.isEmpty(inputException.getMessage())? inputException.toString() : inputException.getMessage();
+ JSONObject commonHeaderJsonObject = Mapper.toJsonObject(dmaapInputBody.get("input").get("common-header"));
+ JSONObject jsonObjectOutput = new JSONObject().accumulate("common-header", commonHeaderJsonObject).accumulate("status", new JSONObject().accumulate("code",200).accumulate("value",errMsg));
+ dmaapOutgoingMessage = new DmaapOutgoingMessage();
+ dmaapOutgoingMessage.setRpcName(rpcName);
+ JSONObject jsonObjectBody = new JSONObject().accumulate("output",jsonObjectOutput);
+ JsonNode jsonNodeBody = Mapper.toJsonNodeFromJsonString(jsonObjectBody.toString());
+ dmaapOutgoingMessage.setBody(jsonNodeBody);
+ return dmaapOutgoingMessage;
+ }
+
+ public static String extractRequestIdWithSubId(JsonNode dmaapBody) {
+ String requestId;
+ //TODO: null pointer exception if dmaapBody is null, check if null or ensure is not null before calling
+ JsonNode commonHeaderJsonNode = dmaapBody.get("input").get("common-header");
+ requestId = commonHeaderJsonNode.get("request-id").asText();
+ requestId = requestId != null ? requestId : "";
+ String subRequestId = commonHeaderJsonNode.get("sub-request-id").asText();
+ if(!StringUtils.isEmpty(subRequestId)){
+ requestId = requestId +"-"+subRequestId;
+ }
+ return requestId;
+ }
+
+ public static Integer extractStatusCode(JsonNode event) {
+ Integer statusCode;
+ statusCode = event.get("output").get("status").get("code").asInt();
+ return statusCode;
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/ListenerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/ListenerImpl.java
new file mode 100644
index 000000000..dd877e98a
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/ListenerImpl.java
@@ -0,0 +1,159 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.impl;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.commons.lang3.StringUtils;
+import org.openecomp.appc.listener.AbstractListener;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.LCM.conv.Converter;
+import org.openecomp.appc.listener.LCM.model.DmaapIncomingMessage;
+import org.openecomp.appc.listener.LCM.operation.ProviderOperations;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import com.att.eelf.i18n.EELFResourceManager;
+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.concurrent.RejectedExecutionException;
+
+public class ListenerImpl extends AbstractListener {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(ListenerImpl.class);
+
+ private long startTime = 0;
+
+ public ListenerImpl(ListenerProperties props) {
+ super(props);
+
+ String url = props.getProperty("provider.url");
+ LOG.info("DMaaP Provider Endpoint: " + url);
+ ProviderOperations.setUrl(url);
+
+ // Set Basic Auth
+ String user = props.getProperty("provider.user");
+ String pass = props.getProperty("provider.pass");
+ ProviderOperations.setAuthentication(user, pass);
+ }
+
+ @Override
+ public void run() {
+ // Some vars for benchmarking
+ startTime = System.currentTimeMillis();
+
+ LOG.info("Running DMaaP Listener");
+
+ while (run.get()) {
+ // Only update if the queue is low. otherwise we read in more
+ // messages than we need
+ try {
+ if (executor.getQueue().size() <= QUEUED_MIN) {
+ LOG.debug("DMaaP queue running low. Querying for more jobs");
+
+
+ List<DmaapIncomingMessage> messages = dmaap.getIncomingEvents(DmaapIncomingMessage.class, QUEUED_MAX);
+ LOG.debug(String.format("Read %d messages from dmaap", messages.size()));
+ for (DmaapIncomingMessage incoming : messages) {
+ // Acknowledge that we read the event
+ if (isValid(incoming)) {
+ String requestIdWithSubId = getRequestIdWithSubId(incoming.getBody());
+ LOG.info("Acknowledging Message: " + requestIdWithSubId);
+// dmaap.postStatus(incoming.toOutgoing(OperationStatus.PENDING));
+ }
+ }
+ for (DmaapIncomingMessage incoming : messages) {
+ String requestIdWithSubId = getRequestIdWithSubId(incoming.getBody());
+ // Add to pool if still running
+ if (run.get()) {
+ if (isValid(incoming)) {
+ LOG.info(String.format("Adding DMaaP message to pool queue [%s]", requestIdWithSubId));
+ try {
+ executor.execute(new WorkerImpl(incoming.getRpcName(),incoming.getBody(), dmaap));
+ } catch (RejectedExecutionException rejectEx) {
+ LOG.error("Task Rejected: ", rejectEx);
+ }
+ } else {
+ // Badly formed message
+ LOG.error("Message was not valid. Rejecting message: "+incoming);
+ }
+ } else {
+ if (isValid(incoming)) {
+ LOG.info("Run stopped. Orphaning Message: " + requestIdWithSubId);
+ }
+ else {
+ // Badly formed message
+ LOG.error("Message was not valid. Rejecting message: "+incoming);
+ }
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOG.error("Exception " + e.getClass().getSimpleName() + " caught in DMaaP listener");
+ LOG.error(EELFResourceManager.format(e));
+ LOG.error("DMaaP Listener logging and ignoring the exception, continue...");
+ }
+ }
+
+ LOG.info("Stopping DMaaP Listener thread");
+
+ // We've told the listener to stop
+ // TODO - Should we:
+ // 1) Put a message back on the queue indicating that APP-C never got to
+ // the message
+ // or
+ // 2) Let downstream figure it out after timeout between PENDING and
+ // ACTIVE messages
+ }
+
+ private boolean isValid(DmaapIncomingMessage incoming) {
+ return ((incoming != null) &&
+ incoming.getBody() != null
+ && !StringUtils.isEmpty(incoming.getRpcName()));
+ }
+
+ @Override
+ public String getBenchmark() {
+ long time = System.currentTimeMillis();
+ DateFormat df = new SimpleDateFormat("HH:mm:ss");
+ df.setTimeZone(TimeZone.getTimeZone("UTC"));
+ String runningTime = df.format(new Date(time - startTime));
+
+ String out = String.format("Running for %s and completed %d jobs using %d threads.", runningTime,
+ executor.getCompletedTaskCount(), executor.getPoolSize());
+ LOG.info("***BENCHMARK*** " + out);
+ return out;
+ }
+
+ private String getRequestIdWithSubId(JsonNode event){
+ String requestId = "";
+ try {
+ requestId = Converter.extractRequestIdWithSubId(event);
+ } catch (Exception e) {
+ LOG.error("failed to parse request-id and sub-request-id. Json not in expected format", e);
+ }
+ return requestId;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/WorkerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/WorkerImpl.java
new file mode 100644
index 000000000..af8334ee4
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/WorkerImpl.java
@@ -0,0 +1,120 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.impl;
+
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.EventHandler;
+import org.openecomp.appc.listener.LCM.conv.Converter;
+import org.openecomp.appc.listener.LCM.model.DmaapOutgoingMessage;
+import org.openecomp.appc.listener.LCM.operation.ProviderOperations;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+
+public class WorkerImpl implements Runnable {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(WorkerImpl.class);
+
+ // Should have all of the data we need for processing
+ private JsonNode event;
+
+ // So we can post messages from inside the worker.
+ private EventHandler dmaap;
+ private String rpcName;
+
+ public WorkerImpl(String rpcName,JsonNode message, EventHandler dmaap) {
+ this.rpcName = rpcName;
+ this.event = message;
+ this.dmaap = dmaap;
+ }
+
+ @Override
+ public void run() {
+ String requestIdWithSubId = extractRequestIdWithSubId(event);
+ LOG.debug(String.format("Started working on %s", requestIdWithSubId));
+
+ // Run the dg in a try catch to handle all exceptions and update the
+ // message at the end
+ try {
+ JsonNode outputJsonNode = doDG(rpcName, event);
+ DmaapOutgoingMessage dmaapOutgoingMessage= Converter.convJsonNodeToDmaapOutgoingMessage(outputJsonNode,rpcName);
+ postMessageToDMaaP(dmaapOutgoingMessage,requestIdWithSubId);
+ Integer statusCode = extractStatusCode(dmaapOutgoingMessage.getBody());
+ if (ProviderOperations.isSucceeded(statusCode)) {
+ LOG.debug(String.format("Event %s finished successfully", requestIdWithSubId));
+ } else {
+ LOG.warn(String.format("Event %s failed", requestIdWithSubId));
+ }
+
+ } catch (Exception e) {
+ // Unknown exception from DG method. Fail and pass the exception
+ // along
+ String msg = "Exception: " + e.getMessage();
+ LOG.error(String.format("Event %s finished with failure. %s", requestIdWithSubId, msg));
+ DmaapOutgoingMessage dmaapOutgoingMessage= Converter.buildDmaapOutgoingMessageWithUnexpectedError(event,rpcName,e);
+ postMessageToDMaaP(dmaapOutgoingMessage,requestIdWithSubId);
+ }
+
+ LOG.debug("Done working on " + requestIdWithSubId);
+ }
+
+
+ private Integer extractStatusCode(JsonNode event) {
+ Integer statusCode = null;
+ try {
+ statusCode = Converter.extractStatusCode(event);
+ } catch (Exception e) {
+ LOG.error("failed to parse statusCode. Json not in expected format", e);
+ }
+ return statusCode;
+ }
+
+
+ private String extractRequestIdWithSubId(JsonNode event){
+ String requestId = "";
+ try {
+ requestId = Converter.extractRequestIdWithSubId(event);
+ } catch (Exception e) {
+ LOG.error("failed to parse request-id and sub-request-id. Json not in expected format", e);
+ }
+ return requestId;
+ }
+
+
+
+ private void postMessageToDMaaP(DmaapOutgoingMessage dmaapOutgoingMessage,String requestIdWithSubId) {
+ String dmaapOutgoingMessageJsonString;
+ try {
+ dmaapOutgoingMessageJsonString = Converter.convDmaapOutgoingMessageToJsonString(dmaapOutgoingMessage);
+ dmaap.postStatus(dmaapOutgoingMessage.getCambriaPartition(),dmaapOutgoingMessageJsonString);
+ } catch (JsonProcessingException e) {
+ LOG.error("failed to postMessageToDMaaP requestIdWithSubId: "+requestIdWithSubId+" dmaapOutgoingMessage: "+dmaapOutgoingMessage, e);
+ }
+ }
+
+ private JsonNode doDG(String rpcName, JsonNode msg) throws APPCException {
+ return ProviderOperations.topologyDG(rpcName,msg);
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ActionIdentifiers.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ActionIdentifiers.java
new file mode 100644
index 000000000..f17b18147
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ActionIdentifiers.java
@@ -0,0 +1,81 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+
+
+public class ActionIdentifiers implements Serializable {
+
+ @JsonProperty("service-instance-id")
+ private String serviceInstanceId;
+ @JsonProperty("vnf-id")
+ private String vnfID;
+ @JsonProperty("vnfc-name")
+ private String vnfcName;
+ @JsonProperty("vserver-id")
+ private String vserverId;
+
+ public ActionIdentifiers() {
+ }
+
+ public ActionIdentifiers(ActionIdentifiers actionIdentifiers) {
+ this.serviceInstanceId=actionIdentifiers.getServiceInstanceId();
+ this.vnfID=actionIdentifiers.getVnfID();
+ this.vnfcName=actionIdentifiers.getVnfcName();
+ this.vserverId=actionIdentifiers.getVserverId();
+ }
+
+ public String getServiceInstanceId() {
+ return serviceInstanceId;
+ }
+
+ public void setServiceInstanceId(String serviceInstanceId) {
+ this.serviceInstanceId = serviceInstanceId;
+ }
+
+ public String getVnfID() {
+ return vnfID;
+ }
+
+ public void setVnfID(String vnfID) {
+ this.vnfID = vnfID;
+ }
+
+ public String getVnfcName() {
+ return vnfcName;
+ }
+
+ public void setVnfcName(String vnfcName) {
+ this.vnfcName = vnfcName;
+ }
+
+ public String getVserverId() {
+ return vserverId;
+ }
+
+ public void setVserverId(String vserverId) {
+ this.vserverId = vserverId;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/CommonHeader.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/CommonHeader.java
new file mode 100644
index 000000000..363c6407d
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/CommonHeader.java
@@ -0,0 +1,112 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Map;
+
+import org.openecomp.appc.util.Time;
+
+public class CommonHeader implements Serializable {
+ @JsonProperty("timestamp")
+ private String timeStamp;
+ @JsonProperty("api-ver")
+ private String apiVer;
+ @JsonProperty("originator-id")
+ private String originatorId;
+ @JsonProperty("request-id")
+ private String requestID;
+ @JsonProperty("sub-request-id")
+ private String subRequestId;
+ @JsonProperty("flags")
+ private Map<String, String> flags;
+
+ private static final DateFormat ZULU_FORMATTER = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SS'Z'");
+
+ public CommonHeader() {
+ }
+
+ public CommonHeader(CommonHeader commonHeader) {
+ // changed to current system time
+ timeStamp = ZULU_FORMATTER.format(Time.utcDate());
+
+ apiVer = commonHeader.getApiVer();
+ originatorId = commonHeader.getOriginatorId();
+ requestID = commonHeader.getRequestID();
+ subRequestId = commonHeader.getSubRequestId();
+ flags = commonHeader.getFlags();
+ }
+
+
+
+ public String getTimeStamp() {
+ return timeStamp;
+ }
+
+ public void setTimeStamp(String timeStamp) {
+ this.timeStamp = timeStamp;
+ }
+
+ public String getApiVer() {
+ return apiVer;
+ }
+
+ public void setApiVer(String apiVer) {
+ this.apiVer = apiVer;
+ }
+
+ public String getRequestID() {
+ return requestID;
+ }
+
+ public void setRequestID(String requestID) {
+ this.requestID = requestID;
+ }
+
+ public String getOriginatorId() {
+ return originatorId;
+ }
+
+ public void setOriginatorId(String originatorId) {
+ this.originatorId = originatorId;
+ }
+
+ public String getSubRequestId() {
+ return subRequestId;
+ }
+
+ public void setSubRequestId(String subRequestId) {
+ this.subRequestId = subRequestId;
+ }
+
+ public Map<String, String> getFlags() {
+ return flags;
+ }
+
+ public void setFlags(Map<String, String> flags) {
+ this.flags = flags;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapIncomingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapIncomingMessage.java
new file mode 100644
index 000000000..ec1e9e767
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapIncomingMessage.java
@@ -0,0 +1,50 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import org.json.JSONObject;
+import org.openecomp.appc.listener.util.Mapper;
+
+/**
+ * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request
+ *
+ */
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DmaapIncomingMessage extends DmaapMessage{
+ private final static String defaultCambriaPartition = "APP-C";
+
+ public DmaapIncomingMessage() {
+ super();
+ setCambriaPartition(defaultCambriaPartition);
+ }
+
+
+ @Override
+ public String toString() {
+ return "DmaapIncomingMessage{"+super.toString()+"}";
+ }
+}
+
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapMessage.java
new file mode 100644
index 000000000..56099cb7e
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapMessage.java
@@ -0,0 +1,79 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.JsonNode;
+
+
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DmaapMessage{
+
+ @JsonProperty("cambria.partition")
+ private String cambriaPartition;
+
+ @JsonProperty("rpc-name")
+ private String rpcName;
+
+ @JsonProperty("body")
+ private JsonNode body;
+
+ public DmaapMessage() {
+ }
+
+ public String getCambriaPartition() {
+ return cambriaPartition;
+ }
+
+ public void setCambriaPartition(String cambriaPartition) {
+ this.cambriaPartition = cambriaPartition;
+ }
+
+ public String getRpcName() {
+ return rpcName;
+ }
+
+ public void setRpcName(String rpcName) {
+ this.rpcName = rpcName;
+ }
+
+ public JsonNode getBody() {
+ return body;
+ }
+
+ public void setBody(JsonNode body) {
+ this.body = body;
+ }
+
+ @Override
+ public String toString() {
+ return "DmaapMessage{" +
+ "cambriaPartition='" + cambriaPartition + '\'' +
+ ", rpcName='" + rpcName + '\'' +
+ ", body=" + body +
+ '}';
+ }
+}
+
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapOutgoingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapOutgoingMessage.java
new file mode 100644
index 000000000..6d7cb997e
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapOutgoingMessage.java
@@ -0,0 +1,50 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import org.json.JSONObject;
+import org.openecomp.appc.listener.util.Mapper;
+
+/**
+ * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request
+ *
+ */
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DmaapOutgoingMessage extends DmaapMessage{
+ private final static String defaultCambriaPartition = "MSO";
+
+ public DmaapOutgoingMessage() {
+ super();
+ setCambriaPartition(defaultCambriaPartition);
+ }
+
+
+ @Override
+ public String toString() {
+ return "DmaapOutgoingMessage{"+super.toString()+"}";
+ }
+}
+
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/InputBody.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/InputBody.java
new file mode 100644
index 000000000..6fd0572f3
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/InputBody.java
@@ -0,0 +1,102 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.model;
+
+import org.openecomp.appc.listener.LCM.operation.OperationStatus;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class InputBody {
+
+ @JsonProperty("common-header")
+ private CommonHeader commonHeader;
+
+ @JsonProperty("action-identifiers")
+ private ActionIdentifiers actionIdentifiers;
+
+ @JsonProperty("action")
+ private String action;
+
+ @JsonProperty("payload")
+ private Object payload;
+
+
+ public Object getPayload() {
+ return payload;
+ }
+
+ public void setPayload(Object payload) {
+ this.payload = payload;
+ }
+ @JsonIgnore
+ public void setPayloadAsString(String payload) {
+ this.payload = payload;
+ }
+
+ public String getAction() {
+ return action;
+ }
+
+ public void setAction(String action) {
+ this.action = action;
+ }
+
+ public ActionIdentifiers getActionIdentifiers() {
+ return actionIdentifiers;
+ }
+
+ public void setActionIdentifiers(ActionIdentifiers actionIdentifiers) {
+ this.actionIdentifiers = actionIdentifiers;
+ }
+
+ public CommonHeader getCommonHeader() {
+ return commonHeader;
+ }
+
+ public void setCommonHeader(CommonHeader commonHeader) {
+ this.commonHeader = commonHeader;
+ }
+
+// public String toOutgoing(OperationStatus operationStatus) {
+// OutputBody out = new OutputBody(this);
+// out.setStatus(new ResponseStatus(operationStatus.getCode(), operationStatus.getValue()));
+// return out.toResponse().toString();
+// }
+
+// public String toOutgoing(OperationStatus operationStatus,String islocked) {
+// OutputBody out = new OutputBody(this);
+// out.setStatus(new ResponseStatus(operationStatus.getCode(), operationStatus.getValue()));
+// out.setLocked(islocked);
+// return out.toResponse().toString();
+// }
+
+ @JsonIgnore
+ public boolean isValid() {
+ return getCommonHeader() != null;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/OutputBody.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/OutputBody.java
new file mode 100644
index 000000000..54b407b5f
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/OutputBody.java
@@ -0,0 +1,98 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import org.json.JSONObject;
+import org.openecomp.appc.listener.util.Mapper;
+
+/**
+ * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request
+ *
+ */
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class OutputBody {
+
+ @JsonProperty("common-header")
+ private CommonHeader header;
+
+ @JsonProperty("status")
+ private ResponseStatus status;
+
+ @JsonProperty("payload")
+ private String payload;
+
+ @JsonProperty("locked")
+ private String locked;
+
+ public OutputBody() {
+ }
+
+
+ public JSONObject toResponse() {
+ return Mapper.toJsonObject(this);
+ }
+
+ public OutputBody(InputBody msg) {
+ this.header = new CommonHeader(msg.getCommonHeader());
+ }
+
+ public String getLocked() {
+ return locked;
+ }
+
+ public void setLocked(String locked) {
+ this.locked = locked;
+ }
+ public String getPayload() {
+ return payload;
+ }
+
+ public void setPayload(String payload) {
+ this.payload = payload;
+ }
+
+ public CommonHeader getHeader() {
+ return header;
+ }
+
+ public void setHeader(CommonHeader header) {
+ this.header = header;
+ }
+
+ public ResponseStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(ResponseStatus status) {
+ this.status = status;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s - %s(%s)", getHeader().getRequestID(), getStatus().getCode(), getStatus().getValue());
+ }
+}
+
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ResponseStatus.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ResponseStatus.java
new file mode 100644
index 000000000..9186c444a
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ResponseStatus.java
@@ -0,0 +1,70 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ResponseStatus {
+ @JsonProperty("code")
+ private Integer code;
+ @JsonProperty("message")
+ private String value;
+
+ public ResponseStatus() {
+ }
+
+ public ResponseStatus(Integer code, String value) {
+ this.code = code;
+ this.value = value;
+ }
+
+ public Integer getCode() {
+ return code;
+ }
+
+ public void setCode(Integer code) {
+ code = code;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+
+
+ public void setValue(String value) {
+ value = value;
+ }
+
+ @Override
+ public String toString() {
+ return "ResponseStatus{" +
+ "code=" + code +
+ ", value='" + value + '\'' +
+ '}';
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/GenericProviderOperationRequestFormatter.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/GenericProviderOperationRequestFormatter.java
new file mode 100644
index 000000000..5c54845cc
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/GenericProviderOperationRequestFormatter.java
@@ -0,0 +1,115 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.operation;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.json.JSONObject;
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.LCM.model.InputBody;
+import org.openecomp.appc.listener.LCM.model.ResponseStatus;
+import org.openecomp.appc.listener.util.Mapper;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+import java.net.URL;
+
+
+
+public class GenericProviderOperationRequestFormatter implements ProviderOperationRequestFormatter {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(GenericProviderOperationRequestFormatter.class);
+
+ //@formatter:off
+ @SuppressWarnings("nls")
+ private final static String TEMPLATE = "{\"input\": %s}";
+ //@formatter:on
+
+ @Override
+ public String buildPath(URL url, String rpcName) {
+ return url.getPath() + ":"+rpcName;
+ }
+
+// private String convertActionToUrl(String action) {
+// String res = "";
+// switch (action) {
+// case "SoftwareUpload":
+// res = "software-upload";
+// break;
+// case "LiveUpgrade":
+// res = "live-upgrade";
+// break;
+// case "HealthCheck":
+// res = "health-check";
+// break;
+// case "CheckLock":
+// res = "check-lock";
+// break;
+// default :
+// res = action;
+// }
+// return res.toLowerCase();
+// }
+
+ @Override
+ public String buildRequest(InputBody msg) {
+ JSONObject jsonObject = Mapper.toJsonObject(msg);
+ return String.format(TEMPLATE, jsonObject.toString());
+ }
+
+/* @Override
+ public OperationStatus getOperationStatus(JSONObject responseBody) throws APPCException {
+ try {
+ JSONObject status = responseBody.getJSONObject("output").getJSONObject("status");
+ return new OperationStatus(String.valueOf(status.getInt("code")), status.getString("message"));
+ } catch (Exception e) {
+ LOG.error("Unknown error processing failed response from provider. Json not in expected format", e);
+ throw new APPCException("APPC has an unknown RPC error");
+ }
+ }*/
+
+ @Override
+ public ResponseStatus getResponseStatus(JsonNode responseBody)throws APPCException{
+ try {
+ JsonNode status = responseBody.get("output").get("status");
+ return new ResponseStatus(status.get("code").asInt(), status.get("message").asText());
+ } catch (Exception e) {
+ LOG.error("Unknown error processing failed response from provider. Json not in expected format", e);
+ throw new APPCException("APPC has an unknown RPC error");
+ }
+ }
+
+ @Override
+ public String getLocked(JSONObject responseBody) throws APPCException {
+ try {
+ JSONObject outputObject=responseBody.getJSONObject("output");
+ if(outputObject.has("locked")){
+ return outputObject.getString("locked");
+ }else{
+ return null;
+ }
+ } catch (Exception e) {
+ LOG.error("Unknown error processing failed response from provider. Json not in expected format", e);
+ throw new APPCException("APPC has an unknown RPC error");
+ }
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/OperationStatus.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/OperationStatus.java
new file mode 100644
index 000000000..b2388af45
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/OperationStatus.java
@@ -0,0 +1,66 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.operation;
+
+
+public class OperationStatus {
+ public final static OperationStatus PENDING = new OperationStatus("PENDING", "PENDING");
+ public final static OperationStatus ACTIVE = new OperationStatus("ACTIVE", "ACTIVE");
+ public final static OperationStatus SUCCESS = new OperationStatus("SUCCESS", "SUCCESS");
+ public final static OperationStatus FAILURE = new OperationStatus("FAILURE", "FAILURE");
+
+ private String code;
+ private String value;
+
+ public OperationStatus() {
+ }
+
+
+ public OperationStatus(String code, String value) {
+ this.code = code;
+ this.value = value;
+ }
+
+ public String getCode() {
+ return code;
+ }
+
+ public void setCode(String code) {
+ this.code = code;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ public void setValue(String value) {
+ this.value = value;
+ }
+
+ public boolean isSucceeded() {
+ if (code == null) {
+ return false;
+ }
+ int intCode = Integer.parseInt(code);
+ return (intCode >= 200) && (intCode < 300); // All 2xx statuses are success
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperationRequestFormatter.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperationRequestFormatter.java
new file mode 100644
index 000000000..33aa5b673
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperationRequestFormatter.java
@@ -0,0 +1,42 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.operation;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.json.JSONObject;
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.LCM.model.InputBody;
+import org.openecomp.appc.listener.LCM.model.ResponseStatus;
+
+import java.net.URL;
+
+
+public interface ProviderOperationRequestFormatter {
+
+ String buildPath(URL url, String rpcName);
+
+ String buildRequest(InputBody msg);
+
+ ResponseStatus getResponseStatus(JsonNode json)throws APPCException;
+
+ String getLocked(JSONObject json)throws APPCException;
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperations.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperations.java
new file mode 100644
index 000000000..f5672a575
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperations.java
@@ -0,0 +1,260 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM.operation;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.io.IOUtils;
+import org.apache.http.HttpHeaders;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpVersion;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.conn.ClientConnectionManager;
+import org.apache.http.conn.scheme.PlainSocketFactory;
+import org.apache.http.conn.scheme.Scheme;
+import org.apache.http.conn.scheme.SchemeRegistry;
+import org.apache.http.conn.ssl.SSLSocketFactory;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
+import org.apache.http.params.BasicHttpParams;
+import org.apache.http.params.HttpParams;
+import org.apache.http.params.HttpProtocolParams;
+import org.apache.http.protocol.HTTP;
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.LCM.model.ResponseStatus;
+import org.openecomp.appc.listener.util.Mapper;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.X509TrustManager;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.MalformedURLException;
+import java.net.Socket;
+import java.net.URL;
+import java.net.UnknownHostException;
+import java.security.*;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+
+public class ProviderOperations {
+
+ private static final EELFLogger LOG = EELFManager.getInstance().getLogger(ProviderOperations.class);
+
+ private static URL url;
+ private static String basic_auth;
+
+ private static ProviderOperationRequestFormatter requestFormatter = new GenericProviderOperationRequestFormatter();
+
+ /**
+ * Calls the AppcProvider to run a topology directed graph
+ *
+ * @param msg The incoming message to be run
+ * @return True if the result is success. Never returns false and throws an exception instead.
+ * @throws UnsupportedEncodingException
+ * @throws Exception if there was a failure processing the request. The exception message is the failure reason.
+ */
+ @SuppressWarnings("nls")
+ public static JsonNode topologyDG(String rpcName, JsonNode msg) throws APPCException {
+ if (msg == null) {
+ throw new APPCException("Provided message was null");
+ }
+
+ HttpPost post = null;
+ try {
+
+ // Concatenate the "action" on the end of the URL
+ String path = requestFormatter.buildPath(url, rpcName);
+ URL serviceUrl = new URL(url.getProtocol(), url.getHost(), url.getPort(), path);
+
+ post = new HttpPost(serviceUrl.toExternalForm());
+ post.setHeader(HttpHeaders.CONTENT_TYPE, "application/json");
+ post.setHeader(HttpHeaders.ACCEPT, "application/json");
+
+ // Set Auth
+ if (basic_auth != null) {
+ post.setHeader(HttpHeaders.AUTHORIZATION, "Basic " + basic_auth);
+ }
+
+ String body = Mapper.toJsonString(msg);
+ StringEntity entity = new StringEntity(body);
+ entity.setContentType("application/json");
+ post.setEntity(new StringEntity(body));
+ } catch (UnsupportedEncodingException | MalformedURLException e) {
+ throw new APPCException(e);
+ }
+
+ HttpClient client = getHttpClient();
+
+ int httpCode = 0;
+ String respBody = null;
+ try {
+ HttpResponse response = client.execute(post);
+ httpCode = response.getStatusLine().getStatusCode();
+ respBody = IOUtils.toString(response.getEntity().getContent());
+ } catch (IOException e) {
+ throw new APPCException(e);
+ }
+
+ if (httpCode >= 200 && httpCode < 300 && respBody != null) {
+ JsonNode json;
+ try {
+ json = Mapper.toJsonNodeFromJsonString(respBody);
+ } catch (Exception e) {
+ LOG.error("Error processing response from provider. Could not map response to json", e);
+ throw new APPCException("APPC has an unknown RPC error");
+ }
+
+ ResponseStatus responseStatus = requestFormatter.getResponseStatus(json);
+
+ if (!isSucceeded(responseStatus.getCode())) {
+ LOG.warn(String.format("Operation failed [%s]", msg.toString()));
+ }
+
+ return json;
+ }
+
+ throw new APPCException(String.format("Unexpected response from endpoint: [%d] - %s ", httpCode, respBody));
+ }
+
+ /**
+ * Updates the static var URL and returns the value;
+ *
+ * @return The new value of URL
+ */
+ public static String getUrl() {
+ return url.toExternalForm();
+ }
+
+ public static void setUrl(String newUrl) {
+ try {
+ url = new URL(newUrl);
+ } catch (MalformedURLException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * Sets the basic authentication header for the given user and password. If either entry is null then set basic auth
+ * to null
+ *
+ * @param user The user with optional domain name
+ * @param password The password for the user
+ * @return The new value of the basic auth string that will be used in the request headers
+ */
+ public static String setAuthentication(String user, String password) {
+ if (user != null && password != null) {
+ String authStr = user + ":" + password;
+ basic_auth = new String(Base64.encodeBase64(authStr.getBytes()));
+ } else {
+ basic_auth = null;
+ }
+ return basic_auth;
+ }
+
+ @SuppressWarnings("deprecation")
+ private static HttpClient getHttpClient() throws APPCException {
+ HttpClient client;
+ if (url.getProtocol().equals("https")) {
+ try {
+ KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
+ trustStore.load(null, null);
+ MySSLSocketFactory sf = new MySSLSocketFactory(trustStore);
+ sf.setHostnameVerifier(MySSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
+
+ HttpParams params = new BasicHttpParams();
+ HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1);
+ HttpProtocolParams.setContentCharset(params, HTTP.UTF_8);
+
+ SchemeRegistry registry = new SchemeRegistry();
+ registry.register(new Scheme("http", PlainSocketFactory.getSocketFactory(), 80));
+ registry.register(new Scheme("https", sf, 443));
+ registry.register(new Scheme("https", sf, 8443));
+ registry.register(new Scheme("http", sf, 8181));
+
+ ClientConnectionManager ccm = new ThreadSafeClientConnManager(params, registry);
+ client = new DefaultHttpClient(ccm, params);
+ } catch (Exception e) {
+ client = new DefaultHttpClient();
+ }
+ } else if (url.getProtocol().equals("http")) {
+ client = new DefaultHttpClient();
+ } else {
+ throw new APPCException(
+ "The provider.topology.url property is invalid. The url did not start with http[s]");
+ }
+ return client;
+ }
+
+ @SuppressWarnings("deprecation")
+ public static class MySSLSocketFactory extends SSLSocketFactory {
+ private SSLContext sslContext = SSLContext.getInstance("TLS");
+
+ public MySSLSocketFactory(KeyStore truststore) throws NoSuchAlgorithmException, KeyManagementException,
+ KeyStoreException, UnrecoverableKeyException {
+ super(truststore);
+
+ TrustManager tm = new X509TrustManager() {
+ @Override
+ public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+ }
+
+ @Override
+ public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+ }
+
+ @Override
+ public X509Certificate[] getAcceptedIssuers() {
+ return null;
+ }
+ };
+
+ sslContext.init(null, new TrustManager[]{
+ tm
+ }, null);
+ }
+
+ @Override
+ public Socket createSocket(Socket socket, String host, int port, boolean autoClose)
+ throws IOException, UnknownHostException {
+ return sslContext.getSocketFactory().createSocket(socket, host, port, autoClose);
+ }
+
+ @Override
+ public Socket createSocket() throws IOException {
+ return sslContext.getSocketFactory().createSocket();
+ }
+ }
+
+ public static boolean isSucceeded(Integer code) {
+ if (code == null) {
+ return false;
+ }
+ return ((code == 100) || (code == 400)); // only 100 & 400 statuses are success
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Listener.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Listener.java
new file mode 100644
index 000000000..a8e180404
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Listener.java
@@ -0,0 +1,72 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+/**
+ * This interface defines a listener that subscribes to a DMaaP topic and continually polls for messages. The
+ * listener does all operations in the run() method and long running operations should be created in a separate worker
+ * thread.
+ *
+ */
+public interface Listener extends Runnable {
+
+ /**
+ * Should start a continuous poll to get messages from the message bus only ending when stop() or stopNow() are
+ * called.
+ *
+ * @see java.lang.Runnable#run()
+ */
+ @Override
+ public void run();
+
+ /**
+ * Signals the listener to stop accepting new messages to the queue and to cleanly finish processing all remaining
+ * messages in the queue. This can take a significant amount of time to complete depending on the thread pool
+ * characteristics. Similar to {@link #stopNow()}
+ */
+ public void stop();
+
+ /**
+ * Signals the listener to stop accepting new messages to the queue and to destroy all remaining messages in the
+ * queue. This will complete quicker than {@link #stop()} at the cost of discarded requests. Recovery of these
+ * requests would have to be caught downstream. Similar to {@link #stop()}
+ */
+ public void stopNow();
+
+ /**
+ * @return A string that shows various benchmarking data. Can be used by humans to tune the thread pool.
+ */
+ public String getBenchmark();
+
+ /**
+ * @return The listener's id when requesting messages from DMaaP. Also known as the group id.
+ */
+ public String getListenerId();
+
+ /**
+ * Sets the listener's id to use when requesting messages from DMaaP. Also known as the group id.
+ *
+ * @param idString
+ * The new listener id
+ */
+ public void setListenerId(String idString);
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/ListenerProperties.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/ListenerProperties.java
new file mode 100644
index 000000000..c6c374eb0
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/ListenerProperties.java
@@ -0,0 +1,319 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+import java.util.Properties;
+
+/**
+ * A class for instantiating Listener objects. It is primarily used to hold properties that start with the given prefix.
+ * It also holds a class that implements {@see Listener} and will be used by the controller to spawn a new listener
+ * object.
+ *
+ * @since Apr 25, 2016
+ * @version $Id$
+ */
+public class ListenerProperties {
+
+ private String prefix;
+
+ private Class<? extends Listener> listenerClass;
+
+ private Properties props;
+
+ /**
+ * Creates a new listener object with the given prefix and properties. Any property starting with the prefix is
+ * added to the internal properties object with the prefix removed. All other properties are ignored.
+ * ListenerProperties constructor
+ *
+ * @param prefix
+ * The prefix of the properties to load
+ * @param allProps
+ * The properties object to load from.
+ */
+ public ListenerProperties(String prefix, Properties allProps) {
+ this.prefix = prefix;
+ props = new Properties();
+
+ String dottedPrefix = String.format("%s.", prefix);
+ for (String key : allProps.stringPropertyNames()) {
+ if (key.startsWith(dottedPrefix) && key.length() > dottedPrefix.length()) {
+ props.put(key.substring(dottedPrefix.length()), allProps.get(key));
+ }
+ }
+ }
+
+ /**
+ * @return The prefix of these properties
+ */
+ public String getPrefix() {
+ return prefix;
+ }
+
+ /**
+ * Sets the listener class. Will be used by {@see Controller} to instantiate the Listener thread for this object
+ *
+ * @param cls
+ * The class to be created. Implements {@see Listener}
+ */
+ public void setListenerClass(Class<? extends Listener> cls) {
+ this.listenerClass = cls;
+ }
+
+ /**
+ * @return The class that will be used by {@see Controller} to instantiate the Listener thread for this object
+ */
+ public Class<? extends Listener> getListenerClass() {
+ return listenerClass;
+ }
+
+ /**
+ * Returns a property matching a given KEYS
+ *
+ * @param key
+ * The KEYS object who's value to return.
+ * @return The value of the property or null if none exists
+ */
+ public String getProperty(KEYS key) {
+ return getProperty(key, null);
+ }
+
+ /**
+ * Returns a property matching a given string.
+ *
+ * @param key
+ * The key who's value to return.
+ * @return The value of the property or null if none exists
+ */
+ public String getProperty(String key) {
+ return getProperty(key, null);
+ }
+
+ /**
+ * Returns a property matching a given KEYS
+ *
+ * @param key
+ * The KEYS object who's value to return.
+ * @param defaultValue
+ * The value to return if the property is not found
+ * @return The value of the property or null if none exists
+ */
+ public String getProperty(KEYS key, String defaultValue) {
+ return getProperty(key.getPropertySuffix(), defaultValue);
+ }
+
+ /**
+ * Returns a property matching a given string.
+ *
+ * @param key
+ * The key who's value to return.
+ * @param defaultValue
+ * The value to return if the property is not found
+ * @return The value of the property or null if none exists
+ */
+ public String getProperty(String key, String defaultValue) {
+ return props.getProperty(key, defaultValue);
+ }
+
+ /**
+ * @return The properties object containing all properties
+ */
+ public Properties getProperties() {
+ return props;
+ }
+
+ /**
+ * Reads the <i>prefix</i>.disabled property to determine if the listener is disabled and should not be run by the
+ * controller. Defaults to false if property not set or value cannot be parsed.
+ *
+ * @return true if the listener is disabled and should not be started. false if the listener should be start
+ * normally (default).
+ */
+ public boolean isDisabled() {
+ return Boolean.valueOf(getProperty(KEYS.DISABLED, "false"));
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%s", prefix);
+ }
+
+ /**
+ * The message service types that are available. Choices are DMaaP and DMaaP.
+ *
+ * @since Apr 25, 2016
+ * @version $Id$
+ */
+ public enum MessageService {
+ DMaaP("dmaap");
+
+ private String val;
+
+ private MessageService(String val) {
+ this.val = val;
+ }
+
+ public String getValue() {
+ return val;
+ }
+
+ /**
+ * Tries to match a string to a MessageService. If no match is found, returns the default (DMaaP)
+ *
+ * @param input
+ * the string to try and match
+ * @return A MessasgeService
+ */
+ public static MessageService parse(String input) {
+ if (input != null) {
+ for (MessageService ms : MessageService.values()) {
+ if (ms.getValue().equals(input.toLowerCase())) {
+ return ms;
+ }
+ }
+ }
+ return MessageService.DMaaP; // Default
+ }
+ }
+
+ /**
+ * Set of common properties that will be used by most systems. Primarily relating to DMaaP and ThreadPools
+ *
+ * @since Apr 25, 2016
+ * @version $Id$
+ */
+ public enum KEYS {
+ /**
+ * Property to determine if the listener should be disabled. If not set, defaults to false
+ */
+ DISABLED("disabled"),
+
+ /**
+ * Property for the message service type. Should be a lower case string. See MessageService.
+ */
+ MESSAGE_SERVICE("service"),
+
+ /**
+ * A hostname or comma separated list (no spaces) of hostnames of servers in a cluster. Can have ports included
+ * as well.<br>
+ * Examples:
+ * <ul>
+ * <li>server1.appc.openecomp.org</li>
+ * <li>server1.appc.openecomp.org:3904</li>
+ * <li>server1.appc.openecomp.org,server2.appc.openecomp.org</li>
+ * </ul>
+ */
+ HOSTS("poolMembers"),
+
+ /**
+ * The topic that will be used for DMaaP read operations. Can only support a single topic.
+ */
+ TOPIC_READ("topic.read"),
+
+ /**
+ * The topic or topics that will be used to write to. If multiple topics are provided, should be in a comma
+ * seperated list with no spaces.<br>
+ * Examples:
+ * <ul>
+ * <li>TOPIC-1</li>
+ * <li>TOPIC-1,TOPIC-2,ANOTHER-TOPIC</li>
+ * </ul>
+ */
+ TOPIC_WRITE("topic.write"),
+
+ /**
+ * The highland park filter to use on read requests. If you are reading and writing to the same topic this must
+ * be provided. Filter should be in JSON format (not url escaped).
+ */
+ TOPIC_READ_FILTER("topic.read.filter"),
+
+ /**
+ * The amount of time in seconds that the DMaaP polling connection should stay open for. Recommended to be set
+ * high (around 60 seconds) as most clients will return immediately and not wait until the timeout is up to
+ * return if they have data.
+ */
+ TOPIC_READ_TIMEOUT("topic.read.timeout"),
+
+ /**
+ * The name of the client to use. Should be unique to the application.
+ */
+ CLIENT_NAME("client.name"),
+
+ /**
+ * The id of the client to use. Should be unique for each instance of the application in an environment.
+ */
+ CLIENT_ID("client.name.id"),
+
+ /**
+ * The User (DMaaP) to use for authentication. If a user is provided, you must include the
+ * domain name (e.g. example<b>@example.com</b>).
+ */
+ AUTH_USER_KEY("client.key"),
+
+ /**
+ * The password (DMaaP) to use for authentication.
+ */
+ AUTH_SECRET_KEY("client.secret"),
+
+ /**
+ * The minimum amount of size of the queue. A client should request new messages once the queue has dropped
+ * below this size.
+ */
+ THREADS_MIN_QUEUE("threads.queuesize.min"),
+
+ /**
+ * The maximum size of the queue. A client will request no new messages once this maximum size has been reached.
+ */
+ THREADS_MAX_QUEUE("threads.queuesize.max"),
+
+ /**
+ * The minimum size of the worker threads pool. This is the pool each listener will use to launch longer running
+ * operations.
+ */
+ THREADS_MIN_POOL("threads.poolsize.min"),
+
+ /**
+ * The maximum size of the worker threads pool. This is the pool each listener will use to launch longer running
+ * operations.
+ */
+ THREADS_MAX_POOL("threads.poolsize.max");
+
+ private String suffix;
+
+ private KEYS(String val) {
+ this.suffix = val;
+ }
+
+ /**
+ * @param prefix
+ * The prefix to prepend
+ * @return a fully property name that corroponds to what is used in the properties file. Format is PREFIX.KEY
+ */
+ public String getFullProp(String prefix) {
+ return String.format("%s.%s", prefix, suffix);
+ }
+
+ public String getPropertySuffix() {
+ return suffix;
+ }
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/ControllerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/ControllerImpl.java
new file mode 100644
index 000000000..9e171915a
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/ControllerImpl.java
@@ -0,0 +1,126 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.impl;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.openecomp.appc.listener.Controller;
+import org.openecomp.appc.listener.Listener;
+import org.openecomp.appc.listener.ListenerProperties;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+
+/**
+ * A common implementation of a controller. This controller should not need to be modified to implement new listeners
+ *
+ */
+public class ControllerImpl implements Controller {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(ControllerImpl.class);
+
+ private int LISTENER_COUNT = 1;
+
+ private Map<ListenerProperties, Listener> listeners = null;
+
+ private ThreadPoolExecutor executor;
+
+ /**
+ * Creates a Controller with the set of listener properties which will be used to start listener threads.
+ *
+ * @param properties
+ * A non null Set of ListenerProperties
+ */
+ public ControllerImpl(Set<ListenerProperties> properties) {
+ listeners = new HashMap<ListenerProperties, Listener>();
+ for (ListenerProperties props : properties) {
+ if (props.getClass() != null) {
+ listeners.put(props, null);
+ } else {
+ LOG.error(String.format(
+ "The ListenerProperties %s has no Listener class associated with it and will not run.", props));
+ }
+ }
+
+ LISTENER_COUNT = properties.size();
+
+ executor = new ThreadPoolExecutor(LISTENER_COUNT, LISTENER_COUNT, 1, TimeUnit.SECONDS,
+ new ArrayBlockingQueue<Runnable>(LISTENER_COUNT));
+
+ // Custom Named thread factory
+ BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("Appc-Listener-%d").build();
+ executor.setThreadFactory(threadFactory);
+ }
+
+ @Override
+ public void start() {
+ LOG.info("Starting DMaaP Controller.");
+ for (ListenerProperties props : listeners.keySet()) {
+ try {
+ if (props.isDisabled()) {
+ LOG.warn(String.format("The listener %s is disabled and will not be run", props.getPrefix()));
+ } else {
+ Listener l = props.getListenerClass().getConstructor(ListenerProperties.class).newInstance(props);
+ l.setListenerId(props.getPrefix());
+ listeners.put(props, l);
+ executor.execute(l);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOG.error(String.format("Exception while starting listener %s.", props), e);
+ }
+ }
+ }
+
+ @Override
+ public void stop(boolean stopNow) {
+ LOG.info("Stopping DMaaP Controller.");
+ Iterator<Listener> itr = listeners.values().iterator();
+ while (itr.hasNext()) {
+ Listener l = itr.next();
+ if (stopNow) {
+ l.stopNow();
+ } else {
+ l.stop();
+ }
+ itr.remove();
+ }
+ executor.shutdown();
+ try {
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ }
+ }
+
+ @Override
+ public Map<ListenerProperties, Listener> getListeners() {
+ return listeners;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java
new file mode 100644
index 000000000..590afbe8b
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java
@@ -0,0 +1,326 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.impl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.openecomp.appc.adapter.dmaap.Consumer;
+import org.openecomp.appc.adapter.dmaap.DmaapConsumer;
+import org.openecomp.appc.adapter.dmaap.DmaapProducer;
+import org.openecomp.appc.adapter.dmaap.Producer;
+import org.openecomp.appc.adapter.dmaap.DmaapConsumer;
+import org.openecomp.appc.adapter.dmaap.DmaapProducer;
+import org.openecomp.appc.listener.EventHandler;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.ListenerProperties.MessageService;
+import org.openecomp.appc.listener.util.Mapper;
+import org.openecomp.appc.logging.LoggingConstants;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import org.slf4j.MDC;
+
+/**
+ * This class is a wrapper for the DMaaP client provided in appc-dmaap-adapter. Its aim is to ensure that only well formed
+ * messages are sent and received on DMaaP.
+ *
+ */
+public class EventHandlerImpl implements EventHandler {
+
+ private final EELFLogger LOG = EELFManager.getInstance().getLogger(EventHandlerImpl.class);
+
+ /*
+ * The amount of time in seconds to keep a connection to a topic open while waiting for data
+ */
+ private int READ_TIMEOUT = 60;
+
+ /*
+ * The pool of hosts to query against
+ */
+ private Collection<String> pool;
+
+ /*
+ * The topic to read messages from
+ */
+ private String readTopic;
+
+ /*
+ * The topic to write messages to
+ */
+ private Set<String> writeTopics;
+
+ /*
+ * The client (group) name to use for reading messages
+ */
+ private String clientName;
+
+ /*
+ * The id of the client (group) that is reading messages
+ */
+ private String clientId;
+
+ /*
+ * The api public key to use for authentication
+ */
+ private String apiKey;
+
+ /*
+ * The api secret key to use for authentication
+ */
+ private String apiSecret;
+
+ /*
+ * A json object containing filter arguments.
+ */
+ private String filter_json;
+
+ private MessageService messageService;
+
+ private Consumer reader = null;
+ private Producer producer = null;
+
+ public EventHandlerImpl(ListenerProperties props) {
+ pool = new HashSet<String>();
+ writeTopics = new HashSet<String>();
+
+ if (props != null) {
+ readTopic = props.getProperty(ListenerProperties.KEYS.TOPIC_READ);
+ clientName = props.getProperty(ListenerProperties.KEYS.CLIENT_NAME, "APP-C");
+ clientId = props.getProperty(ListenerProperties.KEYS.CLIENT_ID, "0");
+ apiKey = props.getProperty(ListenerProperties.KEYS.AUTH_USER_KEY);
+ apiSecret = props.getProperty(ListenerProperties.KEYS.AUTH_SECRET_KEY);
+
+ filter_json = props.getProperty(ListenerProperties.KEYS.TOPIC_READ_FILTER);
+
+ READ_TIMEOUT = Integer
+ .valueOf(props.getProperty(ListenerProperties.KEYS.TOPIC_READ_TIMEOUT, String.valueOf(READ_TIMEOUT)));
+
+ String hostnames = props.getProperty(ListenerProperties.KEYS.HOSTS);
+ if (hostnames != null && !hostnames.isEmpty()) {
+ for (String name : hostnames.split(",")) {
+ pool.add(name);
+ }
+ }
+
+ String writeTopicStr = props.getProperty(ListenerProperties.KEYS.TOPIC_WRITE);
+ if (writeTopicStr != null) {
+ for (String topic : writeTopicStr.split(",")) {
+ writeTopics.add(topic);
+ }
+ }
+
+ messageService = MessageService.parse(props.getProperty(ListenerProperties.KEYS.MESSAGE_SERVICE));
+
+ LOG.info(String.format(
+ "Configured to use %s client on host pool [%s]. Reading from [%s] filtered by %s. Wriring to [%s]. Authenticated using %s",
+ messageService, hostnames, readTopic, filter_json, writeTopics, apiKey));
+ }
+ }
+
+ @Override
+ public List<String> getIncomingEvents() {
+ return getIncomingEvents(1000);
+ }
+
+ @Override
+ public List<String> getIncomingEvents(int limit) {
+ List<String> out = new ArrayList<String>();
+ LOG.info(String.format("Getting up to %d incoming events", limit));
+ // reuse the consumer object instead of creating a new one every time
+ if (reader == null) {
+ LOG.info("Getting Consumer...");
+ reader = getConsumer();
+ }
+ for (String item : reader.fetch(READ_TIMEOUT * 1000, limit)) {
+ out.add(item);
+ }
+ LOG.info(String.format("Read %d messages from %s as %s/%s.", out.size(), readTopic, clientName, clientId));
+ return out;
+ }
+
+ @Override
+ public <T> List<T> getIncomingEvents(Class<T> cls) {
+ return getIncomingEvents(cls, 1000);
+ }
+
+ @Override
+ public <T> List<T> getIncomingEvents(Class<T> cls, int limit) {
+ List<String> incomingStrings = getIncomingEvents(limit);
+ return Mapper.mapList(incomingStrings, cls);
+ }
+
+ @Override
+ public void postStatus(String event) {
+ postStatus(null, event);
+ }
+
+ @Override
+ public void postStatus(String partition, String event) {
+ LOG.debug(String.format("Posting Message [%s]", event));
+ if (producer == null) {
+ LOG.info("Getting Producer...");
+ producer = getProducer();
+ }
+ producer.post(partition, event);
+ }
+
+ /**
+ * Returns a consumer object for direct access to our Cambria consumer interface
+ *
+ * @return An instance of the consumer interface
+ */
+ protected Consumer getConsumer() {
+ LOG.debug(String.format("Getting Consumer: %s %s/%s/%s", pool, readTopic, clientName, clientId));
+ if (filter_json == null && writeTopics.contains(readTopic)) {
+ LOG.error(
+ "*****We will be writing and reading to the same topic without a filter. This will cause an infinite loop.*****");
+ }
+ Consumer out;
+ out = new DmaapConsumer(pool, readTopic, clientName, clientId, filter_json, apiKey, apiSecret);
+ for (String url : pool) {
+ if (url.contains("3905") || url.contains("https")) {
+ out.useHttps(true);
+ break;
+ }
+ }
+ return out;
+ }
+
+ /**
+ * Returns a consumer object for direct access to our Cambria producer interface
+ *
+ * @return An instance of the producer interface
+ */
+ protected Producer getProducer() {
+ LOG.debug(String.format("Getting Producer: %s %s", pool, readTopic));
+
+ Producer out;
+ out = new DmaapProducer(pool,writeTopics);
+
+ if (apiKey != null && apiSecret != null) {
+ out.updateCredentials(apiKey, apiSecret);
+ }
+
+ for (String url : pool) {
+ if (url.contains("3905") || url.contains("https")) {
+ out.useHttps(true);
+ break;
+ }
+ }
+ return out;
+ }
+
+ @Override
+ public void closeClients() {
+ LOG.debug("Closing Consumer and Producer DMaaP clients");
+ switch (messageService) {
+ case DMaaP:
+ if (reader != null) {
+ ((DmaapConsumer) reader).close();
+ }
+ if (producer != null) {
+ ((DmaapProducer) producer).close();
+ }
+ break;
+ default:
+ // close DMaaP clients
+ if (reader != null) {
+ ((DmaapConsumer) reader).close();
+ }
+ if (producer != null) {
+ ((DmaapProducer) producer).close();
+ }
+ }
+ }
+
+ @Override
+ public String getClientId() {
+ return clientId;
+ }
+
+ @Override
+ public void setClientId(String clientId) {
+ this.clientId = clientId;
+ }
+
+ @Override
+ public String getClientName() {
+ return clientName;
+ }
+
+ @Override
+ public void setClientName(String clientName) {
+ this.clientName = clientName;
+ MDC.put(LoggingConstants.MDCKeys.PARTNER_NAME, clientName);
+ }
+
+ @Override
+ public void addToPool(String hostname) {
+ pool.add(hostname);
+ }
+
+ @Override
+ public Collection<String> getPool() {
+ return pool;
+ }
+
+ @Override
+ public void removeFromPool(String hostname) {
+ pool.remove(hostname);
+ }
+
+ @Override
+ public String getReadTopic() {
+ return readTopic;
+ }
+
+ @Override
+ public void setReadTopic(String readTopic) {
+ this.readTopic = readTopic;
+ }
+
+ @Override
+ public Set<String> getWriteTopics() {
+ return writeTopics;
+ }
+
+ @Override
+ public void setWriteTopics(Set<String> writeTopics) {
+ this.writeTopics = writeTopics;
+ }
+
+ @Override
+ public void clearCredentials() {
+ apiKey = null;
+ apiSecret = null;
+ }
+
+ @Override
+ public void setCredentials(String key, String secret) {
+ apiKey = key;
+ apiSecret = secret;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/util/Mapper.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/util/Mapper.java
new file mode 100644
index 000000000..cdb0d14f0
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/util/Mapper.java
@@ -0,0 +1,134 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.util;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.json.JSONObject;
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class Mapper {
+
+ private static final EELFLogger LOG = EELFManager.getInstance().getLogger(Mapper.class);
+
+ private static ObjectMapper mapper = new ObjectMapper();
+
+ /**
+ * @return The object mapper that we are using.
+ */
+ public static ObjectMapper getMapper() {
+ return mapper;
+ }
+
+ /**
+ * Convert a String to a DcaeMessage
+ *
+ * @param data
+ * The json string to try and parse
+ * @return A DcaeMessage from the json string or null if it could not
+ */
+ public static <T> T mapOne(String data, Class<T> cls) {
+ try {
+ return mapper.readValue(data, cls);
+ } catch (Exception e) {
+ LOG.warn(String.format("Could not map [ %s ] to %s", data, cls.getName()), e);
+ return null;
+ }
+ }
+
+ public static <T> List<T> mapList(List<String> data, Class<T> cls) {
+ List<T> out = new ArrayList<T>();
+ for (String s : data) {
+ T tmp = Mapper.mapOne(s, cls);
+ if (tmp != null) {
+ out.add(tmp);
+ }
+ }
+ return out;
+ }
+
+ /**
+ * Convenience method to try and convert objects to json String
+ *
+ * @param obj
+ * The object to try and convert
+ * @return A json string representing the object or null if it could not be converted
+ */
+ public static String toJsonString(Object obj) {
+ String jsonStr;
+ try {
+ if (obj instanceof JSONObject) {
+ jsonStr = obj.toString();
+ }else {
+ jsonStr = mapper.writeValueAsString(obj);
+ }
+ return jsonStr;
+ } catch (Exception e) {
+ LOG.warn(String.format("Could not map %s to JSONObject.", obj), e);
+ return null;
+ }
+ }
+
+ public static JSONObject toJsonObject(Object obj) {
+ String jsonStr;
+ try {
+ if (obj.getClass().equals(String.class)) {
+ jsonStr = (String) obj;
+ } else {
+ jsonStr = mapper.writeValueAsString(obj);
+ }
+ return new JSONObject(jsonStr);
+ } catch (Exception e) {
+ LOG.warn(String.format("Could not map %s to JSONObject.", obj), e);
+ return null;
+ }
+ }
+ public static JsonNode toJsonNodeFromJsonString(String jsonStr) {
+ JsonNode jsonNode = null;
+ if(jsonStr != null) {
+ try {
+ jsonNode = mapper.readTree(jsonStr);
+ } catch (IOException e) {
+ LOG.warn(String.format("Could not map %s to jsonNode.", jsonStr), e);
+ }
+ }
+ return jsonNode;
+ }
+ public static JsonNode toJsonNode(Object obj) {
+ JsonNode jsonNode = null;
+ String jsonStr = toJsonString(obj);
+ if(jsonStr != null) {
+ try {
+ jsonNode = mapper.readTree(jsonStr);
+ } catch (IOException e) {
+ LOG.warn(String.format("Could not map %s to JSONObject.", obj), e);
+ }
+ }
+ return jsonNode;
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/resources/org/openecomp/appc/default.properties b/appc-event-listener/appc-event-listener-bundle/src/main/resources/org/openecomp/appc/default.properties
new file mode 100644
index 000000000..25c49ac4a
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/main/resources/org/openecomp/appc/default.properties
@@ -0,0 +1,99 @@
+###
+# ============LICENSE_START=======================================================
+# openECOMP : APP-C
+# ================================================================================
+# Copyright (C) 2017 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=========================================================
+###
+
+# ${user.home} usually goes to /root if instantiation uses the appc-docker approach
+
+### ###
+### ###
+###Properties below that are commented out are/need to be provided in appc.properties###
+### ###
+### ###
+
+
+
+
+# Define the name and path of any user-provided configuration (bootstrap) file that can be loaded to supply configuration options
+org.openecomp.appc.bootstrap.file=appc.properties
+org.openecomp.appc.bootstrap.path=/opt/openecomp/appc/data/properties,${user.home},.
+
+
+
+
+
+### ###
+###Closed Loop - properties ###
+### ###
+appc.ClosedLoop.poolMembers=192.168.1.2:3904
+appc.ClosedLoop.topic.read=APPC-TEST2
+appc.ClosedLoop.topic.write=APPC-TEST2
+appc.ClosedLoop.topic.read.filter={"class": "And","filters": [{"class": "Assigned","field": "request"},{"class": "Unassigned","field": "response"}]}
+appc.ClosedLoop.client.name=APPC-TEST-CLIENT-EVT-LST-MAIN
+appc.ClosedLoop.client.name.id=0
+
+# KEY AND SECRET BELOW NEED TO BE MODIFIED TO APPROPRIATE VALUES WHEN THE TOPIC IS API-KEY-BASED AUTH
+#dmaap.client.key=random
+#dmaap.client.secret=random
+
+appc.ClosedLoop.threads.queuesize.min=1
+appc.ClosedLoop.threads.queuesize.max=1000
+appc.ClosedLoop.threads.poolsize.min=1
+appc.ClosedLoop.threads.poolsize.max=2
+appc.ClosedLoop.provider.url=http://localhost:8181/restconf/operations/appc-provider
+appc.ClosedLoop.provider.url.user=ODL_USER
+appc.ClosedLoop.provider.url.pass=ODL_PASS
+
+
+
+
+
+### ###
+###Closed Loop - 1607 properties ###
+### ###
+appc.ClosedLoop1607.poolMembers=192.168.1.2:3904
+appc.ClosedLoop1607.topic.read=MY_DMAAP_TOPIC
+appc.ClosedLoop1607.topic.write=MY_DMAAP_TOPIC
+appc.ClosedLoop1607.topic.read.filter={"class":"Unassigned","field":"Status"}
+appc.ClosedLoop1607.client.name=MY_DMAAP_CLIENT_NAME
+appc.ClosedLoop1607.client.name.id=0
+#dmaap.client.key=random
+#dmaap.client.secret=random
+appc.ClosedLoop1607.threads.queuesize.min=1
+appc.ClosedLoop1607.threads.queuesize.max=1000
+appc.ClosedLoop1607.threads.poolsize.min=1
+appc.ClosedLoop1607.threads.poolsize.max=2
+appc.ClosedLoop.provider.user=ODL_USER
+appc.ClosedLoop.provider.pass=ODL_PASS
+appc.ClosedLoop1607.provider.url=http://localhost:8181/restconf/operations/appc-provider
+
+
+
+
+
+### ###
+### This needs to be changed so that the action can be appended to the end of the URL path ###
+### ###
+appc.LCM.provider.url=http://localhost:8181/restconf/operations/appc-provider
+appc.LCM.poolMembers=192.168.1.2:3904
+appc.LCM.topic.read=DMAAP_TOPIC
+appc.LCM.topic.write=DMAAP_TOPIC
+appc.LCM.client.name=DMAAP_CLIENT_NAME
+appc.LCM.provider.user=ODL_USER
+appc.LCM.provider.pass=ODL_URL
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestListenerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestListenerImpl.java
new file mode 100644
index 000000000..63c98458a
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestListenerImpl.java
@@ -0,0 +1,43 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.impl;
+
+import java.util.Properties;
+
+import org.junit.Before;
+import org.openecomp.appc.configuration.ConfigurationFactory;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.CL.impl.ListenerImpl;
+
+public class TestListenerImpl {
+
+ ListenerImpl listener;
+
+ @Before
+ public void setup() {
+ Properties props = ConfigurationFactory.getConfiguration().getProperties();
+ ListenerProperties lProps = new ListenerProperties((String) props.get("test.prefix"), props);
+ lProps.getProperties().setProperty("topic.read.timeout", "5"); // Just for this test
+ listener = new ListenerImpl(lProps);
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestProviderOperations.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestProviderOperations.java
new file mode 100644
index 000000000..3d7a7182f
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestProviderOperations.java
@@ -0,0 +1,122 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+import java.util.Properties;
+
+import org.eclipse.osgi.internal.signedcontent.Base64;
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.appc.configuration.ConfigurationFactory;
+import org.openecomp.appc.exceptions.APPCException;
+import org.openecomp.appc.listener.CL.impl.ProviderOperations;
+import org.openecomp.appc.listener.CL.model.IncomingMessage;
+
+public class TestProviderOperations {
+
+ private String ACTIVE_ENDPOINT;
+
+ @Before
+ public void setup() {
+ Properties props = ConfigurationFactory.getConfiguration().getProperties();
+ ACTIVE_ENDPOINT = props.getProperty("appc.ClosedLoop.provider.url");
+ assertNotNull(ACTIVE_ENDPOINT);
+ ProviderOperations.setUrl(ACTIVE_ENDPOINT);
+
+ props.getProperty("test.vm_url");
+ assertNotNull("VM_URL");
+ }
+
+ @Test
+ public void testTopologyOperation() {
+ IncomingMessage msg = new IncomingMessage();
+ // Client and Time are for ID
+ msg.setRequestClient("APPC");
+ msg.setRequestTime("TEST");
+ msg.setRequest("Restart");
+
+ // Null Input
+ try {
+ ProviderOperations.topologyDG(null);
+ fail("Topology Operation with null input should fail");
+ } catch (APPCException e) {
+ assertNotNull(e.getMessage());
+ }
+
+ // Bad URL
+ msg.setUrl("some bad url here");
+ try {
+ ProviderOperations.topologyDG(msg);
+ // Could also be issue in IaaS Adapter
+ fail("Topology Operation with bad url should fail");
+ } catch (APPCException e) {
+ assertNotNull(e.getMessage());
+ }
+
+ // Will be tested in worker
+ // msg.setUrl(VM_URL);
+ // System.out.println("Rebooting real VM. Test can take up to 90s");
+ // try {
+ // assertTrue(ProviderOperations.topologyDG(msg));
+ // } catch (APPCException e) {
+ // fail("Topology Operation with good url should succeed. Check url in gui first");
+ // }
+
+ }
+
+ @Test
+ public void testConfigurationOperation() {
+ try {
+ ProviderOperations.topologyDG(null);
+ fail("Configuration Operation should throw execption. Not yet supported");
+ } catch (APPCException e) {
+ assertNotNull(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testBasicAuthFormating() {
+ String user = "user";
+ String pass = "pass";
+
+ String result = ProviderOperations.setAuthentication(user, pass);
+
+ assertNotNull(result);
+ String decode = new String(Base64.decode(result.getBytes()));
+ assertEquals(user + ":" + pass, decode);
+ }
+
+ @Test
+ public void testGetSet() {
+ // Every test URL will get reset
+ assertEquals(ACTIVE_ENDPOINT, ProviderOperations.getUrl());
+
+ String newUrl = "http://example.com";
+ ProviderOperations.setUrl(newUrl);
+ assertEquals(newUrl, ProviderOperations.getUrl());
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestWorker.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestWorker.java
new file mode 100644
index 000000000..1cd9cdc44
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/impl/TestWorker.java
@@ -0,0 +1,72 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.impl;
+
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Properties;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.appc.configuration.ConfigurationFactory;
+import org.openecomp.appc.listener.EventHandler;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.CL.impl.ProviderOperations;
+import org.openecomp.appc.listener.CL.impl.WorkerImpl;
+import org.openecomp.appc.listener.CL.model.IncomingMessage;
+import org.openecomp.appc.listener.impl.EventHandlerImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestWorker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WorkerImpl.class);
+
+ private IncomingMessage msg;
+ private EventHandler dmaap;
+
+ @Before
+ public void setup() {
+ Properties props = ConfigurationFactory.getConfiguration().getProperties();
+ String activeEndpoint = props.getProperty("appc.ClosedLoop.provider.url");
+ assertNotNull(activeEndpoint);
+ ProviderOperations.setUrl(activeEndpoint);
+
+ String vmUrl = props.getProperty("test.vm_url");
+ assertNotNull(vmUrl);
+ msg = new IncomingMessage();
+ // Client and Time are for ID
+ msg.setRequestClient("APPC");
+ msg.setRequestTime("TEST");
+ msg.setRequest("Restart");
+ msg.setUrl(vmUrl);
+
+ dmaap = new EventHandlerImpl(new ListenerProperties("appc.ClosedLoop", props));
+ }
+
+ @Test
+ public void testWorker() {
+ WorkerImpl w = new WorkerImpl(msg, dmaap);
+ w.run();
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/model/TestEnums.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/model/TestEnums.java
new file mode 100644
index 000000000..a934cb432
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/model/TestEnums.java
@@ -0,0 +1,58 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.model;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import org.junit.Test;
+import org.openecomp.appc.listener.CL.model.Action;
+import org.openecomp.appc.listener.CL1607.model.Status;
+
+public class TestEnums {
+
+ @Test
+ public void testAction() {
+ assertEquals(Action.Rebuild, Action.toAction("Rebuild"));
+ assertEquals(Action.Restart, Action.toAction("restart"));
+ assertEquals(Action.Migrate, Action.toAction("MIGRATE"));
+ assertEquals(Action.Evacuate, Action.toAction("Evacuate"));
+ assertNull(Action.toAction("Unknown"));
+ assertNull(Action.toAction(null));
+
+ assertEquals(4, Action.values().length);
+ }
+
+ @Test
+ public void testStatus() {
+
+ assertEquals(Status.ACCEPTED, Status.toStatus("ACCEPTED"));
+ assertEquals(Status.SUCCESS, Status.toStatus("SuCcEsS"));
+ assertEquals(Status.FAILURE, Status.toStatus("Failure"));
+ assertNull(Status.toStatus("Unknown"));
+ assertNull(Status.toStatus(null));
+
+ assertEquals(3, Status.values().length);
+
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/model/TestMessages.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/model/TestMessages.java
new file mode 100644
index 000000000..e9fa1ace8
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL/model/TestMessages.java
@@ -0,0 +1,151 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL.model;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import org.apache.commons.io.IOUtils;
+import org.json.JSONObject;
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.appc.listener.CL.model.IncomingMessage;
+import org.openecomp.appc.listener.CL.model.OutgoingMessage;
+import org.openecomp.appc.listener.CL.model.Status;
+import org.openecomp.appc.listener.util.Mapper;
+
+public class TestMessages {
+ private IncomingMessage in;
+ private OutgoingMessage out;
+
+ private String incomingStr;
+ private String outgoingStr;
+
+ @Before
+ public void setup() {
+ try {
+ incomingStr = IOUtils.toString(getClass().getResourceAsStream("/IncomingMessage.txt"), "UTF-8");
+ outgoingStr = IOUtils.toString(getClass().getResourceAsStream("/OutgoingMessage.txt"), "UTF-8");
+ assertNotNull(incomingStr);
+ assertNotNull(outgoingStr);
+
+ in = Mapper.mapOne(incomingStr, IncomingMessage.class);
+
+ out = Mapper.mapOne(in.toOutgoing(Status.PENDING), OutgoingMessage.class);
+ out.updateResponseTime();
+
+ assertNotNull(in);
+ assertNotNull(out);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // NOTE Test Mapper will be used to test an event from dmaap.
+ @Test
+ public void testGetterSetter() {
+ assertNotNull(in);
+ assertNotNull(in.getRequestClient());
+ assertNotNull(in.getRequestTime());
+ assertNotNull(in.getMessage());
+ assertNotNull(in.getTenantId());
+ assertNotNull(in.getVmId());
+ assertNotNull(in.getVmName());
+ assertNotNull(in.getAction());
+ assertNotNull(in.getId());
+ assertNotNull(in.getPolicyName());
+ assertNotNull(in.getPolicyVersion());
+ assertNotNull(in.getRequest());
+
+ out = Mapper.mapOne(in.toOutgoing(null), OutgoingMessage.class);
+ assertNotNull(out.getRequestClient());
+ assertEquals(in.getRequestClient(), out.getRequestClient());
+ assertNotNull(out.getRequestTime());
+ assertEquals(in.getRequestTime(), out.getRequestTime());
+ assertNotNull(out.getMessage());
+ assertEquals(in.getMessage(), out.getMessage());
+ assertNotNull(out.getVmName());
+ assertEquals(in.getVmName(), out.getVmName());
+ assertNotNull(out.getPolicyName());
+ assertEquals(in.getPolicyName(), out.getPolicyName());
+ assertNotNull(out.getPolicyVersion());
+ assertEquals(in.getPolicyVersion(), out.getPolicyVersion());
+ assertNotNull(out.getOriginalRequest());
+ assertNotNull(in.getRequest(), out.getOriginalRequest());
+ }
+
+ @Test
+ public void testToString() {
+ in = new IncomingMessage();
+ assertNotNull(in.toString());
+ String id = "test";
+ in.setId(id);
+ assertNotNull(in.toString());
+ assertTrue(in.toString().contains(id));
+ }
+
+ @Test
+ public void testOutgoingUpdateTime() {
+ String old = out.getResponseTime();
+ out.updateResponseTime();
+ assertFalse(old.equals(out.getResponseTime()));
+ }
+
+ // Testing for 1510
+ @Test
+ public void testOutgoingToJson() {
+ // Message Set
+ String message = "MSG";
+ out.setMessage(message);
+ JSONObject json = out.toResponse();
+ assertNotNull(json);
+ String respStr = json.getString("response");
+ assertTrue(respStr.contains(out.getResponse().getValue()));
+
+ String msgStr = json.getString("message");
+ assertNotNull(msgStr);
+ assertFalse(msgStr.contains(out.getOriginalRequest())); // False for 1602
+ assertTrue(msgStr.contains(out.getMessage()));
+
+ // Null Message
+ out.setMessage(null);
+ json = out.toResponse();
+ assertNotNull(json);
+ msgStr = json.getString("message");
+ assertNotNull(msgStr);
+ assertFalse(msgStr.contains(out.getOriginalRequest())); // False for 1602
+ assertTrue(msgStr.contains(out.getResponse().getValue()));
+
+ // Echoing request
+ assertNotNull(out.getOriginalRequest());
+ }
+
+ @Test
+ public void testOutgoingToString() {
+ String s = out.toString();
+ assertTrue(s.contains(out.getId()));
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL1607/model/TestEnums.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL1607/model/TestEnums.java
new file mode 100644
index 000000000..73e38aa4f
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL1607/model/TestEnums.java
@@ -0,0 +1,59 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.model;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import org.junit.Test;
+import org.openecomp.appc.listener.CL.model.Action;
+import org.openecomp.appc.listener.CL.model.Status;
+
+public class TestEnums {
+
+ @Test
+ public void testAction() {
+ assertEquals(Action.Rebuild, Action.toAction("Rebuild"));
+ assertEquals(Action.Restart, Action.toAction("restart"));
+ assertEquals(Action.Migrate, Action.toAction("MIGRATE"));
+ assertEquals(Action.Evacuate, Action.toAction("Evacuate"));
+ assertNull(Action.toAction("Unknown"));
+ assertNull(Action.toAction(null));
+
+ assertEquals(4, Action.values().length);
+ }
+
+ @Test
+ public void testStatus() {
+
+ assertEquals(Status.PENDING, Status.toStatus("Pending"));
+ assertEquals(Status.ACTIVE, Status.toStatus("active"));
+ assertEquals(Status.SUCCESS, Status.toStatus("SuCcEsS"));
+ assertEquals(Status.FAILURE, Status.toStatus("Failure"));
+ assertNull(Status.toStatus("Unknown"));
+ assertNull(Status.toStatus(null));
+
+ assertEquals(4, Status.values().length);
+
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL1607/model/TestMessages.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL1607/model/TestMessages.java
new file mode 100644
index 000000000..c7bc86300
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/CL1607/model/TestMessages.java
@@ -0,0 +1,160 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.CL1607.model;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import org.apache.commons.io.IOUtils;
+import org.json.JSONObject;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.openecomp.appc.listener.CL1607.model.IncomingMessage;
+import org.openecomp.appc.listener.CL1607.model.OutgoingMessage;
+import org.openecomp.appc.listener.CL1607.model.Status;
+import org.openecomp.appc.listener.util.Mapper;
+
+public class TestMessages {
+ private IncomingMessage in;
+ private OutgoingMessage out;
+
+ private String incomingStr;
+ private String outgoingStr;
+
+ @Before
+ public void setup() {
+ try {
+ incomingStr = IOUtils.toString(getClass().getResourceAsStream("/IncomingMessage1607.txt"), "UTF-8");
+ outgoingStr = IOUtils.toString(getClass().getResourceAsStream("/OutgoingMessage1607.txt"), "UTF-8");
+ assertNotNull(incomingStr);
+ assertNotNull(outgoingStr);
+
+ in = Mapper.mapOne(incomingStr, IncomingMessage.class);
+
+ out = Mapper.mapOne(outgoingStr, OutgoingMessage.class);
+
+ assertNotNull(in);
+ assertNotNull(out);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // NOTE Test Mapper will be used to test an event from dmaap.
+ @Test
+ public void testGetterSetter() {
+ assertNotNull(in);
+ assertNotNull(in.getAction());
+ assertNotNull(in.getHeader().getApiVer());
+ assertNotNull(in.getHeader().getOriginatorId());
+ assertNotNull(in.getHeader().getRequestID());
+ assertNotNull(in.getHeader().getSubRequestId());
+ assertNotNull(in.getHeader().getTimeStamp());
+
+ assertNotNull(out);
+ assertNotNull(out.getHeader().getApiVer());
+ assertNotNull(out.getHeader().getOriginatorId());
+ assertNotNull(out.getHeader().getRequestID());
+ assertNotNull(out.getHeader().getSubRequestId());
+ assertNotNull(out.getHeader().getTimeStamp());
+ assertNotNull(out.getStatus().getCode());
+ assertNotNull(out.getStatus().getValue());
+
+ }
+
+ @Test
+ @Ignore
+ public void testIncommingToOutgoing(){
+ OutgoingMessage newOut;
+ newOut = Mapper.mapOne(in.toOutgoing(Status.ACCEPTED), OutgoingMessage.class);
+ assertNotNull(newOut);
+ assertNotNull(newOut.getHeader().getApiVer());
+ assertNotNull(newOut.getHeader().getOriginatorId());
+ assertNotNull(newOut.getHeader().getRequestID());
+ assertNotNull(newOut.getHeader().getSubRequestId());
+ assertNotNull(newOut.getHeader().getTimeStamp());
+ assertNotNull(newOut.getStatus().getCode());
+ assertNotNull(newOut.getStatus().getValue());
+ }
+
+ @Test
+ @Ignore
+ public void testToString() {
+ in = new IncomingMessage();
+ assertNotNull(in.toString());
+ String id = "test";
+ //in.setId(id);
+ assertNotNull(in.toString());
+ assertTrue(in.toString().contains(id));
+ }
+
+
+ @Test
+ @Ignore
+ public void testOutgoingUpdateTime() {
+ //String old = out.getResponseTime();
+ out.updateResponseTime();
+ //assertFalse(old.equals(out.getResponseTime()));
+ }
+
+ // Testing for 1510
+ @Test
+ @Ignore
+ public void testOutgoingToJson() {
+ // Message Set
+ String message = "MSG";
+ //out.setMessage(message);
+ JSONObject json = out.toResponse();
+ assertNotNull(json);
+ String respStr = json.getString("response");
+ //assertTrue(respStr.contains(out.getResponse().getValue()));
+
+ String msgStr = json.getString("message");
+ assertNotNull(msgStr);
+ //assertFalse(msgStr.contains(out.getOriginalRequest())); // False for 1602
+ //assertTrue(msgStr.contains(out.getMessage()));
+
+ // Null Message
+ //out.setMessage(null);
+ json = out.toResponse();
+ assertNotNull(json);
+ msgStr = json.getString("message");
+ assertNotNull(msgStr);
+ //assertFalse(msgStr.contains(out.getOriginalRequest())); // False for 1602
+ //assertTrue(msgStr.contains(out.getResponse().getValue()));
+
+ // Echoing request
+ //assertNotNull(out.getOriginalRequest());
+ }
+
+ @Test
+ @Ignore
+ public void testOutgoingToString() {
+ String s = out.toString();
+ //assertTrue(s.contains(out.getId()));
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/LCM/TestConverter.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/LCM/TestConverter.java
new file mode 100644
index 000000000..2062e80d8
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/LCM/TestConverter.java
@@ -0,0 +1,91 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.junit.Assert;
+import org.junit.Test;
+import org.openecomp.appc.listener.LCM.conv.Converter;
+import org.openecomp.appc.listener.LCM.model.DmaapIncomingMessage;
+import org.openecomp.appc.listener.LCM.model.DmaapOutgoingMessage;
+import org.openecomp.appc.listener.util.Mapper;
+
+public class TestConverter {
+
+ private String jsonInputBodyStr ="{\"input\":{ \"common-header\": { \"timestamp\": \"2016-08-03T08:50:18.97Z\", \"api-ver\": \"1\", \"originator-id\": \"1\", \"request-id\": \"123\", \"sub-request-id\": \"1\", \"flags\": { \"force\":\"TRUE\", \"ttl\":\"9900\" } }, \"action\": \"Stop\", \"action-identifiers\": { \"vnf-id\": \"TEST\" } }}";
+ private String jsonOutputBodyStr ="{\"output\":{\"common-header\":{\"timestamp\":\"2016-08-03T08:50:18.97Z\",\"api-ver\":\"1\",\"flags\":{\"force\":\"TRUE\",\"ttl\":\"9900\"},\"sub-request-id\":\"1\",\"request-id\":\"123\",\"originator-id\":\"1\"},\"status\":{\"value\":\"TestException\",\"code\":200}}}";
+
+ @Test
+ public void buildDmaapOutgoingMessageWithUnexpectedErrorTest() throws JsonProcessingException {
+ DmaapIncomingMessage dmaapIncomingMessage = buildDmaapIncomingMessage();
+ String errMsg = "TestException";
+ DmaapOutgoingMessage dmaapOutgoingMessage = Converter.buildDmaapOutgoingMessageWithUnexpectedError(dmaapIncomingMessage.getBody(), "test", new Exception(errMsg));
+ int code = dmaapOutgoingMessage.getBody().get("output").get("status").get("code").asInt();
+ String value = dmaapOutgoingMessage.getBody().get("output").get("status").get("value").asText();
+ Assert.assertEquals(200,code);
+ Assert.assertEquals(errMsg,value);
+ }
+
+ private static String expectedDmaapOutgoingMessageAsJsonString = "{\"body\":{\"output\":{\"common-header\":{\"timestamp\":\"2016-08-03T08:50:18.97Z\",\"api-ver\":\"1\",\"flags\":{\"force\":\"TRUE\",\"ttl\":\"9900\"},\"sub-request-id\":\"1\",\"request-id\":\"123\",\"originator-id\":\"1\"},\"status\":{\"value\":\"TestException\",\"code\":200}}},\"cambria.partition\":\"MSO\",\"rpc-name\":\"test\"}";
+ @Test
+ public void convDmaapOutgoingMessageToJsonStringTest() throws JsonProcessingException {
+ DmaapOutgoingMessage dmaapOutgoingMessage = buildDmaapOutgoingMessage();
+ String dmaapOutgoingMessageAsJsonString = Converter.convDmaapOutgoingMessageToJsonString(dmaapOutgoingMessage);
+// Assert.assertEquals(dmaapOutgoingMessageAsJsonString,dmaapOutgoingMessageAsJsonString);
+ Assert.assertEquals(expectedDmaapOutgoingMessageAsJsonString,dmaapOutgoingMessageAsJsonString);
+ }
+
+ private DmaapIncomingMessage buildDmaapIncomingMessage() {
+ DmaapIncomingMessage dmaapIncomingMessage = new DmaapIncomingMessage();
+ dmaapIncomingMessage.setRpcName("test");
+ JsonNode jsonNode = Mapper.toJsonNodeFromJsonString(jsonInputBodyStr);
+ dmaapIncomingMessage.setBody(jsonNode);
+ return dmaapIncomingMessage;
+
+ }
+
+ private DmaapOutgoingMessage buildDmaapOutgoingMessage() {
+ DmaapOutgoingMessage dmaapOutgoingMessage = new DmaapOutgoingMessage();
+ dmaapOutgoingMessage.setRpcName("test");
+ JsonNode jsonNode = Mapper.toJsonNodeFromJsonString(jsonOutputBodyStr);
+ dmaapOutgoingMessage.setBody(jsonNode);
+ return dmaapOutgoingMessage;
+
+ }
+
+
+ @Test
+ public void extractRequestIdWithSubIdTest() {
+ DmaapIncomingMessage dmaapIncomingMessage = buildDmaapIncomingMessage();
+ String equestIdWithSubId = Converter.extractRequestIdWithSubId(dmaapIncomingMessage.getBody());
+ Assert.assertEquals("123-1",equestIdWithSubId);
+ }
+
+ @Test
+ public void extractStatusCodeTest() {
+ DmaapOutgoingMessage dmaapOutgoingMessage = buildDmaapOutgoingMessage();
+ Integer statusCode = Converter.extractStatusCode(dmaapOutgoingMessage.getBody());
+ Assert.assertEquals(200L,statusCode.longValue());
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/LCM1607/model/TestJsonGenericMessages.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/LCM1607/model/TestJsonGenericMessages.java
new file mode 100644
index 000000000..203a13dce
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/LCM1607/model/TestJsonGenericMessages.java
@@ -0,0 +1,99 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.LCM1607.model;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+import org.openecomp.appc.listener.util.Mapper;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+
+public class TestJsonGenericMessages {/*
+
+ @Test
+ public void serializeIncomingMessage() {
+
+ final String expectedJson = "{\"CommonHeader\":{\"TimeStamp\":\"2016-05-02 19:50:37.09\",\"TransactionID\":\"1\",\"APIver\":\"1.01\",\"RequestTrack\":[\"1\",\"4\",\"12\",\"3\"],\"Flags\":null,\"SubrequestID\":null,\"OriginatorID\":\"2\"},\"Payload\":\"{ \\\"command\\\": \\\"start\\\", \\\"target-id\\\": \\\"111\\\", \\\"flag10\\\": {\\\"object-1\\\": {\\\"key-1\\\": \\\"key\\\", \\\"value-1\\\": \\\"value\\\" }} }\",\"Action\":\"CONFIGURE\",\"ObjectID\":\"200\",\"TargetID\":\"100\"}";
+ InputBody msg = createIncomingMessage();
+
+ String json = Mapper.toJsonObject(msg).toString();
+ //System.out.println(json);
+ Assert.assertEquals(expectedJson, json);
+ }
+
+ @Test
+ public void deserializeIncomingMessage() throws IOException {
+ final String originalJson = "{\"CommonHeader\":{\"TimeStamp\":\"2016-05-02 19:50:37.09\",\"TransactionID\":\"1\",\"Flags\":{\"FORCE\":\"Y\",\"TTL\":\"12\"},\"SubrequestID\":\"2345\",\"OriginatorID\":\"2\",\"APIver\":\"1.01\"}, \"Payload\": \" \\\"Graceful\\\" : \\\"Yes\\\" \",\"Action\":\"CONFIGURE\",\"ObjectID\":\"200\",\"TargetID\":\"100\"}";
+
+ ObjectMapper mapper = new ObjectMapper();
+ InputBody msg = mapper.readValue(originalJson, InputBody.class);
+
+ Assert.assertNotNull(msg);
+ Assert.assertEquals("2016-05-02 19:50:37.09", msg.getCommonHeader().getTimeStamp());
+ Assert.assertEquals("1", msg.getCommonHeader().getRequestID());
+ Assert.assertEquals("1.01", msg.getCommonHeader().getApiVer());
+ Assert.assertEquals("200", msg.getObjectId());
+ Assert.assertEquals("100", msg.getTargetId());
+ Assert.assertEquals(" \"Graceful\" : \"Yes\" ", msg.getPayload());
+ Assert.assertEquals("CONFIGURE", msg.getAction());
+
+ }
+
+ @Test
+ public void serializeResponseMessage() {
+ InputBody imsg = createIncomingMessage();
+ OutputBody omsg = new OutputBody(imsg);
+ omsg.setStatus(new ResponseStatus("200", "OK"));
+
+ String json = Mapper.toJsonObject(omsg).toString();
+ System.out.println(json);
+ //Assert.assertEquals(expectedJson, json);
+ Assert.assertNotEquals("", json);
+
+ }
+
+ private InputBody createIncomingMessage() {
+ InputBody msg = new InputBody();
+ CommonHeader rh = new CommonHeader();
+ rh.setTimeStamp("2016-05-02 19:50:37.09");
+ rh.setApiVer("1.01");
+ rh.setRequestID("1");
+ rh.setOriginatorId("2");
+
+
+ Map<String, String> flags = new HashMap<>();
+ flags.put("FORCE", "Y");
+ flags.put("TTL", "12");
+
+ msg.setCommonHeader(rh);
+ msg.setAction("CONFIGURE");
+ msg.setTargetId("100");
+ msg.setObjectId("200");
+ msg.setPayloadAsString("{ \"command\": \"start\", \"target-id\": \"111\", \"flag10\": {\"object-1\": {\"key-1\": \"key\", \"value-1\": \"value\" }} }");
+ return msg;
+ }
+*/
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestAbstractListener.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestAbstractListener.java
new file mode 100644
index 000000000..c193f7c2d
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestAbstractListener.java
@@ -0,0 +1,98 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Properties;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.appc.listener.AbstractListener;
+import org.openecomp.appc.listener.ListenerProperties;
+
+public class TestAbstractListener {
+
+ private class DummyListener extends AbstractListener {
+ public DummyListener(ListenerProperties props) {
+ super(props);
+ }
+
+ public boolean getRun() {
+ return run.get();
+ }
+
+ public ThreadPoolExecutor getExecutor() {
+ return executor;
+ }
+ }
+
+ private DummyListener listener;
+ private ListenerProperties props;
+
+ @Before
+ public void setup() throws Exception {
+ Properties regularProps = new Properties();
+ regularProps.load(getClass().getResourceAsStream("/org/openecomp/appc/default.properties"));
+ props = new ListenerProperties("", regularProps);
+ listener = new DummyListener(props);
+ }
+
+ @Test
+ public void testRun() {
+ Thread t = new Thread(listener);
+ t.run();
+ assertFalse(t.isAlive()); // Should die immediately
+ }
+
+ @Test
+ public void testStop() {
+ listener.stop();
+ assertFalse(listener.getRun());
+ assertTrue(listener.getExecutor().isShutdown());
+ }
+
+ @Test
+ public void testStopNow() {
+ listener.stopNow();
+ assertFalse(listener.getRun());
+ assertTrue(listener.getExecutor().isShutdown());
+ }
+
+ @Test
+ public void testBenchmark() {
+ String out = listener.getBenchmark();
+ assertNotNull(out);
+ assertTrue(out.contains(listener.getListenerId()));
+ }
+
+ @Test
+ public void testListenerId() {
+ assertEquals(props.getPrefix(), listener.getListenerId());
+ listener.setListenerId("newId");
+ assertEquals("newId", listener.getListenerId());
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestAppcDmaapListenerActivator.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestAppcDmaapListenerActivator.java
new file mode 100644
index 000000000..a5348d145
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestAppcDmaapListenerActivator.java
@@ -0,0 +1,46 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+import org.junit.Test;
+import org.openecomp.appc.listener.AppcEventListenerActivator;
+
+public class TestAppcDmaapListenerActivator {
+
+ @Test
+ public void testStartStop() {
+ // TODO - How do we tests activators
+ AppcEventListenerActivator appc = new AppcEventListenerActivator();
+ try {
+ appc.start(null);
+ Thread.sleep(2000);
+ appc.stop(null);
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ assertNotNull(appc.getName());
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestListenerProperties.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestListenerProperties.java
new file mode 100644
index 000000000..9a70a8427
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/TestListenerProperties.java
@@ -0,0 +1,154 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Properties;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.appc.listener.AbstractListener;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.ListenerProperties.KEYS;
+
+public class TestListenerProperties {
+
+ private Properties good, bad, both;
+ private String prefix;
+
+ private ListenerProperties props;
+
+ @Before
+ public void setup() {
+ prefix = "test";
+ good = new Properties();
+ bad = new Properties();
+ both = new Properties();
+
+ good.setProperty(String.format("%s.%s", prefix, "a"), "1");
+ good.setProperty(String.format("%s.%s", prefix, "a.b"), "2");
+ good.setProperty(String.format("%s.%s", prefix, "a.b.c"), "3");
+
+ bad.setProperty(prefix, "NA");
+ bad.setProperty(prefix + ".", "NA");
+ bad.setProperty(String.format("%s.%s", prefix + "x", "bad"), "NA");
+ bad.setProperty(String.format("%s.%s", "x" + prefix, "bad"), "NA");
+
+ for (String key : good.stringPropertyNames()) {
+ both.put(key, good.getProperty(key));
+ }
+ for (String key : bad.stringPropertyNames()) {
+ both.put(key, bad.getProperty(key));
+ }
+
+ props = new ListenerProperties(prefix, both);
+ }
+
+ @Test
+ public void testConstructor() {
+ props = new ListenerProperties(prefix, good);
+ assertEquals(prefix, props.getPrefix());
+ assertEquals(good.size(), props.getProperties().size());
+
+ props = new ListenerProperties(prefix, bad);
+ assertEquals(prefix, props.getPrefix());
+ assertTrue(props.getProperties().isEmpty());
+
+ props = new ListenerProperties(prefix, both);
+ assertEquals(prefix, props.getPrefix());
+ assertEquals(good.size(), props.getProperties().size());
+
+ for (Object val : props.getProperties().values()) {
+ assertFalse("NA".equals(val.toString()));
+ }
+
+ assertTrue(props.toString().contains(prefix));
+ }
+
+ @Test
+ public void testGetClass() {
+ assertNull(props.getListenerClass());
+ props.setListenerClass(AbstractListener.class);
+ assertNotNull(props.getListenerClass());
+ assertEquals(AbstractListener.class, props.getListenerClass());
+ }
+
+ @Test
+ public void testMessageServices() {
+ // Hardcode count so tests must be updated when values are added
+ assertEquals(1, ListenerProperties.MessageService.values().length);
+
+ // Bad Input
+ ListenerProperties.MessageService def = ListenerProperties.MessageService.DMaaP;
+ assertEquals(def, ListenerProperties.MessageService.parse(null));
+ assertEquals(def, ListenerProperties.MessageService.parse(""));
+ assertEquals(def, ListenerProperties.MessageService.parse("NotDmaapOrDMaaP"));
+
+
+ // DMaaP case sensitivity
+ assertEquals(ListenerProperties.MessageService.DMaaP, ListenerProperties.MessageService.parse("dmaap"));
+ assertEquals(ListenerProperties.MessageService.DMaaP, ListenerProperties.MessageService.parse("DMAAP"));
+ assertEquals(ListenerProperties.MessageService.DMaaP, ListenerProperties.MessageService.parse("DMaaP"));
+ }
+
+ @Test
+ public void testKeys() {
+ // Hardcode count so tests must be updated when values are added
+ assertEquals(15, ListenerProperties.KEYS.values().length);
+
+ Properties tmp = new Properties();
+ try {
+ tmp.load(getClass().getResourceAsStream("/org/openecomp/appc/default.properties"));
+ } catch (Exception e) {
+ fail("Could not load properties to test");
+ }
+ String realPrefix = tmp.getProperty("test.prefix");
+ assertNotNull(realPrefix);
+ props = new ListenerProperties(realPrefix, tmp);
+
+ for (KEYS key : ListenerProperties.KEYS.values()) {
+ assertNotNull(key.getFullProp(realPrefix));
+ assertNotNull(props.getProperty(key));
+ assertNotNull(props.getProperty(key.getPropertySuffix()));
+ }
+ }
+
+ @Test
+ public void testDisabled() throws Exception {
+ assertFalse(props.isDisabled());
+ props.getProperties().put(KEYS.DISABLED.getPropertySuffix(), "TRUE");
+ assertTrue(props.isDisabled());
+ props.getProperties().put(KEYS.DISABLED.getPropertySuffix(), "N/A");
+ assertFalse(props.isDisabled());
+ props.getProperties().put(KEYS.DISABLED.getPropertySuffix(), "fAlse");
+ assertFalse(props.isDisabled());
+ props.getProperties().remove(KEYS.DISABLED.getPropertySuffix());
+ assertFalse(props.isDisabled());
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestController.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestController.java
new file mode 100644
index 000000000..a67b94647
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestController.java
@@ -0,0 +1,26 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.impl;
+
+public class TestController {
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestEventHandler.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestEventHandler.java
new file mode 100644
index 000000000..788aa961a
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestEventHandler.java
@@ -0,0 +1,171 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.commons.lang3.StringUtils;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.openecomp.appc.listener.EventHandler;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.impl.EventHandlerImpl;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+/**
+ * Test the ProviderAdapter implementation.
+ *
+ */
+
+public class TestEventHandler {
+
+ private ListenerProperties prop;
+
+ private EventHandler adapter;
+
+ private static final String PROP_FILE = "/org/openecomp/appc/default.properties";
+
+ private static final String MESSAGE_FILE = "/DCAEResponse.txt";
+
+ /**
+ * Setup the test environment.
+ *
+ * @throws NoSuchMethodException
+ * @throws SecurityException
+ * @throws NoSuchFieldException
+ */
+ @Before
+ public void setup() {
+ Properties allProps = new Properties();
+ try {
+ allProps.load(getClass().getResourceAsStream(PROP_FILE));
+ allProps.remove("appc.ClosedLoop.topic.read.filter");
+ prop = new ListenerProperties("appc.ClosedLoop", allProps);
+ } catch (IOException e) {
+ System.out.println("WARNING: Failed to load properties file: " + PROP_FILE);
+ }
+ adapter = new EventHandlerImpl(prop);
+ }
+
+ @Test
+ public void testInitialProperties() {
+ assertEquals(prop.getProperty("topic.read"), adapter.getReadTopic());
+ assertTrue(adapter.getWriteTopics().contains(prop.getProperty("topic.write")));
+ assertEquals(prop.getProperty("client.name"), adapter.getClientName());
+ assertEquals(prop.getProperty("client.name.id"), adapter.getClientId());
+
+ String hostStr = prop.getProperty("poolMembers");
+ int hostCount = hostStr.length()>0 ? hostStr.split(",").length : 0;
+ assertEquals(hostCount, adapter.getPool().size());
+ }
+
+ @Test
+ public void testGettersAndSetters() {
+ String readTopic = "read";
+ Set<String> writeTopic = new HashSet<String>();
+ writeTopic.add("write");
+ String clientName = "APPC-TEST";
+ String clientId = "00";
+ String newHost = "google.com";
+
+ adapter.setReadTopic(readTopic);
+ assertEquals(readTopic, adapter.getReadTopic());
+
+ adapter.setWriteTopics(writeTopic);
+ assertEquals(writeTopic, adapter.getWriteTopics());
+
+ adapter.setClientName(clientName);
+ assertEquals(clientName, adapter.getClientName());
+
+ adapter.setClientId(clientId);
+ assertEquals(clientId, adapter.getClientId());
+
+ adapter.setCredentials("fake", "secret");
+ adapter.clearCredentials();
+
+ int oldSize = adapter.getPool().size();
+ adapter.addToPool(newHost);
+ assertEquals(oldSize + 1, adapter.getPool().size());
+ assertTrue(adapter.getPool().contains(newHost));
+
+ adapter.removeFromPool(newHost);
+ assertEquals(oldSize, adapter.getPool().size());
+ assertFalse(adapter.getPool().contains(newHost));
+
+ }
+
+ @Test
+ @Ignore
+ public void testRun() {
+ // Runoff any old data
+ List<String> result1 = adapter.getIncomingEvents();
+ assertNotNull(result1);
+
+ // Post new data
+ DummyObj data = new DummyObj();
+ data.key = "value";
+ adapter.postStatus(data.toJson());
+
+ // Wait to account for network delay
+ sleep(2000);
+
+ // Get data back
+ List<DummyObj> result2 = adapter.getIncomingEvents(DummyObj.class);
+ assertNotNull(result2);
+ assertEquals(1, result2.size());
+ assertEquals(data.toJson(), result2.get(0).toJson());
+ }
+
+ @JsonSerialize
+ public static class DummyObj implements Serializable {
+ @JsonProperty("request") // Call request for default filter
+ public String key;
+
+ public DummyObj() {
+ }
+
+ public String toJson() {
+ return String.format("{\"request\": \"%s\"}", key);
+ }
+ }
+
+ private void sleep(long ms) {
+ try {
+ Thread.sleep(ms);
+ } catch (Exception e) {
+ return;
+ }
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestListener.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestListener.java
new file mode 100644
index 000000000..a1c7917f9
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/impl/TestListener.java
@@ -0,0 +1,112 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.util.Properties;
+
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.openecomp.appc.listener.Listener;
+import org.openecomp.appc.listener.ListenerProperties;
+import org.openecomp.appc.listener.CL.impl.ListenerImpl;
+
+@Ignore
+public class TestListener {
+
+ private static final String PROP_FILE = "/org/openecomp/appc/default.properties";
+
+ private Listener listener;
+
+ private Properties props;
+
+ @Before
+ public void setup() {
+ props = new Properties();
+ try {
+ props.load(getClass().getResourceAsStream(PROP_FILE));
+ props.setProperty("topic.read", "DCAE-CLOSED-LOOP-EVENTS-DEV1510SIM");
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail("Failed to setup test: " + e.getMessage());
+ }
+ listener = new ListenerImpl(new ListenerProperties("appc.ClosedLoop", props));
+ }
+
+ @Test
+ public void testListenerId() {
+ String originalId = listener.getListenerId();
+ String newId = originalId + "-new";
+
+ listener.setListenerId(newId);
+ assertEquals(newId, listener.getListenerId());
+ }
+
+ @Test
+ public void testRun() {
+ try {
+ Thread t = new Thread(listener);
+ t.start();
+
+ Thread.sleep(5000);
+
+ listener.stopNow();
+
+ System.out.println(listener.getBenchmark());
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testUpdateProperties() {
+
+ }
+
+ @Test
+ public void printSampleData() {
+ try {
+ props.setProperty("threads.queuesize.min", "1");
+ props.setProperty("threads.queuesize.max", "1");
+ props.setProperty("threads.poolsize.min", "1");
+ props.setProperty("threads.poolsize.max", "1");
+
+ Thread t = new Thread(listener);
+ t.start();
+
+ Thread.sleep(2000);
+
+ listener.stop();
+
+ System.out.println(listener.getBenchmark());
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/util/TestMapper.java b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/util/TestMapper.java
new file mode 100644
index 000000000..430b28e61
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/java/org/openecomp/appc/listener/util/TestMapper.java
@@ -0,0 +1,104 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * openECOMP : APP-C
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.appc.listener.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.json.JSONObject;
+import org.junit.Before;
+import org.junit.Test;
+import org.openecomp.appc.listener.util.Mapper;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+public class TestMapper {
+
+ private String dummyJson = "{\"a\":\"%s\"}";
+ private DummyObj dummyObj = new DummyObj();
+
+ @JsonSerialize
+ public static class DummyObj implements Serializable {
+ @JsonProperty("a")
+ public String a;
+
+ public DummyObj() {
+ }
+ }
+
+ @Before
+ public void setup() {
+ }
+
+ @Test
+ public void testGetMapper() {
+ assertNotNull(Mapper.getMapper());
+ }
+
+ @Test
+ public void testToJsonObject() {
+ JSONObject out;
+ out = Mapper.toJsonObject(".");
+ assertNull(out);
+
+ String value = "b";
+ out = Mapper.toJsonObject(String.format(dummyJson, value));
+ assertNotNull(out);
+ assertEquals(value, out.get("a"));
+ }
+
+ @Test
+ public void testConstructor() {
+ // Only here for code coverage
+ Mapper m = new Mapper();
+ assertNotNull(m);
+ }
+
+ @Test
+ public void testMap() {
+ List<String> in = new ArrayList<String>();
+ in.add("");
+ in.add(null);
+
+ List<DummyObj> out = Mapper.mapList(in, DummyObj.class);
+ assertNotNull(out);
+ assertTrue(out.isEmpty());
+
+ in.add(String.format(dummyJson, "1"));
+ in.add("{\"invalid\":\"yes\"}");
+ in.add(String.format(dummyJson, "2"));
+
+ out = Mapper.mapList(in, DummyObj.class);
+ assertNotNull(out);
+ assertEquals(2, out.size());
+ assertEquals("1", out.get(0).a);
+ assertEquals("2", out.get(1).a);
+ }
+
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/resources/IncomingMessage.txt b/appc-event-listener/appc-event-listener-bundle/src/test/resources/IncomingMessage.txt
new file mode 100644
index 000000000..0771a790a
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/resources/IncomingMessage.txt
@@ -0,0 +1,24 @@
+{
+"eventID": "SomeEventID-TBD",
+
+"APPCVDCName": "Delete this field not needed by App-C",
+"requestTime": "0000-00-00 00:00:00.000000",
+"policyVersion": "1",
+"VMName": "123",
+"from": "test",
+"msgOid": ".1.3.6.1.4.1.193.183.4.1.3.5.1.4",
+"VMID": "abc12345-1234-5678-890a-abcdefg12345",
+"trapID": "1234567",
+"requestClient": "test",
+"message": "Abnormal condition detected",
+"time": "123567890",
+"policyName": "RESTART",
+"trapIDOID": ".1.3.6.1.4.1.193.183.4.1.3.5.1.3",
+"request": "Restart",
+"OPS_CL_timer": "15",
+"nOID": ".1.3.6.1.4.1.193.183.4.2.0.4",
+"AgentAddress": "192.168.1.2",
+"vmOID": ".1.3.6.1.4.1.193.183.4.1.2.1",
+"TenantID": "abcde12345fghijk6789lmnopq123rst",
+"VServerSelfLink": "http://192.168.1.2:8774/v2/abcde12345fghijk6789lmnopq123rst/servers/abc12345-1234-5678-890a-abcdefg12345"
+} \ No newline at end of file
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/resources/IncomingMessage1607.txt b/appc-event-listener/appc-event-listener-bundle/src/test/resources/IncomingMessage1607.txt
new file mode 100644
index 000000000..667cfae8c
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/resources/IncomingMessage1607.txt
@@ -0,0 +1,24 @@
+{
+ "CommonHeader": {
+ "TimeStamp": "0000-00-00T00:00:00.000Z",
+ "APIver": "1.01",
+ "OriginatorID": "policy.pdp01",
+ "RequestID": "b74d13c5-bb26-4b04-992c-4679dfc8280e",
+ "SubrequestID": "1"
+ },
+ "Action": "RESTART",
+ "Payload": {
+ "VServerSelfLink": "http://192.168.1.2:8774/v2/abcde12345fghijk6789lmnopq123rst/servers/abc12345-1234-5678-890a-abcdefg12345",
+ "VNF_NAME": "test",
+ "VMID": "abc12345-1234-5678-890a-abcdefg12345",
+ "TenantID": "abcde12345fghijk6789lmnopq123rst",
+ "LOC_ID": "Test",
+ "in-maint": "false",
+ "Identity": "http://example.com:5000/v2.0",
+ "Prov_status": "ACTIVE",
+ "OAM_IPV4": "192.168.1.2",
+ "is-closed-loop-disabled": "false",
+ "VM_NAME": "test",
+ "OAM_IPV6": "0000::0000:0000:0000:0000/64"
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/resources/OutgoingMessage.txt b/appc-event-listener/appc-event-listener-bundle/src/test/resources/OutgoingMessage.txt
new file mode 100644
index 000000000..c5d6c5b56
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/resources/OutgoingMessage.txt
@@ -0,0 +1,14 @@
+{
+"eventID": "SomeEventID-TBD",
+
+"VMName": "test",
+"from": "appc@test",
+"message": "Some Text Here",
+"requestClient": "test",
+"requestTime": "0000-00-00 00:00:00.000000",
+"response": "PENDING",
+"responseTime": "0000-00-00 00:00:00.000000",
+
+"policyName": "RESTART",
+"policyVersion": "1"
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/resources/OutgoingMessage1607.txt b/appc-event-listener/appc-event-listener-bundle/src/test/resources/OutgoingMessage1607.txt
new file mode 100644
index 000000000..898eb622e
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/resources/OutgoingMessage1607.txt
@@ -0,0 +1,27 @@
+{
+ "CommonHeader": {
+ "TimeStamp": "0000-00-00T00:00:00.000Z",
+ "APIver": "1.01",
+ "OriginatorID": "policy.pdp01",
+ "RequestID": "b74d13c5-bb26-4b04-992c-4679dfc8280e",
+ "SubrequestID": "1"
+ },
+ "Status": {
+ "Code": 100,
+ "Value": "ACCEPT"
+ },
+ "Payload": {
+ "VServerSelfLink": "http://192.168.1.2:8774/v2/abcde12345fghijk6789lmnopq123rst/servers/abc12345-1234-5678-890a-abcdefg12345",
+ "VNF_NAME": "test",
+ "VMID": "abc12345-1234-5678-890a-abcdefg12345",
+ "TenantID": "abcde12345fghijk6789lmnopq123rst",
+ "LOC_ID": "Test",
+ "in-maint": "false",
+ "Identity": "http://example.com:5000/v2.0",
+ "Prov_status": "ACTIVE",
+ "OAM_IPV4": "192.168.1.2",
+ "is-closed-loop-disabled": "false",
+ "VM_NAME": "test",
+ "OAM_IPV6": "0000::0000:0000:0000:0000/64"
+ }
+}
diff --git a/appc-event-listener/appc-event-listener-bundle/src/test/resources/org/openecomp/appc/default.properties b/appc-event-listener/appc-event-listener-bundle/src/test/resources/org/openecomp/appc/default.properties
new file mode 100644
index 000000000..7a65b7195
--- /dev/null
+++ b/appc-event-listener/appc-event-listener-bundle/src/test/resources/org/openecomp/appc/default.properties
@@ -0,0 +1,89 @@
+###
+# ============LICENSE_START=======================================================
+# openECOMP : APP-C
+# ================================================================================
+# Copyright (C) 2017 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=========================================================
+###
+
+
+
+
+
+### ###
+### ###
+###Properties below are default values to be provided with real valuesin appc.properties###
+### ###
+### ###
+
+
+
+
+
+### ###
+###Closed Loop - properties ###
+### ###
+test.prefix=appc.ClosedLoop
+appc.ClosedLoop.disabled=false
+appc.ClosedLoop.service=UEB
+appc.ClosedLoop.poolMembers=192.168.1.2:3904
+appc.ClosedLoop.topic.read=APPC-TEST1
+appc.ClosedLoop.topic.read.timeout=5
+appc.ClosedLoop.topic.write=APPC-TEST1
+appc.ClosedLoop.topic.read.filter={"class":"Assigned","field":"request"}
+appc.ClosedLoop.client.name=APPC-TEST1-CLOSED-LOOP
+appc.ClosedLoop.client.name.id=0
+
+# KEY AND SECRET BELOW NEED TO BE MODIFIED TO APPROPRIATE VALUES WHEN THE TOPIC IS API-KEY-BASED AUTH
+appc.ClosedLoop.client.key=MY_API_KEY
+appc.ClosedLoop.client.secret=MY_API_SECRET
+
+appc.ClosedLoop.threads.queuesize.min=1
+appc.ClosedLoop.threads.queuesize.max=1000
+appc.ClosedLoop.threads.poolsize.min=1
+appc.ClosedLoop.threads.poolsize.max=2
+appc.ClosedLoop.provider.url=https://admin:password@localhost:8443/restconf/operations/appc-provider:topology-operation
+
+
+
+
+
+### ###
+###Closed Loop - 1607 properties ###
+### ###
+appc.ClosedLoop1607.poolMembers=192.168.1.2:3904
+appc.ClosedLoop1607.topic.read=MY_DMAAP_TOPIC
+appc.ClosedLoop1607.topic.write=MY_DMAAP_TOPIC
+appc.ClosedLoop1607.topic.read.filter={"class":"Unassigned","field":"Status"}
+appc.ClosedLoop1607.client.name=DMAAP-CLIENT-NAME
+appc.ClosedLoop1607.client.name.id=0
+#dmaap.client.key=random
+#dmaap.client.secret=random
+appc.ClosedLoop1607.threads.queuesize.min=1
+appc.ClosedLoop1607.threads.queuesize.max=1000
+appc.ClosedLoop1607.threads.poolsize.min=1
+appc.ClosedLoop1607.threads.poolsize.max=2
+appc.ClosedLoop1607.provider.url=https://admin:password@localhost:8443/restconf/operations/appc-provider:topology-operation
+
+
+
+
+
+### ###
+### POINT TO AN ACTIVE TEST VM IN OPENSTACK INSTANCE ###
+### ###
+test.vm_url=https://example.com/v2/123/servers/123-345
+