Optimize collector publish mechanism

Change-Id: I6d5fb2ad45257831e8e0ff62812df1fd1401464d
Issue-ID: DCAEGEN2-281
Signed-off-by: Vijay VK <vv770d@att.com>

4 files changed, 202 insertions, 48 deletions

diff --git a/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java b/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java
index b4fae2a6..d1dbca92 100644
--- a/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java
+++ b/src/main/java/org/onap/dcae/commonFunction/CommonStartup.java
@@ -195,7 +195,10 @@ public class CommonStartup extends NsaBaseEndpoint implements Runnable {
 			// Thread epThread=new Thread(ep);
 			// epThread.start();
 			executor = Executors.newFixedThreadPool(20);
-			executor.execute(ep);
+			//executor.execute(ep);
+			for (int i = 0; i < 20; ++i) {
+				executor.execute(ep);
+				}
 
 		} catch (loadException | missingReqdSetting | IOException | invalidSettingValue | ServletException
 				| InterruptedException e) {
diff --git a/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java b/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java
index 7e42bb57..4fb5adf0 100644
--- a/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java
+++ b/src/main/java/org/onap/dcae/commonFunction/ConfigProcessors.java
@@ -41,9 +41,7 @@ public class ConfigProcessors {
 		event = eventJson;
 	}
 
-	/**
-	 * 
-	 */
+
 	public void getValue(JSONObject J) {
 		// log.info("addAttribute");
 		final String field = J.getString(FIELD);
@@ -57,9 +55,7 @@ public class ConfigProcessors {
 			log.info("Filter not met");
 	}
 
-	/**
-	 * 
-	 */
+
 	public void setValue(JSONObject J) {
 		// log.info("addAttribute");
 		final String field = J.getString(FIELD);
@@ -73,9 +69,8 @@ public class ConfigProcessors {
 			log.info("Filter not met");
 	}
 
-	/**
-	 * 
-	 */
+
+
 	public String evaluate(String str) {
 		String value = str;
 		if (str.startsWith("$")) {
@@ -85,9 +80,7 @@ public class ConfigProcessors {
 		return value;
 	}
 
-	/**
-	 * { "functionName":"suppressEvent", "args":{} }
-	 */
+
 	public void suppressEvent(JSONObject J) {
 		// log.info("addAttribute");
 		final JSONObject filter = J.optJSONObject(FILTER);
@@ -100,9 +93,7 @@ public class ConfigProcessors {
 			log.info("Filter not met");
 	}
 
-	/**
-	 * 
-	 */
+
 	public void addAttribute(JSONObject J) {
 		// log.info("addAttribute begin");
 		final String field = J.getString(FIELD);
@@ -119,9 +110,7 @@ public class ConfigProcessors {
 		// log.info("addAttribute End");
 	}
 
-	/**
-	 * 
-	 */
+
 	public void updateAttribute(JSONObject J) {
 		// log.info("updateAttribute");
 		final String field = J.getString(FIELD);
@@ -135,9 +124,7 @@ public class ConfigProcessors {
 			log.info("Filter not met");
 	}
 
-	/**
-	 * 
-	 */
+
 	public void removeAttribute(JSONObject J) {
 		// log.info("removeAttribute");
 		final String field = J.getString(FIELD);
@@ -149,9 +136,7 @@ public class ConfigProcessors {
 			log.info("Filter not met");
 	}
 
-	/**
-	 * 
-	 */
+
 	public void renameArrayInArray(JSONObject J) // map
 	{
 		log.info("renameArrayInArray");
@@ -185,9 +170,7 @@ public class ConfigProcessors {
 			log.info("Filter not met");
 	}
 
-	/**
-	 * 
-	 */
+
 	public void map(JSONObject J) {
 		// log.info("mapAttribute");
 		final String field = J.getString(FIELD);
@@ -200,9 +183,6 @@ public class ConfigProcessors {
 			mapAttribute(J);
 	}
 
-	/**
-	 * 
-	 */
 	public String performOperation(String operation, String value) {
 		log.info("performOperation");
 		if (operation != null) {
@@ -214,9 +194,7 @@ public class ConfigProcessors {
 		return value;
 	}
 
-	/**
-	 * 
-	 */
+
 	// public void mapAttributeToArrayAttribute(JSONObject J)
 	public void mapAttribute(JSONObject J) {
 		// log.info("mapAttribute");
@@ -241,9 +219,7 @@ public class ConfigProcessors {
 			log.info("Filter not met");
 	}
 
-	/**
-	 * 
-	 */
+
 	public void mapToJArray(JSONObject J) {
 		log.info("mapToJArray");
 		String field = J.getString(FIELD);
@@ -355,9 +331,7 @@ public class ConfigProcessors {
 			log.info("Filter not met");
 	}
 
-	/**
-	 * 
-	 */
+
 	private void removeEventKey(String field) {
 		String[] keySet = field.split("\\.", field.length());
 		JSONObject keySeries = event;
@@ -371,9 +345,7 @@ public class ConfigProcessors {
 
 	}
 
-	/**
-	 * 
-	 */
+
 	private boolean checkFilter(JSONObject jo, String key, String logicKey) {
 		String filterValue = jo.getString(key);
 		boolean retVal = true;
@@ -431,9 +403,7 @@ public class ConfigProcessors {
 		return retVal;
 	}
 
-	/**
-	 * 
-	 */
+
 	public boolean isFilterMet(JSONObject jo) {
 		boolean retval = true;
 		// log.info("Filter==" + jo.toString());
@@ -579,3 +549,4 @@ public class ConfigProcessors {
 
 	private JSONObject event = new JSONObject();
 }
+
diff --git a/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java b/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
index 79dea790..6811c672 100644
--- a/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
+++ b/src/main/java/org/onap/dcae/commonFunction/EventProcessor.java
@@ -70,7 +70,7 @@ public class EventProcessor implements Runnable {
 		try {
 
 			event = CommonStartup.fProcessingInputQueue.take();
-			log.info("EventProcessor\tRemoving element: " + event);
+			log.info("QueueSize:" + CommonStartup.fProcessingInputQueue.size()+  "\tEventProcessor\tRemoving element: " + event );
 
 			// EventPublisher Ep=new EventPublisher();
 			while (event != null) {
@@ -95,7 +95,8 @@ public class EventProcessor implements Runnable {
 					for (String aStreamIdList : streamIdList) {
 						log.info("Invoking publisher for streamId:" + aStreamIdList);
 						this.overrideEvent();
-						EventPublisher.getInstance(aStreamIdList).sendEvent(event);
+						//EventPublisher.getInstance(aStreamIdList).sendEvent(event);
+						EventPublisherHash.getInstance().sendEvent(event, aStreamIdList);
 
 					}
 				}
diff --git a/src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java b/src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java
new file mode 100644
index 00000000..f8bdcaa2
--- /dev/null
+++ b/src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java
@@ -0,0 +1,179 @@
+/*-
+ * ============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.onap.dcae.commonFunction;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import com.att.nsa.cambria.client.CambriaClientBuilders;
+import com.att.nsa.clock.SaClock;
+import com.att.nsa.logging.LoggingContext;
+import com.att.nsa.logging.log4j.EcompFields;
+import com.google.gson.JsonArray;
+
+import org.json.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class EventPublisherHash {
+
+	private static final String VES_UNIQUE_ID = "VESuniqueId";
+	private static EventPublisherHash instance;
+	private  CambriaBatchingPublisher pub;
+
+	private String streamid = "";
+	private String ueburl = "";
+	private String topic = "";
+	private String authuser = "";
+	private String authpwd = "";
+
+	private static Logger log = LoggerFactory.getLogger(EventPublisherHash.class);
+
+	protected static HashMap<String, CambriaBatchingPublisher> map = new HashMap<String, CambriaBatchingPublisher>();
+	
+
+
+	public CambriaBatchingPublisher Dmaaphash(String newstreamid) {
+		pub = null;
+		streamid = newstreamid;
+		if (map != null && map.containsKey(streamid)){
+			pub =  map.get(streamid);
+			
+		}
+		else
+		{
+		
+		try {
+			ueburl = DmaapPropertyReader.getInstance(CommonStartup.cambriaConfigFile).dmaap_hash
+					.get(streamid + ".cambria.url");
+
+			if (ueburl == null) {
+				ueburl = DmaapPropertyReader.getInstance(CommonStartup.cambriaConfigFile).dmaap_hash
+						.get(streamid + ".cambria.hosts");
+			}
+			topic = DmaapPropertyReader.getInstance(CommonStartup.cambriaConfigFile)
+					.getKeyValue(streamid + ".cambria.topic");
+			authuser = DmaapPropertyReader.getInstance(CommonStartup.cambriaConfigFile)
+					.getKeyValue(streamid + ".basicAuthUsername");
+
+			if (authuser != null) {
+				authpwd = DmaapPropertyReader.getInstance(CommonStartup.cambriaConfigFile).dmaap_hash
+						.get(streamid + ".basicAuthPassword");
+			}
+			
+			
+			if (authuser != null) {
+				log.debug(String.format("URL:%sTOPIC:%sAuthUser:%sAuthpwd:%s", ueburl, topic, authuser, authpwd));
+				pub = new CambriaClientBuilders.PublisherBuilder().usingHosts(ueburl).onTopic(topic).usingHttps()
+						.authenticatedByHttp(authuser, authpwd).logSendFailuresAfter(5)
+						// .logTo(log)
+						// .limitBatch(100, 10)
+						.build();
+			} else {
+
+				log.debug(String.format("URL:%sTOPIC:%s", ueburl, topic));
+				pub = new CambriaClientBuilders.PublisherBuilder().usingHosts(ueburl).onTopic(topic)
+						// .logTo(log)
+						.logSendFailuresAfter(5)
+						// .limitBatch(100, 10)
+						.build();
+
+			}
+			
+			map.put(streamid, pub);
+			
+
+		}  catch ( Exception e) {
+			log.error("CambriaClientBuilders connection reader exception : streamID - " + newstreamid + " " + e.getMessage());
+		} 
+
+		}
+		return pub;
+	
+	}
+
+	/**
+	 * Returns event publisher
+	 * @return event publisher
+	 */
+	public static synchronized EventPublisherHash getInstance() {
+		if (instance == null) {
+			instance = new EventPublisherHash();
+		}
+		return instance;
+
+	}
+
+
+	public synchronized void sendEvent(JSONObject event, String streamid) {
+
+		log.debug("EventPublisher.sendEvent: instance for publish is ready");
+
+		if (event.has(VES_UNIQUE_ID)) {
+			String uuid = event.get(VES_UNIQUE_ID).toString();
+			LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid);
+			localLC.put(EcompFields.kBeginTimestampMs, SaClock.now());
+			log.debug("Removing VESuniqueid object from event");
+			event.remove(VES_UNIQUE_ID);
+		}
+
+		try {
+			
+			
+			int pendingMsgs = Dmaaphash(streamid).send("MyPartitionKey", event.toString());
+			// this.wait(2000);
+
+			if (pendingMsgs > 100) {
+				log.info("Pending Message Count=" + pendingMsgs);
+			}
+
+			log.info("pub.send invoked - no error");
+			//CommonStartup.oplog.info(String.format("URL:%sTOPIC:%sEvent Published:%s", ueburl, topic, event));
+			CommonStartup.oplog.info(String.format("StreamID:%s Event Published:%s ", streamid, event));
+		} catch (IOException  | IllegalArgumentException e) {
+			log.error("Unable to publish event: {} streamid: {}. Exception: {}", event, streamid, e);
+			Dmaaphash(streamid).close();
+			map.remove(streamid);
+		} 
+
+	}
+
+	public synchronized void closePublisher() {
+
+		try {
+			if (pub != null) {
+				
+				final List<?> stuck = pub.close(20, TimeUnit.SECONDS);
+				if (!stuck.isEmpty()) {
+					log.error(stuck.size() + " messages unsent");
+				}
+			}
+		} catch (InterruptedException | IOException e) {
+			log.error("Caught Exception on Close event: {}", e);
+		}
+
+	}
+}
+