summaryrefslogtreecommitdiffstats
path: root/src/main
diff options
context:
space:
mode:
Diffstat (limited to 'src/main')
-rw-r--r--src/main/java/org/onap/dcae/commonFunction/CambriaPublisherFactory.java69
-rw-r--r--src/main/java/org/onap/dcae/commonFunction/DmaapPublishers.java96
-rw-r--r--src/main/java/org/onap/dcae/commonFunction/EventPublisherHash.java213
3 files changed, 229 insertions, 149 deletions
diff --git a/src/main/java/org/onap/dcae/commonFunction/CambriaPublisherFactory.java b/src/main/java/org/onap/dcae/commonFunction/CambriaPublisherFactory.java
new file mode 100644
index 00000000..35d3e317
--- /dev/null
+++ b/src/main/java/org/onap/dcae/commonFunction/CambriaPublisherFactory.java
@@ -0,0 +1,69 @@
+package org.onap.dcae.commonFunction;
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import com.att.nsa.cambria.client.CambriaClientBuilders;
+import java.net.MalformedURLException;
+import java.security.GeneralSecurityException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CambriaPublisherFactory {
+
+ private static Logger log = LoggerFactory.getLogger(CambriaPublisherFactory.class);
+
+ public CambriaBatchingPublisher createCambriaPublisher(String streamId)
+ throws MalformedURLException, GeneralSecurityException {
+ String authpwd = null;
+ String 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");
+ }
+ String topic = DmaapPropertyReader.getInstance(CommonStartup.cambriaConfigFile)
+ .getKeyValue(streamId + ".cambria.topic");
+ String 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));
+ return 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));
+ return new CambriaClientBuilders.PublisherBuilder().usingHosts(ueburl).onTopic(topic)
+ // .logTo(log)
+ .logSendFailuresAfter(5)
+ // .limitBatch(100, 10)
+ .build();
+ }
+ }
+}
diff --git a/src/main/java/org/onap/dcae/commonFunction/DmaapPublishers.java b/src/main/java/org/onap/dcae/commonFunction/DmaapPublishers.java
new file mode 100644
index 00000000..c86a3742
--- /dev/null
+++ b/src/main/java/org/onap/dcae/commonFunction/DmaapPublishers.java
@@ -0,0 +1,96 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 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.
+ * 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.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.security.GeneralSecurityException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class DmaapPublishers {
+
+ private static Logger log = LoggerFactory.getLogger(DmaapPublishers.class);
+ private final LoadingCache<String, CambriaBatchingPublisher> publishers;
+
+ private DmaapPublishers(
+ LoadingCache<String, CambriaBatchingPublisher> publishers) {
+ this.publishers = publishers;
+ }
+
+ static DmaapPublishers create() {
+ return create(new CambriaPublisherFactory());
+ }
+
+ static DmaapPublishers create(final CambriaPublisherFactory publisherFactory) {
+ final LoadingCache<String, CambriaBatchingPublisher> cache = CacheBuilder.<String, CambriaBatchingPublisher>newBuilder()
+// .expireAfterAccess(10, TimeUnit.MINUTES)
+ .removalListener(new RemovalListener<String, CambriaBatchingPublisher>() {
+ @Override
+ public void onRemoval(RemovalNotification<String, CambriaBatchingPublisher> notification) {
+ if (notification.getValue() != null) {
+ onCacheItemInvalidated(notification.getValue());
+ }
+ }
+ })
+ .build(new CacheLoader<String, CambriaBatchingPublisher>() {
+ @Override
+ public CambriaBatchingPublisher load(String streamId)
+ throws MalformedURLException, GeneralSecurityException {
+ try {
+ return publisherFactory.createCambriaPublisher(streamId);
+ } catch (MalformedURLException | GeneralSecurityException e) {
+ log.error("CambriaClientBuilders connection reader exception : streamID - " + streamId + " "
+ + e.getMessage());
+ throw e;
+ }
+ }
+ });
+ return new DmaapPublishers(cache);
+ }
+
+ public CambriaBatchingPublisher getByStreamId(String streamId) {
+ return publishers.getUnchecked(streamId);
+ }
+
+ public void closeByStreamId(String streamId) {
+ publishers.invalidate(streamId);
+ }
+
+ private static void onCacheItemInvalidated(CambriaBatchingPublisher pub) {
+ try {
+ 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);
+ }
+ }
+}
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);
+ }
}
-