Integratation of DMaaP, Mapping File

DMaaP integratation, Mapping File, Initialization of Adapter

Change-Id: I826aa2e64fa7c155f088a7519c24887ce88e2ec4
Issue-ID: DCAEGEN2-335
Signed-off-by: Pooja03 <PM00501616@techmahindra.com>

5 files changed, 365 insertions, 98 deletions

diff --git a/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/DmaapConfig.java b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/DmaapConfig.java
new file mode 100644
index 0000000..34bbc8e
--- /dev/null
+++ b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/DmaapConfig.java
@@ -0,0 +1,288 @@
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 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.onap.universalvesadapter.utils;
+
+import javax.validation.constraints.NotEmpty;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.stereotype.Component;
+
+@Component
+@PropertySource(value = {"classpath:application.properties","classpath:DMaapMR.properties"})
+@ConfigurationProperties
+public class DmaapConfig {
+	
+	@Value("${mr.hostname}")
+    @NotEmpty
+	private String hostname;
+	
+	 // default port number
+	@Value("${mr.DEFAULT_PORT_NUMBER}")
+    @NotEmpty
+	private int DEFAULT_PORT_NUMBER;
+	
+	// default to no username
+	@Value("${mr.DEFAULT_USER_NAME}")
+	private String DEFAULT_USER_NAME;
+			
+			
+	//defaults to no userPassword
+	@Value("${mr.DEFAULT_USER_PASSWORD}")
+	private String DEFAULT_USER_PASSWORD;
+	
+	//defaults to using https protocol
+	@Value("${mr.DEFAULT_PROTOCOL}")
+    @NotEmpty
+	private String DEFAULT_PROTOCOL;
+	
+	//defaults to json content type
+	@Value("${mr.DEFAULT_CONTENT_TYPE}")
+    @NotEmpty
+	private String DEFAULT_CONTENT_TYPE;
+
+	@Value("${mr.DMAAP_URI_PATH_PREFIX}")
+    @NotEmpty
+	private String DMAAP_URI_PATH_PREFIX;
+	
+	@Value("${mr.DMAAP_DEFAULT_CONSUMER_ID}")
+    @NotEmpty
+	private String DMAAP_DEFAULT_CONSUMER_ID;
+			 
+	@Value("${mr.DMAAP_GROUP_PREFIX}")
+    @NotEmpty
+	private String DMAAP_GROUP_PREFIX;
+  
+    // Publisher Constants
+			 
+    //Dmaap Publisher Topic
+	@Value("${mr.publisher.topic}")
+    @NotEmpty		
+	private String publisherTopic;
+			 
+    //disable batching by default
+	@Value("${mr.publisher.DEFAULT_PUBLISHER_MAX_BATCH_SIZE}")
+    @NotEmpty
+	private int publisherDEFAULT_PUBLISHER_MAX_BATCH_SIZE;
+	
+  //default recovery messages size
+	@Value("${mr.publisher.DEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE}")
+    @NotEmpty
+	private int publisherDEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE;
+			  
+//number of retries when flushing messages
+	@Value("${mr.publisher.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE}")
+    @NotEmpty
+	private int publisherPUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE;
+
+  //delay in retrying for flushing messages
+	@Value("${mr.publisher.PUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE}")
+    @NotEmpty	
+	private int publisherPUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE;
+
+  // Subscriber Constants
+	
+  //Dmaap Subcriber Topic
+	@Value("${mr.subscriber.topic}")
+    @NotEmpty
+	private String subscriberTopic;
+	
+	@Value("${mr.subcriber.DEFAULT_SUBSCRIBER_TIMEOUT_MS}")
+    @NotEmpty
+	private int subcriberDEFAULT_SUBSCRIBER_TIMEOUT_MS;
+	
+	@Value("${mr.subcriber.DEFAULT_SUBSCRIBER_MESSAGE_LIMIT}")
+    @NotEmpty
+	private int subcriberDEFAULT_SUBSCRIBER_MESSAGE_LIMIT;
+	
+	@Value("${mr.subcriber.DEFAULT_SUBSCRIBER_GROUP_PREFIX}")
+    @NotEmpty
+	private String subcriberDEFAULT_SUBSCRIBER_GROUP_PREFIX;
+	
+	@Value("${mr.subcriber.SUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME}")
+    @NotEmpty
+	private String subcriberSUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME;
+	
+	@Value("${mr.subcriber.SUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME}")
+    @NotEmpty
+	private String subcriberSUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME;
+
+	public void setHostname(String hostname) {
+		this.hostname = hostname;
+	}
+
+	public String getHostname() {
+		return hostname;
+	}
+
+	public int getDEFAULT_PORT_NUMBER() {
+		return DEFAULT_PORT_NUMBER;
+	}
+
+	public void setDEFAULT_PORT_NUMBER(int dEFAULT_PORT_NUMBER) {
+		DEFAULT_PORT_NUMBER = dEFAULT_PORT_NUMBER;
+	}
+
+	public String getDEFAULT_USER_NAME() {
+		return DEFAULT_USER_NAME;
+	}
+
+	public void setDEFAULT_USER_NAME(String dEFAULT_USER_NAME) {
+		DEFAULT_USER_NAME = dEFAULT_USER_NAME;
+	}
+
+	public String getDEFAULT_USER_PASSWORD() {
+		return DEFAULT_USER_PASSWORD;
+	}
+
+	public void setDEFAULT_USER_PASSWORD(String dEFAULT_USER_PASSWORD) {
+		DEFAULT_USER_PASSWORD = dEFAULT_USER_PASSWORD;
+	}
+
+	public String getDEFAULT_PROTOCOL() {
+		return DEFAULT_PROTOCOL;
+	}
+
+	public void setDEFAULT_PROTOCOL(String dEFAULT_PROTOCOL) {
+		DEFAULT_PROTOCOL = dEFAULT_PROTOCOL;
+	}
+
+	public String getDEFAULT_CONTENT_TYPE() {
+		return DEFAULT_CONTENT_TYPE;
+	}
+
+	public void setDEFAULT_CONTENT_TYPE(String dEFAULT_CONTENT_TYPE) {
+		DEFAULT_CONTENT_TYPE = dEFAULT_CONTENT_TYPE;
+	}
+
+	public String getDMAAP_URI_PATH_PREFIX() {
+		return DMAAP_URI_PATH_PREFIX;
+	}
+
+	public void setDMAAP_URI_PATH_PREFIX(String dMAAP_URI_PATH_PREFIX) {
+		DMAAP_URI_PATH_PREFIX = dMAAP_URI_PATH_PREFIX;
+	}
+
+	public String getDMAAP_DEFAULT_CONSUMER_ID() {
+		return DMAAP_DEFAULT_CONSUMER_ID;
+	}
+
+	public void setDMAAP_DEFAULT_CONSUMER_ID(String dMAAP_DEFAULT_CONSUMER_ID) {
+		DMAAP_DEFAULT_CONSUMER_ID = dMAAP_DEFAULT_CONSUMER_ID;
+	}
+
+	public String getDMAAP_GROUP_PREFIX() {
+		return DMAAP_GROUP_PREFIX;
+	}
+
+	public void setDMAAP_GROUP_PREFIX(String dMAAP_GROUP_PREFIX) {
+		DMAAP_GROUP_PREFIX = dMAAP_GROUP_PREFIX;
+	}
+
+	public String getPublisherTopic() {
+		return publisherTopic;
+	}
+
+	public void setPublisherTopic(String publisherTopic) {
+		this.publisherTopic = publisherTopic;
+	}
+
+	public int getPublisherDEFAULT_PUBLISHER_MAX_BATCH_SIZE() {
+		return publisherDEFAULT_PUBLISHER_MAX_BATCH_SIZE;
+	}
+
+	public void setPublisherDEFAULT_PUBLISHER_MAX_BATCH_SIZE(int publisherDEFAULT_PUBLISHER_MAX_BATCH_SIZE) {
+		this.publisherDEFAULT_PUBLISHER_MAX_BATCH_SIZE = publisherDEFAULT_PUBLISHER_MAX_BATCH_SIZE;
+	}
+
+	public int getPublisherDEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE() {
+		return publisherDEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE;
+	}
+
+	public void setPublisherDEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE(
+			int publisherDEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE) {
+		this.publisherDEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE = publisherDEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE;
+	}
+
+	public int getPublisherPUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE() {
+		return publisherPUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE;
+	}
+
+	public void setPublisherPUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE(int publisherPUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE) {
+		this.publisherPUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE = publisherPUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE;
+	}
+
+	public int getPublisherPUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE() {
+		return publisherPUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE;
+	}
+
+	public void setPublisherPUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE(int publisherPUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE) {
+		this.publisherPUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE = publisherPUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE;
+	}
+
+	public String getSubscriberTopic() {
+		return subscriberTopic;
+	}
+
+	public void setSubscriberTopic(String subscriberTopic) {
+		this.subscriberTopic = subscriberTopic;
+	}
+
+	public int getSubcriberDEFAULT_SUBSCRIBER_TIMEOUT_MS() {
+		return subcriberDEFAULT_SUBSCRIBER_TIMEOUT_MS;
+	}
+
+	public void setSubcriberDEFAULT_SUBSCRIBER_TIMEOUT_MS(int subcriberDEFAULT_SUBSCRIBER_TIMEOUT_MS) {
+		this.subcriberDEFAULT_SUBSCRIBER_TIMEOUT_MS = subcriberDEFAULT_SUBSCRIBER_TIMEOUT_MS;
+	}
+
+	public int getSubcriberDEFAULT_SUBSCRIBER_MESSAGE_LIMIT() {
+		return subcriberDEFAULT_SUBSCRIBER_MESSAGE_LIMIT;
+	}
+
+	public void setSubcriberDEFAULT_SUBSCRIBER_MESSAGE_LIMIT(int subcriberDEFAULT_SUBSCRIBER_MESSAGE_LIMIT) {
+		this.subcriberDEFAULT_SUBSCRIBER_MESSAGE_LIMIT = subcriberDEFAULT_SUBSCRIBER_MESSAGE_LIMIT;
+	}
+
+	public String getSubcriberDEFAULT_SUBSCRIBER_GROUP_PREFIX() {
+		return subcriberDEFAULT_SUBSCRIBER_GROUP_PREFIX;
+	}
+
+	public void setSubcriberDEFAULT_SUBSCRIBER_GROUP_PREFIX(String subcriberDEFAULT_SUBSCRIBER_GROUP_PREFIX) {
+		this.subcriberDEFAULT_SUBSCRIBER_GROUP_PREFIX = subcriberDEFAULT_SUBSCRIBER_GROUP_PREFIX;
+	}
+
+	public String getSubcriberSUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME() {
+		return subcriberSUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME;
+	}
+
+	public void setSubcriberSUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME(String subcriberSUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME) {
+		this.subcriberSUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME = subcriberSUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME;
+	}
+
+	public String getSubcriberSUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME() {
+		return subcriberSUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME;
+	}
+
+	public void setSubcriberSUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME(String subcriberSUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME) {
+		this.subcriberSUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME = subcriberSUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME;
+	}
+
+}
diff --git a/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/HTTPUtils.java b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/HTTPUtils.java
new file mode 100644
index 0000000..02632f4
--- /dev/null
+++ b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/HTTPUtils.java
@@ -0,0 +1,62 @@
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 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.onap.universalvesadapter.utils;
+
+/**
+ * Contains common utils to check HTTP Related Utils
+ *
+ * @author Rajiv Singla . Creation Date: 11/2/2016.
+ */
+public abstract class HTTPUtils {
+
+    /**
+     * HTTP Status code for successful HTTP call
+     */
+    public static final Integer HTTP_SUCCESS_STATUS_CODE = 200;
+
+    /**
+     * HTTP Response code when request has been accepted for processing, but the processing has not been completed
+     */
+    public static final Integer HTTP_ACCEPTED_RESPONSE_CODE = 202;
+
+    /**
+     * HTTP Response code when there is no content
+     */
+    public static final Integer HTTP_NO_CONTENT_RESPONSE_CODE = 204;
+
+
+    public static final String JSON_APPLICATION_TYPE = "application/json";
+
+
+    private HTTPUtils() {
+
+    }
+
+    /**
+     *  Checks if HTTP Status code is less than or equal to 200 but less then 300
+     *
+     * @param statusCode http status code
+     * @return true if response code between 200 and 300
+     */
+    public static boolean isSuccessfulResponseCode(Integer statusCode) {
+        return statusCode >= 200 && statusCode < 300;
+    }
+}
diff --git a/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/MapperConfigUtils.java b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/MapperConfigUtils.java
index 4e67ed6..3d1907a 100644
--- a/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/MapperConfigUtils.java
+++ b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/MapperConfigUtils.java
@@ -19,18 +19,22 @@
 */
 package org.onap.universalvesadapter.utils;
 
-import com.att.aft.dme2.internal.apache.commons.lang3.EnumUtils;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.IOException;
 import java.util.Set;
 import java.util.TreeSet;
+
 import org.onap.universalvesadapter.exception.MapperConfigException;
 import org.onap.universalvesadapter.mappingconfig.Entry;
 import org.onap.universalvesadapter.mappingconfig.Evaluation;
 import org.onap.universalvesadapter.mappingconfig.MapperConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
+ * This class will be used in Future for different telemetry data:
  * This class will be utility class to read the mapper config file and parse the
  * config to prepare the grammar to detect the incoming json's event type.
  * 
@@ -39,6 +43,7 @@ import org.onap.universalvesadapter.mappingconfig.MapperConfig;
  */
 public class MapperConfigUtils {
 
+	private final Logger LOGGER = LoggerFactory.getLogger(MapperConfigUtils.class);
     private static Set<Entry> entries = new TreeSet<>((o1, o2) -> o1.getPriority().compareTo(o2.getPriority()));
 
     private enum JoinOperator {
@@ -212,6 +217,7 @@ public class MapperConfigUtils {
                     exception);
         }
         System.out.println("Read config file content into :" + config);
+        
         if (null != config) {
             entries.addAll(config.getEntries());
         } else {
diff --git a/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/ParallelTasks.java b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/ParallelTasks.java
deleted file mode 100644
index 45fdf96..0000000
--- a/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/ParallelTasks.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
-* ============LICENSE_START=======================================================
-* ONAP : DCAE
-* ================================================================================
-* Copyright 2018 TechMahindra
-*=================================================================================
-* Licensed under the Apache License, Version 2.0 (the "License");
-* you may not use this file except in compliance with the License.
-* You may obtain a copy of the License at
-*
-*     http://www.apache.org/licenses/LICENSE-2.0
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License.
-* ============LICENSE_END=========================================================
-*/
-package org.onap.universalvesadapter.utils;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-/**
- * 
- * Utility class to execute parallel tasks
- * 
- * @author kmalbari
- *
- */
-public class ParallelTasks
-{
-    private final Collection<Runnable> tasks = new ArrayList<Runnable>();
-
-    public ParallelTasks()
-    {
-    }
-
-    /**
-     * 
-     * Add task to be executed in parallel
-     * 
-     * @param task
-     */
-    public void add(final Runnable task)
-    {
-        tasks.add(task);
-    }
-
-    /**
-     * starts all the added tasks in parallel
-     * 
-     * @throws InterruptedException
-     */
-    public void startParallelTasks() throws InterruptedException
-    {
-        final ExecutorService threads = Executors.newFixedThreadPool(Runtime.getRuntime()
-                .availableProcessors());
-        try
-        {
-            final CountDownLatch latch = new CountDownLatch(tasks.size());
-            for (final Runnable task : tasks)
-                threads.execute(new Runnable() {
-                    public void run()
-                    {
-                        try
-                        {
-                            task.run();
-                        }
-                        finally
-                        {
-                            latch.countDown();
-                        }
-                    }
-                });
-            latch.await();
-        }
-        finally
-        {
-            threads.shutdown();
-        }
-    }
-}
\ No newline at end of file
diff --git a/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/SmooksUtils.java b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/SmooksUtils.java
index 8c17dc2..1c38783 100644
--- a/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/SmooksUtils.java
+++ b/UniversalVesAdapter/src/main/java/org/onap/universalvesadapter/utils/SmooksUtils.java
@@ -38,6 +38,8 @@ import org.slf4j.LoggerFactory;
  * @author kmalbari
  *
  */
+
+
 public class SmooksUtils {
 
 	
@@ -53,23 +55,19 @@ public class SmooksUtils {
 	 */
 	public static VesEvent getTransformedObjectForInput(Smooks smooks, String incomingJsonString) {
 		
-		LOGGER.debug("Transforming json " + incomingJsonString);
+		LOGGER.info("Transforming incoming json " );
 		ExecutionContext executionContext = smooks.createExecutionContext();
-		LOGGER.debug("Context created");
+		LOGGER.info("Context created");
 		Locale defaultLocale = Locale.getDefault();
         Locale.setDefault(new Locale("en", "IE"));
 
         StringResult result = new StringResult();
 
-        // Configure the execution context to generate a report...
-//        executionContext.setEventListener(new HtmlReportGenerator("target/report/report.html"));
-
-        // Filter the input message to the outputWriter, using the execution context...
         smooks.filterSource(executionContext, new StreamSource(new ByteArrayInputStream(incomingJsonString.getBytes(StandardCharsets.UTF_8))), result);
-        LOGGER.debug("Transformed incoming json now");
+       
         Locale.setDefault(defaultLocale);
         VesEvent vesEvent = (VesEvent) executionContext.getBeanContext().getBean("vesEvent");
-        LOGGER.debug("Converted vesEvent from incoming json"); 
+        LOGGER.debug("consversion successful to VES Event"); 
         
 		return vesEvent;
 	}