summaryrefslogtreecommitdiffstats
path: root/src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java
diff options
context:
space:
mode:
Diffstat (limited to 'src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java')
-rw-r--r--src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java213
1 files changed, 64 insertions, 149 deletions
diff --git a/src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java b/src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java
index 49221418..474424a7 100644
--- a/src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java
+++ b/src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java
@@ -1,15 +1,16 @@
/*-
* ============LICENSE_START=======================================================
- * PROJECT
+ * org.onap.dcaegen2.collectors.ves
* ================================================================================
* Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. 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.
@@ -21,159 +22,73 @@
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 com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
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) && (authpwd != 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);
- }
-
- }
+ private static final String VES_UNIQUE_ID = "VESuniqueId";
+ private static final Logger log = LoggerFactory.getLogger(EventPublisherHash.class);
+ private static volatile EventPublisherHash instance = new EventPublisherHash(DmaapPublishers.create());
+ private final DmaapPublishers dmaapPublishers;
+
+ /**
+ * Returns event publisher
+ *
+ * @return event publisher
+ */
+ public static EventPublisherHash getInstance() {
+ return instance;
+ }
+
+ @VisibleForTesting
+ EventPublisherHash(DmaapPublishers dmaapPublishers) {
+ this.dmaapPublishers = dmaapPublishers;
+ }
+
+ public void sendEvent(JSONObject event, String streamid) {
+ log.debug("EventPublisher.sendEvent: instance for publish is ready");
+ clearVesUniqueId(event);
+
+ try {
+ sendEventUsingCachedPublisher(streamid, event);
+ } catch (IOException | IllegalArgumentException e) {
+ log.error("Unable to publish event: {} streamid: {}. Exception: {}", event, streamid, e);
+ dmaapPublishers.closeByStreamId(streamid);
+ }
+ }
+
+ private void clearVesUniqueId(JSONObject event) {
+ 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);
+ }
+ }
+
+ private void sendEventUsingCachedPublisher(String streamid, JSONObject event) throws IOException {
+ int pendingMsgs = dmaapPublishers.getByStreamId(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));
+ }
+
+ @VisibleForTesting
+ public CambriaBatchingPublisher getDmaapPublisher(String streamId) {
+ return dmaapPublishers.getByStreamId(streamId);
+ }
}
-