Init dcae.collectors.ves

Change-Id: Ic112f107512ac263431613cfc60f848b854fa84b
Signed-off-by: lj1412 <lji@research.att.com>

17 files changed, 1839 insertions, 0 deletions

diff --git a/src/assembly/dep.xml b/src/assembly/dep.xml
new file mode 100644
index 0000000..77fb3f2
--- /dev/null
+++ b/src/assembly/dep.xml
@@ -0,0 +1,57 @@
+<!--

+  ============LICENSE_START=======================================================
+  PROJECT
+  ================================================================================
+  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=========================================================

+  -->

+

+<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2" 

+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

+  xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd">

+  <id>bundle</id>

+  <formats>

+    <format>tar.gz</format>

+  </formats>

+  <files>

+  	<file>

+  		<source>target/${project.artifactId}-${project.version}.jar</source>

+  		<outputDirectory>lib</outputDirectory>

+  	</file>

+  </files>

+  <fileSets>

+    <fileSet>

+        <directory>src/main/scripts</directory>

+        <outputDirectory>bin</outputDirectory>

+        <includes>

+        	<include>**/*.sh</include>

+        </includes>

+        <fileMode>0755</fileMode>

+        <lineEnding>unix</lineEnding>

+    </fileSet>

+    <fileSet>

+        <directory>etc</directory>

+        <outputDirectory>etc</outputDirectory>

+    </fileSet>

+  </fileSets>

+  <dependencySets>

+  	<dependencySet>

+  		<includes>

+  			<include>*:jar</include>

+  		</includes>

+  		<outputDirectory>lib</outputDirectory>

+  	</dependencySet>

+  </dependencySets>

+</assembly>

diff --git a/src/main/java/org/openecomp/dcae/commonFunction/CommonStartup.java b/src/main/java/org/openecomp/dcae/commonFunction/CommonStartup.java
new file mode 100644
index 0000000..7c1ff22
--- /dev/null
+++ b/src/main/java/org/openecomp/dcae/commonFunction/CommonStartup.java
@@ -0,0 +1,324 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.commonFunction;

+

+

+

+import java.io.IOException;

+

+import java.net.URL;

+import java.nio.charset.Charset;

+import java.nio.file.Files;

+import java.nio.file.Paths;

+import java.util.Iterator;

+import java.util.LinkedList;

+import java.util.List;

+import java.util.Map;

+

+import java.util.Queue;

+import java.util.concurrent.LinkedBlockingQueue;

+

+import javax.servlet.ServletException;

+

+import org.apache.catalina.LifecycleException;

+import org.json.JSONArray;

+import org.json.JSONException;

+import org.json.JSONObject;

+import org.openecomp.dcae.restapi.RestfulCollectorServlet;

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+

+import com.att.nsa.apiServer.ApiServer;

+import com.att.nsa.apiServer.ApiServerConnector;

+import com.att.nsa.apiServer.endpoints.NsaBaseEndpoint;

+import com.att.nsa.cmdLine.NsaCommandLineUtil;

+import com.att.nsa.drumlin.service.framework.DrumlinServlet;

+import com.att.nsa.drumlin.till.nv.rrNvReadable;

+import com.att.nsa.drumlin.till.nv.impl.nvPropertiesFile;

+import com.att.nsa.drumlin.till.nv.impl.nvReadableStack;

+import com.att.nsa.drumlin.till.nv.impl.nvReadableTable;

+import com.att.nsa.drumlin.till.nv.rrNvReadable.invalidSettingValue;

+import com.att.nsa.drumlin.till.nv.rrNvReadable.loadException;

+import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;

+import com.fasterxml.jackson.core.JsonParseException;

+import com.fasterxml.jackson.databind.JsonNode;

+import com.github.fge.jsonschema.exceptions.ProcessingException;

+import com.github.fge.jsonschema.main.JsonSchema;

+import com.github.fge.jsonschema.main.JsonSchemaFactory;

+import com.github.fge.jsonschema.report.ProcessingMessage;

+import com.github.fge.jsonschema.report.ProcessingReport;

+import com.github.fge.jsonschema.util.JsonLoader;

+

+

+public class CommonStartup extends NsaBaseEndpoint implements Runnable

+{

+	public static final String kConfig = "c";

+

+	public static final String kSetting_Port = "collector.service.port";

+	public static final int kDefault_Port = 8080;

+	

+	public static final String kSetting_SecurePort = "collector.service.secure.port";

+	public static final int kDefault_SecurePort = -1; 

+	

+	public static final String kSetting_KeystorePassfile = "collector.keystore.passwordfile";

+	public static final String kDefault_KeystorePassfile = "../etc/passwordfile";	

+	public static final String kSetting_KeystoreFile = "collector.keystore.file.location";

+	public static final String kDefault_KeystoreFile = "../etc/keystore";

+	public static final String kSetting_KeyAlias = "collector.keystore.alias";

+	public static final String kDefault_KeyAlias = "tomcat";

+

+	public static final String kSetting_ProcessingConfigs = "collector.hpprocessing";

+	protected static final String[] kDefault_ProcessingConfigs = new String[] { "etc/HPProcessingConfig.json" };

+

+	public static final String kSetting_MaxQueuedEvents = "collector.inputQueue.maxPending";

+	public static final int kDefault_MaxQueuedEvents = 1024*4;

+	

+	public static final String kSetting_schemaValidator = "collector.schema.checkflag";

+	public static final int kDefault_schemaValidator = -1; 

+	

+	public static final String kSetting_schemaFile = "collector.schema.file";

+	public static final String kSetting_ExceptionConfig = "exceptionConfig";

+	

+	public static final String kSetting_dmaapStreamid = "collector.dmaap.streamid";

+	

+	public static final String kSetting_authflag = "header.authflag";

+	public static final int kDefault_authflag = 0;

+	

+	public static final String kSetting_authid = "header.authid";

+	public static final String kSetting_authpwd = "header.authpwd";

+	public static final String kSetting_authstore = "header.authstore";

+	public static final String kSetting_authlist = "header.authlist";

+	

+	

+	

+	public static final Logger inlog = LoggerFactory.getLogger ("org.openecomp.dcae.commonFunction.input" );

+	public static final Logger oplog = LoggerFactory.getLogger ("org.openecomp.dcae.commonFunction.output");

+	public static final Logger eplog = LoggerFactory.getLogger ("org.openecomp.dcae.commonFunction.error");

+	public static final Logger metriclog = LoggerFactory.getLogger ("com.att.ecomp.metrics" );

+	

+	public static int schema_Validatorflag = -1;

+	public static int authflag = 1;

+	public static String schemaFile = null;

+	public static String exceptionConfig = null;

+	public static String cambriaConfigFile = null;

+	private boolean listnerstatus = false;

+	static String streamid = null;

+	

+	private CommonStartup(rrNvReadable settings) throws loadException, missingReqdSetting, IOException, rrNvReadable.missingReqdSetting, rrNvReadable.invalidSettingValue, ServletException, InterruptedException

+	{

+		final List<ApiServerConnector> connectors = new LinkedList<ApiServerConnector> ();

+

+		if (settings.getInt ( kSetting_Port, kDefault_Port ) > 0)

+		{

+			// http service

+				connectors.add (

+					new ApiServerConnector.Builder ( settings.getInt ( kSetting_Port, kDefault_Port ) )

+					.secure ( false )

+					.build ()

+				);

+		}

+

+		// optional https service		

+		final int securePort = settings.getInt(kSetting_SecurePort, kDefault_SecurePort);

+		final String keystoreFile = settings.getString(kSetting_KeystoreFile, kDefault_KeystoreFile);

+		final String keystorePasswordFile = settings.getString(kSetting_KeystorePassfile, kDefault_KeystorePassfile);

+		final String keyAlias = settings.getString (kSetting_KeyAlias, kDefault_KeyAlias);

+		

+		

+		if (securePort > 0)

+		{

+			final String kSetting_KeystorePass = readFile(keystorePasswordFile, Charset.defaultCharset());

+				connectors.add(new ApiServerConnector.Builder(securePort)

+					.secure(true)

+					.keystorePassword(kSetting_KeystorePass)

+					.keystoreFile(keystoreFile)

+					.keyAlias(keyAlias)

+					.build());

+

+		}

+

+		//Reading other config properties

+		

+		schema_Validatorflag = settings.getInt(kSetting_schemaValidator, kDefault_schemaValidator );

+		if (schema_Validatorflag > 0){

+			schemaFile =  settings.getString(kSetting_schemaFile,null);	

+		}

+		exceptionConfig = settings.getString(kSetting_ExceptionConfig, null);

+		authflag = settings.getInt(CommonStartup.kSetting_authflag, CommonStartup.kDefault_authflag );

+		String [] currentconffile = settings.getStrings (CommonStartup.kSetting_ProcessingConfigs,	CommonStartup.kDefault_ProcessingConfigs ) ;

+		cambriaConfigFile= currentconffile[0] ;

+		streamid = settings.getString(kSetting_dmaapStreamid,null);

+

+		fTomcatServer = new ApiServer.Builder(connectors, new RestfulCollectorServlet(settings))

+			.encodeSlashes(true)

+			.name("collector")

+			.build();

+		

+		

+		//Load override exception map 

+		CustomExceptionLoader.LoadMap();

+		setListnerstatus(true);

+	}

+

+	public static void main ( String[] args )

+	{

+		try

+		{

+			// process command line arguments

+			final Map<String, String> argMap = NsaCommandLineUtil.processCmdLine ( args, true );

+			final String config = NsaCommandLineUtil.getSetting ( argMap, kConfig, "collector.properties" );

+			final URL settingStream = DrumlinServlet.findStream ( config, CommonStartup.class );

+

+			final nvReadableStack settings = new nvReadableStack ();

+			settings.push ( new nvPropertiesFile ( settingStream ) );

+			settings.push ( new nvReadableTable ( argMap ) );

+			

+			fProcessingInputQueue = new LinkedBlockingQueue<JSONObject> (CommonStartup.kDefault_MaxQueuedEvents);

+			CommonStartup cs= new CommonStartup ( settings );

+

+			Thread csmain = new Thread(cs);

+			csmain.start();

+			

+			EventProcessor ep = new EventProcessor ();

+			Thread epThread=new Thread(ep);

+		    epThread.start();

+			

+			//cs.startAndAwait ();

+			

+		}

+		catch ( loadException | missingReqdSetting  | IOException | invalidSettingValue | ServletException | InterruptedException e )

+		{

+			CommonStartup.eplog.error("FATAL_STARTUP_ERROR" + e.getMessage() );

+			throw new RuntimeException ( e );

+		}

+	}

+

+	public void run() {

+		try {

+			fTomcatServer.start ();

+		} catch (LifecycleException | IOException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		}

+		fTomcatServer.await ();

+	}

+

+    public boolean isListnerstatus() {

+		return listnerstatus;

+	}

+

+	public void setListnerstatus(boolean listnerstatus) {

+		this.listnerstatus = listnerstatus;

+	}

+		public static Queue<JSONObject> getProcessingInputQueue ()

+	{

+		return fProcessingInputQueue;

+	}

+	

+	public static class QueueFullException extends Exception

+	{

+		private static final long serialVersionUID = 1L;

+	}

+	

+	

+	public static void handleEvents ( JSONArray a ) throws QueueFullException, JSONException, IOException

+	{

+		final Queue<JSONObject> queue = getProcessingInputQueue ();

+		try

+		{

+			

+			CommonStartup.metriclog.info("EVENT_PUBLISH_START" );

+			for (int i = 0; i < a.length(); i++) {

+					if ( !queue.offer ( a.getJSONObject(i) ) )	{

+						throw new QueueFullException ();

+					}

+					

+			} 

+			log.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");

+			CommonStartup.metriclog.info("EVENT_PUBLISH_END");

+			//ecomplogger.debug(secloggerMessageEnum.SEC_COLLECT_AND_PULIBISH_SUCCESS);

+

+		}

+		catch ( JSONException e ){

+			throw e;

+			

+		}

+	}

+

+		

+	static String readFile(String path, Charset encoding) 

+			  throws IOException 

+			{

+			    byte[] encoded = Files.readAllBytes(Paths.get(path));

+			    String pwd = new String(encoded);

+			    return pwd.substring(0,pwd.length()-1);

+			}

+

+	

+	public static String schemavalidate( String jsonData, String jsonSchema) {

+	    ProcessingReport report = null;

+	    String result = "false";

+	    

+	    try {

+	        //System.out.println("Applying schema: @<@<"+jsonSchema+">@>@ to data: #<#<"+jsonData+">#>#");

+	    	log.trace("Schema validation for event:" + jsonData);

+	        JsonNode schemaNode = JsonLoader.fromString(jsonSchema);

+	        JsonNode data = JsonLoader.fromString(jsonData);         

+	        JsonSchemaFactory factory = JsonSchemaFactory.byDefault(); 

+	        JsonSchema schema = factory.getJsonSchema(schemaNode);

+	        report = schema.validate(data);

+	    } catch (JsonParseException e) {

+	    	log.error("schemavalidate:JsonParseException for event:" + jsonData );

+	    	System.out.println(e.getMessage());

+	        return e.getMessage().toString();

+	    } catch (ProcessingException e) {  

+	    	log.error("schemavalidate:Processing exception for event:" + jsonData );

+	    	System.out.println(e.getMessage());

+	    	return e.getMessage().toString();

+	    } catch (IOException e) {

+	        log.error("schemavalidate:IO exception; something went wrong trying to read json data for event:" + jsonData);

+	        System.out.println(e.getMessage());

+	        return e.getMessage().toString();

+	    }

+	    if (report != null) {

+	        Iterator<ProcessingMessage> iter = report.iterator();

+	        while (iter.hasNext()) {

+	            ProcessingMessage pm = iter.next();

+	            log.trace("Processing Message: "+pm.getMessage());

+	        }

+	        result = String.valueOf(report.isSuccess());

+	    }

+	    try {

+	    log.trace("Validation Result:" +result + " Validation report:" + report);

+	    }

+	    catch (NullPointerException e){

+	    	log.error("schemavalidate:NullpointerException on report");

+	    }

+	    return result;

+	}

+

+	

+

+	static LinkedBlockingQueue<JSONObject> fProcessingInputQueue;

+	private static ApiServer fTomcatServer = null;

+	private static final Logger log = LoggerFactory.getLogger ( CommonStartup.class );

+}

diff --git a/src/main/java/org/openecomp/dcae/commonFunction/CustomExceptionLoader.java b/src/main/java/org/openecomp/dcae/commonFunction/CustomExceptionLoader.java
new file mode 100644
index 0000000..0adf7b4
--- /dev/null
+++ b/src/main/java/org/openecomp/dcae/commonFunction/CustomExceptionLoader.java
@@ -0,0 +1,121 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.commonFunction;

+

+import java.io.FileNotFoundException;

+import java.io.FileReader;

+import java.util.HashMap;

+

+import java.util.Map.Entry;

+

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+import com.google.gson.JsonArray;

+import com.google.gson.JsonElement;

+import com.google.gson.JsonIOException;

+import com.google.gson.JsonObject;

+import com.google.gson.JsonParser;

+import com.google.gson.JsonSyntaxException;

+

+

+public class CustomExceptionLoader {

+

+    public static HashMap<String, JsonArray> map = null;

+    private static final Logger log = LoggerFactory.getLogger ( CustomExceptionLoader.class );

+    

+    //For standalone test

+    //LoadMap Invoked from servletSetup

+    /*

+	public static void main(String[] args) {

+

+		System.out.println("CustomExceptionLoader.main --> Arguments -- ExceptionConfig file: " + args[0] + "StatusCode:" + args[1]+ " Error Msg:" + args[2]);

+		CommonStartup.exceptionConfig = args[0];

+		

+		//Read the Custom exception JSON file into map

+	    LoadMap();

+	    System.out.println("CustomExceptionLoader.main --> Map info post LoadMap:" + map);

+	   

+	     String[] str= LookupMap(args[1],args[2]);

+			if (! (str==null)) {

+			System.out.println("CustomExceptionLoader.main --> Return from lookup function" + str[0] + "value:" + str[1]);	

+		}

+	    

+	}

+	*/

+	

+	public static void LoadMap () {

+		

+		 map = new HashMap<String, JsonArray>();

+		 

+		 try {

+			 	JsonElement root = null;

+			 	root = new JsonParser().parse(new FileReader(CommonStartup.exceptionConfig));

+			 	JsonObject jsonObject = root.getAsJsonObject().get("code").getAsJsonObject();

+

+				for (Entry<String, JsonElement> entry : jsonObject.entrySet()) {

+				     map.put(entry.getKey(), (JsonArray) entry.getValue());

+				}

+				

+				log.debug("CustomExceptionLoader.LoadMap --> Map loaded - " + map);

+			} catch (JsonIOException e) {

+				e.printStackTrace();

+			} catch (JsonSyntaxException e) {

+				e.printStackTrace();

+			} catch (FileNotFoundException e) {

+				e.printStackTrace();

+			} catch (Exception e) {

+				e.printStackTrace();

+			}

+	}

+

+	public static String[] LookupMap (String error, String errormsg) {

+		 

+		 String[] retarray = null;

+		 

+		 log.debug("CustomExceptionLoader.LookupMap -->" + " HTTP StatusCode:" + error + " Msg:" + errormsg);

+		 try{

+			 

+			 JsonArray jarray = map.get(error);

+			  for (int i = 0; i < jarray.size(); i++) {

+			    	  

+			     JsonElement val = jarray.get(i).getAsJsonObject().get("Reason");

+			     JsonArray ec = (JsonArray) jarray.get(i).getAsJsonObject().get("ErrorCode");

+				 log.trace("CustomExceptionLoader.LookupMap Parameter -> Error msg : " + errormsg + " Reason text being matched:" + val);			

+				 if (errormsg.contains(val.toString().replace("\"", ""))){

+					 log.trace("CustomExceptionLoader.LookupMap Successful! Exception matched to error message StatusCode:" + ec.get(0).toString() + "ErrorMessage:" + ec.get(1).toString());

+					 retarray = new String[2];

+					 retarray[0]=ec.get(0).toString();

+					 retarray[1]=ec.get(1).toString();

+					 return retarray;

+				 }

+			    }

+  

+		 }

+		 catch (Exception e)

+		 {

+			 System.out.println(e.getMessage());

+		 }

+	        

+		 return retarray;

+	}

+

+}

diff --git a/src/main/java/org/openecomp/dcae/commonFunction/DmaapPropertyReader.java b/src/main/java/org/openecomp/dcae/commonFunction/DmaapPropertyReader.java
new file mode 100644
index 0000000..18e6d59
--- /dev/null
+++ b/src/main/java/org/openecomp/dcae/commonFunction/DmaapPropertyReader.java
@@ -0,0 +1,107 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.commonFunction;

+

+

+import java.io.FileNotFoundException;

+import java.io.FileReader;

+

+import java.util.HashMap;

+

+

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+import com.google.gson.JsonArray;

+import com.google.gson.JsonElement;

+import com.google.gson.JsonIOException;

+

+import com.google.gson.JsonParser;

+import com.google.gson.JsonSyntaxException;

+

+

+

+public class DmaapPropertyReader {

+

+	private static DmaapPropertyReader instance = null;

+

+	   

+	   private static final Logger log = LoggerFactory.getLogger ( DmaapPropertyReader.class );

+       public  HashMap<String, String> dmaap_hash = new HashMap<String, String>();

+	   

+	   private DmaapPropertyReader(String CambriaConfigFile) {

+		   

+		    try {

+		    	JsonElement root = null;

+				root = new JsonParser().parse(new FileReader(CambriaConfigFile));

+				JsonArray jsonObject = (JsonArray) root.getAsJsonObject().get("channels");

+				

+				  for (int i = 0; i < jsonObject.size(); i++) {

+					  log.debug("TOPIC:" + jsonObject.get(i).getAsJsonObject().get("cambria.topic") +

+							    " HOST-URL:" + jsonObject.get(i).getAsJsonObject().get("cambria.url") + 

+							    " HOSTS:" + jsonObject.get(i).getAsJsonObject().get("cambria.hosts") + 

+							    " PWD:" + jsonObject.get(i).getAsJsonObject().get("basicAuthPassword") + 

+							    " USER:" + jsonObject.get(i).getAsJsonObject().get("basicAuthUsername") + 

+							    " NAME:" + jsonObject.get(i).getAsJsonObject().get("name") );

+					  

+					  String convertedname = jsonObject.get(i).getAsJsonObject().get("name").toString().replace("\"","");

+					  dmaap_hash.put(convertedname + ".cambria.topic", jsonObject.get(i).getAsJsonObject().get("cambria.topic").toString().replace("\"","") );

+					  

+					  if (jsonObject.get(i).getAsJsonObject().get("cambria.hosts") != null)

+					  {

+						  dmaap_hash.put(convertedname + ".cambria.hosts", jsonObject.get(i).getAsJsonObject().get("cambria.hosts").toString().replace("\"","") );

+					  }

+					  if (jsonObject.get(i).getAsJsonObject().get("cambria.url") != null)

+					  {

+						  dmaap_hash.put(convertedname + ".cambria.url", jsonObject.get(i).getAsJsonObject().get("cambria.url").toString().replace("\"","") );

+					  }

+					  if (jsonObject.get(i).getAsJsonObject().get("basicAuthPassword") != null)

+					  {

+						  dmaap_hash.put(convertedname + ".basicAuthPassword", jsonObject.get(i).getAsJsonObject().get("basicAuthPassword").toString().replace("\"","") );

+					  }	  

+					  if (jsonObject.get(i).getAsJsonObject().get("basicAuthUsername") != null)

+					  {

+						  dmaap_hash.put(convertedname+ ".basicAuthUsername", jsonObject.get(i).getAsJsonObject().get("basicAuthUsername").toString().replace("\"","") );

+					  }

+

+				}

+			} catch (JsonIOException | JsonSyntaxException | FileNotFoundException e1) {

+				e1.printStackTrace();

+				log.error("Problem loading Dmaap Channel configuration file: " +e1.toString());

+			}

+		   	

+			

+	   }

+	   

+	 

+

+	   public static synchronized DmaapPropertyReader getInstance(String ChannelConfig){

+	       if (instance == null) {

+	           instance = new DmaapPropertyReader(ChannelConfig);

+	       }

+	       return instance;

+	   }

+	   

+	   

+	   public String getKeyValue(String HashKey){

+	       return this.dmaap_hash.get(HashKey);

+	   }

+}

diff --git a/src/main/java/org/openecomp/dcae/commonFunction/EventProcessor.java b/src/main/java/org/openecomp/dcae/commonFunction/EventProcessor.java
new file mode 100644
index 0000000..0e6f7e7
--- /dev/null
+++ b/src/main/java/org/openecomp/dcae/commonFunction/EventProcessor.java
@@ -0,0 +1,64 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.commonFunction;
+
+import org.json.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EventProcessor implements Runnable {
+	private static final Logger log = LoggerFactory.getLogger(EventProcessor.class);
+	private JSONObject event = null;
+
+	public EventProcessor() {
+		log.debug("EventProcessor: Default Constructor");
+	}
+
+	@Override
+	public void run() {
+
+		try {
+			event = CommonStartup.fProcessingInputQueue.take();
+			log.info("EventProcessor\tRemoving element: " + event);
+
+			while (event != null) {
+				// As long as the producer is running,
+				// we remove elements from the queue.
+
+				// log.info("EventProcessor\tRemoving element: " +
+				// this.queue.remove());
+
+				if (CommonStartup.streamid == null) {
+					log.error("No StreamID defined for publish - Message dropped" + event.toString());
+				} else {
+					EventPublisher.getInstance(CommonStartup.cambriaConfigFile, CommonStartup.streamid)
+							.sendEvent(event.toString(), CommonStartup.streamid);
+				}
+				log.debug("Message published" + event.toString());
+				event = CommonStartup.fProcessingInputQueue.take();
+			}
+		} catch (InterruptedException e) {
+			log.error("EventProcessor InterruptedException" + e.getMessage());
+		}
+
+	}
+
+}
diff --git a/src/main/java/org/openecomp/dcae/commonFunction/EventPublisher.java b/src/main/java/org/openecomp/dcae/commonFunction/EventPublisher.java
new file mode 100644
index 0000000..4aa6da4
--- /dev/null
+++ b/src/main/java/org/openecomp/dcae/commonFunction/EventPublisher.java
@@ -0,0 +1,136 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.commonFunction;

+

+import java.io.IOException;

+import java.util.List;

+import java.util.concurrent.TimeUnit;

+

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+import java.security.GeneralSecurityException;

+import java.net.MalformedURLException;

+

+import com.att.nsa.cambria.client.CambriaBatchingPublisher;

+import com.att.nsa.cambria.client.CambriaClientBuilders;

+

+

+public class EventPublisher {

+

+	private static EventPublisher instance = null;

+	private static CambriaBatchingPublisher pub = null;

+	

+	private String streamid = "";

+	private static Logger log = LoggerFactory.getLogger(EventPublisher.class.getName());

+	

+	

+

+	private EventPublisher(String CambriaConfigFile, String newstreamid) {

+		

+		this.streamid = newstreamid;

+		try { 

+			String basicAuthUsername = DmaapPropertyReader.getInstance(CambriaConfigFile).getKeyValue(streamid+".basicAuthUsername");

+			if (basicAuthUsername != null)

+			{

+				//log.debug(streamid+".cambria.url"  + streamid+".cambria.topic");

+				log.debug("URL:" + DmaapPropertyReader.getInstance(CambriaConfigFile).getKeyValue(streamid+".cambria.url") + "TOPIC:" + DmaapPropertyReader.getInstance(CambriaConfigFile).getKeyValue(streamid+".cambria.topic") +  "AuthUser:" + DmaapPropertyReader.getInstance(CambriaConfigFile).getKeyValue(streamid+".basicAuthUsername") +  "Authpwd:" + DmaapPropertyReader.getInstance(CambriaConfigFile).getKeyValue(streamid+".basicAuthPassword"));

+		

+				pub = new CambriaClientBuilders.PublisherBuilder ()

+				 .usingHosts (DmaapPropertyReader.getInstance(CambriaConfigFile).dmaap_hash.get(streamid+".cambria.url"))

+				 .onTopic (DmaapPropertyReader.getInstance(CambriaConfigFile).dmaap_hash.get(streamid+".cambria.topic"))

+				 .usingHttps()

+				 .authenticatedByHttp ( DmaapPropertyReader.getInstance(CambriaConfigFile).dmaap_hash.get(streamid+".basicAuthUsername"), DmaapPropertyReader.getInstance(CambriaConfigFile).dmaap_hash.get(streamid+".basicAuthPassword") )

+				 .build ();

+			} 

+			else

+			{

+				//log.debug(streamid+".cambria.url"  + streamid+".cambria.topic");

+				log.debug("URL:" + DmaapPropertyReader.getInstance(CambriaConfigFile).getKeyValue(streamid+".cambria.url") + "TOPIC:" + DmaapPropertyReader.getInstance(CambriaConfigFile).getKeyValue(streamid+".cambria.topic"));

+				

+				

+				pub = new CambriaClientBuilders.PublisherBuilder ()

+						 .usingHosts (DmaapPropertyReader.getInstance(CambriaConfigFile).dmaap_hash.get(streamid+".cambria.hosts"))

+						 .onTopic (DmaapPropertyReader.getInstance(CambriaConfigFile).dmaap_hash.get(streamid+".cambria.topic"))

+						 .build ();

+							

+			}

+		}

+		catch(GeneralSecurityException | MalformedURLException e ) {

+			log.error("CambriaClientBuilders connection exception : " + e.getMessage());

+		} 

+		catch(Exception e) {

+			log.error("CambriaClientBuilders connection exception : " + e.getMessage());

+		}

+		         

+	}

+	

+	public static synchronized EventPublisher getInstance( String CambriaConfigFile, String streamid){

+	       if (instance == null) {

+	           instance = new EventPublisher(CambriaConfigFile, streamid);

+	       }

+	       return instance;

+	      

+		}

+	

+	public synchronized void sendEvent(String event, String newstreamid ) {

+	

+		//Check if streamid changed

+		if(! newstreamid.equals(this.streamid)) { 

+			closePublisher();

+			instance = new EventPublisher (CommonStartup.cambriaConfigFile,  newstreamid);

+		}

+

+

+		try {

+			int pendingMsgs = pub.send("MyPartitionKey", event.toString());

+			

+			if(pendingMsgs > 100) {

+				log.info("Pending Message Count="+pendingMsgs);

+			}

+		

+			CommonStartup.oplog.info ("Event Published:" + event);

+		} catch(IOException ioe) {

+			log.error("Unable to publish event:" + event + " Exception:" + ioe.toString()); 

+		}

+

+		

+		

+		

+	}

+

+

+    public synchronized void closePublisher() {

+		

+		try { 

+			final List<?> stuck = pub.close(20, TimeUnit.SECONDS);

+			if ( stuck.size () > 0 ) { 

+				log.error(stuck.size() + " messages unsent" ); 

+			}

+		}

+		catch(InterruptedException ie) {

+			log.error("Caught an Interrupted Exception on Close event");

+		}catch(IOException ioe) {

+			log.error("Caught IO Exception: " + ioe.toString()); 

+		}

+		

+	}

+}

diff --git a/src/main/java/org/openecomp/dcae/restapi/RestfulCollectorServlet.java b/src/main/java/org/openecomp/dcae/restapi/RestfulCollectorServlet.java
new file mode 100644
index 0000000..bd9be55
--- /dev/null
+++ b/src/main/java/org/openecomp/dcae/restapi/RestfulCollectorServlet.java
@@ -0,0 +1,146 @@
+

+/*

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.restapi;

+

+import java.io.IOException;

+import java.net.URL;

+

+import javax.servlet.ServletException;

+

+import org.apache.tomcat.util.codec.binary.Base64;

+import org.openecomp.dcae.commonFunction.CommonStartup;

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+import com.att.nsa.apiServer.CommonServlet;

+import com.att.nsa.configs.ConfigDbException;

+import com.att.nsa.drumlin.service.framework.DrumlinErrorHandler;

+import com.att.nsa.drumlin.service.framework.context.DrumlinRequestContext;

+import com.att.nsa.drumlin.service.framework.routing.DrumlinRequestRouter;

+import com.att.nsa.drumlin.service.framework.routing.playish.DrumlinPlayishRoutingFileSource;

+import com.att.nsa.drumlin.service.standards.HttpStatusCodes;

+import com.att.nsa.drumlin.till.nv.rrNvReadable;

+import com.att.nsa.drumlin.till.nv.rrNvReadable.loadException;

+import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;

+import com.att.nsa.security.NsaAuthenticator;

+

+import com.att.nsa.security.authenticators.SimpleAuthenticator;

+import com.att.nsa.security.db.simple.NsaSimpleApiKey;

+

+public class RestfulCollectorServlet extends CommonServlet

+{

+	String authid = null;

+	String authpwd = null;

+	String authlist = null;

+	public RestfulCollectorServlet ( rrNvReadable settings ) throws loadException, missingReqdSetting

+	{

+		super ( settings, "collector", false );

+				authid =  settings.getString(CommonStartup.kSetting_authid,null);

+				if (authid != null)

+				{

+					String authpwdtemp = settings.getString(CommonStartup.kSetting_authpwd,null);

+					authpwd =  new String(Base64.decodeBase64(authpwdtemp));

+				}

+				authlist = settings.getString(CommonStartup.kSetting_authlist,null);

+	}

+

+	

+	/**

+	 * This is called once at server start. Use it to init any shared objects and setup the route mapping.

+	 */

+	@Override

+	protected void servletSetup () throws rrNvReadable.missingReqdSetting, rrNvReadable.invalidSettingValue, ServletException

+	{

+		super.servletSetup ();

+

+		try

+		{

+			// the base class provides a bunch of things like API authentication and ECOMP compliant

+			// logging. The Restful Collector likely doesn't need API authentication, so for now,

+			// we init the base class services with an in-memory (and empty!) config DB.

+			commonServletSetup ( ConfigDbType.MEMORY );

+			

+

+

+			// setup the servlet routing and error handling

+			final DrumlinRequestRouter drr = getRequestRouter ();

+

+			// you can tell the request router what to do when a particular kind of exception is thrown.

+			drr.setHandlerForException( IllegalArgumentException.class, new DrumlinErrorHandler()

+			{

+				@Override

+				public void handle ( DrumlinRequestContext ctx, Throwable cause )

+				{

+					sendJsonReply ( ctx, HttpStatusCodes.k400_badRequest, cause.getMessage() );

+				}

+			});

+

+			// load the routes from the config file

+			final URL routes = findStream ( "routes.conf" );

+			if ( routes == null ) throw new rrNvReadable.missingReqdSetting ( "No routing configuration." );

+			final DrumlinPlayishRoutingFileSource drs = new DrumlinPlayishRoutingFileSource ( routes );

+			drr.addRouteSource ( drs );

+

+			

+			NsaAuthenticator<NsaSimpleApiKey> NsaAuth = new SimpleAuthenticator ();

+			if (authlist != null)

+			{

+				String authpair[] = authlist.split("\\|");

+				for (String pair: authpair) {           

+						String lineid[] = pair.split(",");

+						String listauthid =  lineid[0];	

+						String listauthpwd =  new String(Base64.decodeBase64(lineid[1]));

+						((SimpleAuthenticator) NsaAuth).add(listauthid,listauthpwd);

+			    }

+				

+			}

+			else if (authid != null)

+			{

+				((SimpleAuthenticator) NsaAuth).add(authid,authpwd);

+			}

+			else

+			{

+				//add a default test account

+				((SimpleAuthenticator) NsaAuth).add("admin","collectorpasscode");

+			}

+			this.getSecurityManager().addAuthenticator(NsaAuth);

+			log.info ( "Restful Collector Servlet is up." );

+		}

+		catch ( SecurityException e )

+		{

+			throw new ServletException ( e );

+		}

+		catch ( IOException e )

+		{

+			throw new ServletException ( e );

+		}

+		catch ( ConfigDbException e )

+		{

+			throw new ServletException ( e );

+		}

+	}

+

+	

+

+	private static final long serialVersionUID = 1L;

+	private static final Logger log = LoggerFactory.getLogger ( RestfulCollectorServlet.class );

+}

diff --git a/src/main/java/org/openecomp/dcae/restapi/endpoints/EventReceipt.java b/src/main/java/org/openecomp/dcae/restapi/endpoints/EventReceipt.java
new file mode 100644
index 0000000..54512e7
--- /dev/null
+++ b/src/main/java/org/openecomp/dcae/restapi/endpoints/EventReceipt.java
@@ -0,0 +1,243 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.restapi.endpoints;

+

+

+import java.io.FileReader;

+import java.io.IOException;

+

+import org.json.JSONArray;

+import org.json.JSONException;

+import org.json.JSONObject;

+import org.json.JSONTokener;

+import org.openecomp.dcae.commonFunction.CommonStartup;

+import org.openecomp.dcae.commonFunction.CustomExceptionLoader;

+import org.openecomp.dcae.commonFunction.CommonStartup.QueueFullException;

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+import com.att.nsa.apiServer.endpoints.NsaBaseEndpoint;

+import com.att.nsa.drumlin.service.framework.context.DrumlinRequestContext;

+import com.att.nsa.drumlin.service.standards.HttpStatusCodes;

+import com.att.nsa.drumlin.service.standards.MimeTypes;

+import com.att.nsa.security.db.simple.NsaSimpleApiKey;

+

+import com.google.gson.JsonParser;

+

+

+public class EventReceipt extends NsaBaseEndpoint {

+	static String valresult = null;

+	static JSONObject customerror = null;

+

+	public static void receiveSingleEvent(DrumlinRequestContext ctx) throws IOException {

+

+		NsaSimpleApiKey retkey = null;

+		JSONObject jsonObject = null;

+		//String br = new BufferedReader(new InputStreamReader(ctx.request().getBodyStream())).readLine();

+		

+		try {

+			

+

+			//JsonElement msg = new JsonParser().parse(new BufferedReader(new InputStreamReader(ctx.request().getBodyStream())).readLine());

+			jsonObject = new JSONObject ( new JSONTokener ( ctx.request ().getBodyStream () ) );

+

+			CommonStartup.inlog.info("Input Messsage: " + jsonObject);

+			log.info("Input Messsage: " + jsonObject);

+			

+

+			try {

+

+				if (CommonStartup.authflag == 1) {

+					retkey = NsaBaseEndpoint.getAuthenticatedUser(ctx);

+				}

+			} catch (NullPointerException x) {

+

+				log.info("Invalid user request " + ctx.request().getContentType() + " Message:" + jsonObject.toString());

+				CommonStartup.eplog.info("EVENT_RECEIPT_FAILURE: Unauthorized user" + x.toString() );

+				respondWithCustomMsginJson(ctx, HttpStatusCodes.k401_unauthorized, "Invalid user");

+				return;

+

+			}

+			if (retkey != null || CommonStartup.authflag == 0) {

+

+				if (CommonStartup.schema_Validatorflag > 0) {

+					String schema = new JsonParser().parse(new FileReader(CommonStartup.schemaFile)).toString();

+

+					valresult = CommonStartup.schemavalidate(jsonObject.toString(), schema);

+					if (valresult.equals("true")) {

+						log.info("Validation successful");

+					} else if (valresult.equals("false")) {

+						log.info("Validation failed");

+						respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest, "Schema validation failed");

+						return;

+					} else {

+						log.error("Validation errored" + valresult);

+						respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest,"Couldn't parse JSON object");

+						return;

+					}

+				}

+

+				// reject anything that's not JSON

+				if (!ctx.request().getContentType().equalsIgnoreCase("application/json")) {

+					log.info("Rejecting request with content type " + ctx.request().getContentType() + " Message:"

+							+ jsonObject);

+					respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest,

+							"Incorrect message content-type; only accepts application/json messages");

+					return;

+				}

+				final JSONArray jsonArray = new JSONArray().put(jsonObject);

+

+				CommonStartup.handleEvents(jsonArray);

+			} else {

+				log.info("Unauthorized request " + ctx.request().getContentType() + " Message:" + jsonObject.toString());

+				respondWithCustomMsginJson(ctx, HttpStatusCodes.k401_unauthorized, "Unauthorized user");

+				return;

+			}

+

+		} catch (JSONException | NullPointerException | IOException x) {

+			log.error("Couldn't parse JSON Array - HttpStatusCodes.k400_badRequest" + HttpStatusCodes.k400_badRequest

+					+ " Message:" + x.getMessage());

+			CommonStartup.eplog.info("EVENT_RECEIPT_FAILURE: Invalid user request " + x.toString() );

+			respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest, "Couldn't parse JSON object");

+			return;

+		} catch (QueueFullException e) {

+			e.printStackTrace();

+			log.error("Collector internal queue full  :" + e.getMessage());

+			CommonStartup.eplog.info("EVENT_RECEIPT_FAILURE: QueueFull" + e.toString() );

+			respondWithCustomMsginJson(ctx, HttpStatusCodes.k503_serviceUnavailable, "Queue full");

+			return;

+		}

+		log.info("MessageAccepted and k200_ok to be sent");

+		ctx.response().sendErrorAndBody(HttpStatusCodes.k200_ok, "Message Accepted", MimeTypes.kAppJson);

+	}

+

+	public static void receiveMultipleEvents(DrumlinRequestContext ctx) throws IOException {

+		// the request body carries events. assume for now it's an array

+		// of json objects that fits in memory. (See cambria's parsing for handling large messages)

+

+		NsaSimpleApiKey retkey = null;

+		

+		JSONArray jsonArray = null;

+		

+		try {

+

+			//String br = new BufferedReader(new InputStreamReader(ctx.request().getBodyStream())).readLine();

+			//JsonElement msg = new JsonParser().parse(new BufferedReader(new InputStreamReader(ctx.request().getBodyStream())).readLine());

+			jsonArray = new JSONArray ( new JSONTokener ( ctx.request ().getBodyStream () ) );

+			

+			CommonStartup.inlog.info("Input Messsage: " + jsonArray);

+			log.info("Input Messsage: " + jsonArray);

+			

+			try {

+				if (CommonStartup.authflag == 1) {

+					retkey = NsaBaseEndpoint.getAuthenticatedUser(ctx);

+				}

+			} catch (NullPointerException x) {

+				log.info("Invalid user request " + ctx.request().getContentType() + " Message:" + jsonArray.toString());

+				CommonStartup.eplog.info("EVENT_RECEIPT_FAILURE: Unauthorized user" + x.toString() );

+				respondWithCustomMsginJson(ctx, HttpStatusCodes.k401_unauthorized, "Invalid user");

+				return;

+			}

+

+			if (retkey != null || CommonStartup.authflag == 0) {

+				if (CommonStartup.schema_Validatorflag > 0) {

+

+					String schema = new JsonParser().parse(new FileReader(CommonStartup.schemaFile)).toString();

+					

+					for (int i = 0; i < jsonArray.length(); i++) {

+						valresult = CommonStartup.schemavalidate(jsonArray.getJSONObject(i).toString(), schema);

+						if (valresult.equals("false")) {

+							log.info("Validation failed");

+							respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest,"Standard schema validation failed");

+							return;

+						} else if (!valresult.equals("true")) {

+							log.error("Validation errored" + valresult);

+							respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest,"Couldn't parse JSON object");

+							return;

+

+						}

+					}

+					log.info("Validation successful for all events in batch");

+

+				}

+				// reject anything that's not JSON

+				if (!ctx.request().getContentType().equalsIgnoreCase("application/json")) {

+					log.info("Rejecting request with content type " + ctx.request().getContentType() + " Message:"

+							+ jsonArray.toString());

+					respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest,

+							"Incorrect message content-type; only accepts application/json messages");

+					return;

+				}

+			

+				CommonStartup.handleEvents(jsonArray);

+			} else {

+				log.info("Unauthorized request " + ctx.request().getContentType() + " Message:" + jsonArray.toString());

+				respondWithCustomMsginJson(ctx, HttpStatusCodes.k401_unauthorized, "Unauthorized request");

+				return;

+			}

+		} catch (JSONException | NullPointerException | IOException x) {

+			log.error("Couldn't parse JSON Array - HttpStatusCodes.k400_badRequest" + HttpStatusCodes.k400_badRequest

+					+ " Message:" + x.getMessage());

+			CommonStartup.eplog.info("EVENT_RECEIPT_FAILURE: Invalid user request " + x.toString() );

+			respondWithCustomMsginJson(ctx, HttpStatusCodes.k400_badRequest, "Couldn't parse JSON object");

+			return;

+		} catch (QueueFullException e) {

+			e.printStackTrace();

+			log.error("Collector internal HP queue full  :" + e.getMessage()	);

+			CommonStartup.eplog.info("EVENT_RECEIPT_FAILURE: QueueFull" + e.toString() );

+			respondWithCustomMsginJson(ctx, HttpStatusCodes.k503_serviceUnavailable, "Queue full");

+			return;

+		}

+

+		ctx.response().sendErrorAndBody(HttpStatusCodes.k200_ok, "Message Accepted", MimeTypes.kAppJson);

+	}

+

+	public static void respondWithCustomMsginJson(DrumlinRequestContext ctx, int sc, String msg) {

+		String[] str = null;

+		String ExceptionType = "GeneralException";

+

+		str = CustomExceptionLoader.LookupMap(String.valueOf(sc), msg);

+		System.out.println("Post CustomExceptionLoader.LookupMap" + str);

+

+		if (str != null) {

+

+			if (str[0].matches("SVC")) {

+				ExceptionType = "ServiceException";

+			} else if (str[1].matches("POL")) {

+				ExceptionType = "PolicyException";

+			}

+

+			JSONObject jb = new JSONObject().put("requestError",

+					new JSONObject().put(ExceptionType, new JSONObject().put("MessagID", str[0]).put("text", str[1])));

+

+			log.debug("Constructed json error : " + jb.toString());

+			ctx.response().sendErrorAndBody(sc, jb.toString(), MimeTypes.kAppJson);

+		} else {

+			JSONObject jb = new JSONObject().put("requestError",

+					new JSONObject().put(ExceptionType, new JSONObject().put("Status", sc).put("Error", msg)));

+			ctx.response().sendErrorAndBody(sc, jb.toString(), MimeTypes.kAppJson);

+		}

+

+	}

+

+	private static final Logger log = LoggerFactory.getLogger(EventReceipt.class);

+

+}

diff --git a/src/main/java/org/openecomp/dcae/restapi/endpoints/Ui.java b/src/main/java/org/openecomp/dcae/restapi/endpoints/Ui.java
new file mode 100644
index 0000000..3fbf325
--- /dev/null
+++ b/src/main/java/org/openecomp/dcae/restapi/endpoints/Ui.java
@@ -0,0 +1,34 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.restapi.endpoints;

+

+import java.io.IOException;

+

+import com.att.nsa.apiServer.endpoints.NsaBaseEndpoint;

+import com.att.nsa.drumlin.service.framework.context.DrumlinRequestContext;

+

+public class Ui extends NsaBaseEndpoint

+{

+	public static void hello ( DrumlinRequestContext ctx ) throws IOException

+	{

+		ctx.renderer ().renderTemplate ( "templates/hello.html" );

+	}

+}

diff --git a/src/main/resources/routes.conf b/src/main/resources/routes.conf
new file mode 100644
index 0000000..2a5cba0
--- /dev/null
+++ b/src/main/resources/routes.conf
@@ -0,0 +1,36 @@
+package org.openecomp.dcae.restapi.endpoints
+
+#
+#	We need to deprecate the original non-versioned paths and use /v1/ for them.
+#	Non-versioned paths will be supported "permanently."
+#
+
+#
+#	post events
+#
+POST	/eventListener/v1.1				EventReceipt.receiveSingleEvent
+POST	/eventListener/v1.1/eventBatch	EventReceipt.receiveMultipleEvents
+POST	/eventListener/v1				EventReceipt.receiveSingleEvent
+POST	/eventListener/v1/eventBatch	EventReceipt.receiveMultipleEvents
+#POST	/eventListener/v1/{topic}		EventReceipt.receiveEventsForTopic
+
+
+
+###############################################################################
+#
+#	UI routes don't need to be versioned
+#
+
+#
+#	UI
+#
+GET	/							Ui.hello
+
+
+# typical static file paths
+GET		/css/				staticDir:css
+GET		/js/				staticDir:js
+GET		/images/			staticDir:images
+GET		/font/				staticDir:font
+GET		/favicon.ico		staticFile:images/attLogo.gif
+GET		/font-awesome/		staticDir:font-awesome
diff --git a/src/main/resources/templates/hello.html b/src/main/resources/templates/hello.html
new file mode 100644
index 0000000..3c2b806
--- /dev/null
+++ b/src/main/resources/templates/hello.html
@@ -0,0 +1,27 @@
+<!--

+  ============LICENSE_START=======================================================
+  PROJECT
+  ================================================================================
+  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=========================================================

+  -->

+

+#set($tab="")

+#parse("header.html")

+

+		<h1>RESTful Collector API</h1>

+		<p>This is a RESTful Collector API server.</p>

+

+#parse("footer.html")

diff --git a/src/main/scripts/SErestfulCollector.sh b/src/main/scripts/SErestfulCollector.sh
new file mode 100644
index 0000000..9d39c16
--- /dev/null
+++ b/src/main/scripts/SErestfulCollector.sh
@@ -0,0 +1,122 @@
+#!/bin/sh
+
+###
+# ============LICENSE_START=======================================================
+# PROJECT
+# ================================================================================
+# 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=========================================================
+###
+
+usage() {
+        echo "SErestfulCollector.sh <start/stop>"
+}
+
+
+collector_start() {
+        collectorPid=`pgrep -f org.openecomp.dcae.commonFunction`
+
+        if [ ! -z "$collectorPid" ]; then
+                echo  "WARNING: Restful Standard Event Collector already running as PID $collectorPid";
+                echo  "Startup Aborted!!!"
+                exit 1
+        fi
+
+        # move into base directory
+        BASEDIR=`dirname $0`
+        cd $BASEDIR/..
+
+        # use JAVA_HOME if provided
+        if [ -z "$JAVA_HOME" ]; then
+                echo "ERROR: JAVA_HOME not setup"
+                echo "Startup Aborted!!"
+                exit 1
+                #JAVA=java
+        else
+                JAVA=$JAVA_HOME/bin/java
+        fi
+
+        # run java. The classpath is the etc dir for config files, and the lib dir
+        # for all the jars.
+        nohup $JAVA -cp "etc${PATHSEP}lib/*" $JAVA_OPTS $MAINCLASS $* &
+        if [ $? -ne 0 ]; then
+                echo "Restful Standard Event Collector has been started!!!"
+        fi
+
+
+}
+
+collector_stop() {
+         collectorPid=`pgrep -f org.openecomp.dcae.commonFunction`
+         if [ ! -z "$collectorPid" ]; then
+                echo "Stopping PID $collectorPid"
+                
+                kill -9 $collectorPid
+                sleep 5
+                if [ ! "$(pgrep -f org.openecomp.dcae.commonFunction)" ]; then
+                         echo "Restful Standard Event Collector has been stopped!!!"
+                else
+                         echo "Restful Standard Event Collector is being stopped!!!"
+                fi
+         else
+                echo  "WARNING: No Restful Standard Event Collector is currently running";
+                exit 1
+         fi
+
+
+}
+
+## Check usage
+if [ $# -ne 1 ]; then
+        usage
+        exit
+fi
+
+
+## Pre-setting
+
+MAINCLASS=org.openecomp.dcae.commonFunction.CommonStartup
+
+# determin a path separator that works for this platform
+PATHSEP=":"
+case "$(uname -s)" in
+
+        Darwin)
+                ;;
+
+         Linux)
+                ;;
+
+         CYGWIN*|MINGW32*|MSYS*)
+                PATHSEP=";"
+                ;;
+
+        *)
+                ;;
+esac
+
+
+
+case $1 in
+        "start")
+                collector_start
+                ;;
+        "stop")
+                collector_stop
+                ;;
+        *)
+                usage
+                ;;
+esac
diff --git a/src/test/java/org/openecomp/dcae/sectest/InputJsonValidation.java b/src/test/java/org/openecomp/dcae/sectest/InputJsonValidation.java
new file mode 100644
index 0000000..73ee612
--- /dev/null
+++ b/src/test/java/org/openecomp/dcae/sectest/InputJsonValidation.java
@@ -0,0 +1,169 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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.dcae.sectest;

+import java.io.BufferedReader;

+import java.io.FileNotFoundException;

+import java.io.FileReader;

+import java.io.IOException;

+import java.io.InputStreamReader;

+

+import org.json.simple.JSONObject;

+import org.json.simple.parser.JSONParser;

+import org.junit.Test;

+import org.openecomp.dcae.commonFunction.CommonStartup;

+

+import com.google.gson.JsonIOException;

+import com.google.gson.JsonParser;

+import com.google.gson.JsonSyntaxException;

+

+

+public class InputJsonValidation {

+

+

+	static String valresult = null;

+

+

+	@Test

+	public void nonvalidJSONValidation(){

+

+		JSONObject jsonObject = null;

+		JSONParser parser=new JSONParser();

+		Object obj=null;

+		//String jsonfilepath="C:/Users/vv770d/git/restfulcollector/src/test/resources/fujistu_non_valid_json.txt";

+		String jsonfilepath="src/test/resources/VES_invalid.txt";

+		String retValue="false";

+		try{

+

+			obj=parser.parse(new FileReader(jsonfilepath));

+

+		}

+		catch(Exception e){

+

+			System.out.println("Exception while opening the file");

+

+		}

+		jsonObject=(JSONObject) obj;

+

+		String schema=null;

+		try {

+			schema = new JsonParser().parse(new FileReader(VESCollectorJunitTest.schemaFile)).toString();

+			//System.out.println("Schema value: " + schema.toString());

+		} catch (JsonIOException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		} catch (JsonSyntaxException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		} catch (FileNotFoundException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		}

+

+		if (schema!=null){

+			retValue=CommonStartup.schemavalidate(jsonObject.toString(), schema);	

+		}

+		//return retValue;

+		VESCollectorJunitTest.output = retValue;

+	}

+	

+	

+	// The below test case meant for verifying json schema on provided json file

+	@Test

+	public void validJSONValidation(){

+

+		JSONObject jsonObject = null;

+		JSONParser parser=new JSONParser();

+		Object obj=null;

+

+		String jsonfilepath="src/test/resources/VES_valid.txt";

+		String retValue="false";

+		try{

+

+			obj=parser.parse(new FileReader(jsonfilepath));

+

+

+		}

+		catch(Exception e){

+			System.out.println("Exception while opening the file");

+

+		}

+		jsonObject=(JSONObject) obj;

+		String schema=null;

+		try {

+

+			schema = new JsonParser().parse(new FileReader(VESCollectorJunitTest.schemaFile)).toString();

+		} catch (JsonIOException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		} catch (JsonSyntaxException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		} catch (FileNotFoundException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		}

+

+		if (schema!=null){

+			retValue=CommonStartup.schemavalidate(jsonObject.toString(), schema);	

+		}

+		VESCollectorJunitTest.output = retValue; 

+		//return retValue;

+	}  

+

+	

+	

+	//validating valid json reception and its posting to DMAP.

+	@Test

+	public void eventReception(){

+

+

+		String testCurlCommand = "curl -i -X POST -d @C:/Users/vv770d/git/restfulcollector/src/test/resources/fujistu-3.txt --header \"Content-Type: application/json\" http://localhost:8080/eventListener/v1";

+

+		//final Process terminal = curlCommand.start();

+		try {

+			Process p = Runtime.getRuntime().exec(testCurlCommand);

+			BufferedReader stdInput = new BufferedReader(new 

+					InputStreamReader(p.getInputStream()));

+

+			BufferedReader stdError = new BufferedReader(new 

+					InputStreamReader(p.getErrorStream()));

+

+			// read the output from the command

+

+			String s = null;

+			while ((s = stdInput.readLine()) != null) {

+				if (s.contains("HTTP/1.1 200 OK")){

+

+					//return "true";

+					VESCollectorJunitTest.output = "true";

+				}

+

+			}

+		} catch (IOException e) {

+			// TODO Auto-generated catch block

+

+

+			e.printStackTrace();

+		}

+		

+		//return "false";

+	}

+

+}  

diff --git a/src/test/java/org/openecomp/dcae/sectest/VESCollectorJunitTest.java b/src/test/java/org/openecomp/dcae/sectest/VESCollectorJunitTest.java
new file mode 100644
index 0000000..e95b099
--- /dev/null
+++ b/src/test/java/org/openecomp/dcae/sectest/VESCollectorJunitTest.java
@@ -0,0 +1,109 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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=========================================================

+ */

+

+/*

+ * 

+ * Purpose: CommonCollectorJunitTest is the wrapper class to invoke all prescribed Junit test cases.

+ * 

+ */

+package org.openecomp.dcae.sectest;

+import static org.junit.Assert.assertEquals;

+

+import java.io.FileInputStream;

+import java.io.FileNotFoundException;

+import java.io.IOException;

+import java.io.InputStream;

+import java.util.Properties;

+

+import org.junit.Test;

+

+public class VESCollectorJunitTest {

+		public static String schemaFile=null;

+		public static String output; 

+

+

+

+	String message = "true";	

+	InputJsonValidation messageUtil = new InputJsonValidation();

+

+	@Test

+	public void validJSONValidation() {

+		

+		Properties prop = new Properties();

+		InputStream input = null;

+		output = "true";

+		try {

+			input = new FileInputStream("etc/collector.properties");

+			try {

+				prop.load(input);

+				schemaFile=prop.getProperty("collector.schema.file");

+

+				System.out.println( "Schema file location: "+ schemaFile);

+			} catch (IOException e) {

+				// TODO Auto-generated catch block

+				e.printStackTrace();

+			}

+		} catch (FileNotFoundException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		}

+		

+		assertEquals(message,output);

+	}

+	

+	

+	@Test

+	public void nonvalidJSONValidation() {

+		output = "false";

+		Properties prop = new Properties();

+		InputStream input = null;

+		try {

+			input = new FileInputStream("etc/collector.properties");

+			try {

+				prop.load(input);

+				schemaFile=prop.getProperty("collector.schema.file");

+

+				System.out.println( "Schema file location: "+ schemaFile);

+			} catch (IOException e) {

+				// TODO Auto-generated catch block

+				e.printStackTrace();

+			}

+		} catch (FileNotFoundException e) {

+			// TODO Auto-generated catch block

+			e.printStackTrace();

+		}		

+		//assertEquals("false",messageUtil.nonvalidJSONValidation());

+		assertEquals("false",output);

+	}

+	

+	

+	//The test case requires common collector running in the environment prior to start execution of JUNIT test cases

+	/*

+	@Test

+	public void testValidJSONObjectReception() {

+		

+		

+		assertEquals("true",messageUtil.eventReception());

+		assertEquals("true",output);

+	}*/

+	

+	

+

+}

diff --git a/src/test/java/org/openecomp/dcae/sectest/VESCollectorJunitTestRunner.java b/src/test/java/org/openecomp/dcae/sectest/VESCollectorJunitTestRunner.java
new file mode 100644
index 0000000..b5e8715
--- /dev/null
+++ b/src/test/java/org/openecomp/dcae/sectest/VESCollectorJunitTestRunner.java
@@ -0,0 +1,52 @@
+/*-

+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * 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=========================================================

+ */

+

+/*

+ * Purpose: CommonCollectorJunitTestRunner is the main class where test suit execution starts its test cases execution

+ * the common collector test suit has been written in order to incorporate functional and logical testing of collector features

+ * 

+ */

+

+package org.openecomp.dcae.sectest;

+import org.junit.runner.JUnitCore;

+import org.junit.runner.Result;

+import org.junit.runner.RunWith;

+import org.junit.runner.notification.Failure;

+import org.openecomp.dcae.commonFunction.CommonStartup;

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+public class VESCollectorJunitTestRunner {

+

+	public static void main(String[] args) {

+

+		System.out.println("STARTING TEST SUITE EXECUTION.....");

+		

+		Result result = JUnitCore.runClasses(VESCollectorJunitTest.class);

+

+		for (Failure failure : result.getFailures()) {

+			System.out.println(failure.toString());

+		}

+

+		

+		System.out.println("Execution Final result : "+result.wasSuccessful());

+	}

+	

+}

diff --git a/src/test/resources/VES_invalid.txt b/src/test/resources/VES_invalid.txt
new file mode 100644
index 0000000..48bffd3
--- /dev/null
+++ b/src/test/resources/VES_invalid.txt
@@ -0,0 +1,46 @@
+{

+"event": {

+"commonEventHeader": {

+"sourceId": "owb-rdm-003",

+"startEpochMicrosec": 1469528413000,

+"eventId": "owb-rdm-003 1",

+"reportingEntityId": "127.0.1.1",

+"eventType": "fault_owb-rdm-003_lossOfSignal",

+"priority": "High",

+"version": 1.0,

+"reportingEntityName": "agilevm",

+"sequence": 0,

+"domain": "fault",

+"functionalRole": "rdm",

+"lastEpochMicrosec": 1469528413000,

+"test": "Dummy"

+},

+"faultFields": {

+"eventSeverity": "CRITICAL",

+"alarmCondition": "lossOfSignal",

+"faultFieldsVersion": 1.0,

+"specificProblem": "lossOfSignal",

+"alarmInterfaceA": "1/0/0 E1",

+"alarmAdditionalInformation": [

+{

+"name": "DIR",

+"value": "tx"

+},

+{

+"name": "LOC",

+"value": "NEND"

+},

+{

+"name": "TYPE",

+"value": "communication"

+},

+{

+"name": "CKTID",

+"value": "circuit-1"

+}

+],

+"eventSourceType": "port",

+"vfStatus": "Active"

+}

+}

+}

diff --git a/src/test/resources/VES_valid.txt b/src/test/resources/VES_valid.txt
new file mode 100644
index 0000000..6473937
--- /dev/null
+++ b/src/test/resources/VES_valid.txt
@@ -0,0 +1,46 @@
+{

+"event": {

+"commonEventHeader": {

+"sourceId": "owb-rdm-003",

+"startEpochMicrosec": 1469528413000,

+"eventId": "owb-rdm-003 1",

+"reportingEntityId": "127.0.1.1",

+"eventType": "fault_owb-rdm-003_lossOfSignal",

+"priority": "High",

+"version": 1.0,

+"reportingEntityName": "agilevm",

+"sequence": 0,

+"domain": "fault",

+"functionalRole": "rdm",

+"lastEpochMicrosec": 1469528413000,

+"sourceName": "owb-rdm-003"

+},

+"faultFields": {

+"eventSeverity": "CRITICAL",

+"alarmCondition": "lossOfSignal",

+"faultFieldsVersion": 1.0,

+"specificProblem": "lossOfSignal",

+"alarmInterfaceA": "1/0/0 E1",

+"alarmAdditionalInformation": [

+{

+"name": "DIR",

+"value": "tx"

+},

+{

+"name": "LOC",

+"value": "NEND"

+},

+{

+"name": "TYPE",

+"value": "communication"

+},

+{

+"name": "CKTID",

+"value": "circuit-1"

+}

+],

+"eventSourceType": "port",

+"vfStatus": "Active"

+}

+}

+}