aboutsummaryrefslogtreecommitdiffstats
path: root/dcae-analytics-dmaap/src/main/java
diff options
context:
space:
mode:
Diffstat (limited to 'dcae-analytics-dmaap/src/main/java')
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/DMaaPMRFactory.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/DMaaPMRFactory.java)224
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java)386
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java)64
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java)496
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java)590
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java)76
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java)140
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java)92
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java)80
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java)160
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java)124
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java)756
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java)58
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java)190
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java)98
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java)420
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java)174
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java)90
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java)252
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java)114
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java)94
-rw-r--r--dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java (renamed from dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java)258
22 files changed, 2468 insertions, 2468 deletions
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/DMaaPMRFactory.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/DMaaPMRFactory.java
index a23c1e9..7163fe2 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/DMaaPMRFactory.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/DMaaPMRFactory.java
@@ -1,112 +1,112 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap;
-
-import com.google.inject.AbstractModule;
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.module.AnalyticsDMaaPModule;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nonnull;
-
-/**
- * Creates pre injected implementations for {@link DMaaPMRPublisher} and {@link DMaaPMRSubscriber}
- * <p>
- * Usage:
- * <p>Create an instance of DMaaP MR Factory</p>
- * <pre>
- * DMaaPFactory dmaapFactory = DMaaPFactory.initalize()
- * </pre>
- * <p>Create a new DMaaP MR Publisher</p>
- * <pre>
- * DMaaPMRPublisher publisher = dmaapFactory.createPublisher(publisherConfig)
- * </pre>
- * <p>Create new DMaaP MR Subscriber</p>
- * <pre>
- * DMaaPMRSubscriber subscriber = dmaapFactory.createSubscriber(subscriberConfig)
- * </pre>
- * <p>
- * <strong>All Clients must use this Factory to initalize DMaaP Message Router Publishers and Subscribers</strong>
- * </p>
- * <p>
- * @author Rajiv Singla . Creation Date: 10/20/2016.
- */
-public class DMaaPMRFactory {
-
- private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRFactory.class);
-
- private final Injector injector;
-
- public DMaaPMRFactory(AbstractModule guiceModule) {
- injector = Guice.createInjector(guiceModule);
- }
-
- /**
- * Returns configured instance of {@link DMaaPMRPublisher}
- *
- * @param publisherConfig Publisher Config
- * @return configured instance of DMaaP MR Publisher
- */
- public DMaaPMRPublisher createPublisher(@Nonnull DMaaPMRPublisherConfig publisherConfig) {
- final DMaaPMRPublisherFactory publisherFactory = injector.getInstance(DMaaPMRPublisherFactory.class);
- LOG.debug("Creating new DMaaP MR Publisher Instance with configuration: {}", publisherConfig);
- final DMaaPMRPublisher dMaaPMRPublisher = publisherFactory.create(publisherConfig);
- LOG.info("Created new DMaaP MR Publisher Instance. Publisher creation time: {}",
- dMaaPMRPublisher.getPublisherCreationTime());
- return dMaaPMRPublisher;
- }
-
- /**
- * Returns configured instance of {@link DMaaPMRSubscriber}
- *
- * @param subscriberConfig Subscriber Config
- * @return configured instance of DMaaP MR Subscriber
- */
- public DMaaPMRSubscriber createSubscriber(@Nonnull DMaaPMRSubscriberConfig subscriberConfig) {
- final DMaaPMRSubscriberFactory subscriberFactory = injector.getInstance(DMaaPMRSubscriberFactory.class);
- LOG.debug("Creating new DMaaP MR Subscriber Instance with configuration: {}", subscriberConfig);
- final DMaaPMRSubscriber dMaaPMRSubscriber = subscriberFactory.create(subscriberConfig);
- LOG.info("Created new DMaaP MR Subscriber Instance. Subscriber creation time: {}",
- dMaaPMRSubscriber.getSubscriberCreationTime());
- return dMaaPMRSubscriber;
- }
-
- /**
- * Creates an instance of {@link DMaaPMRFactory}
- *
- * @return {@link DMaaPMRFactory} factory instance
- */
- public static DMaaPMRFactory create() {
- final DMaaPMRFactory dMaaPMRFactory = new DMaaPMRFactory(new AnalyticsDMaaPModule());
- LOG.info("Created new instance of DMaaP MR Factory");
- return dMaaPMRFactory;
- }
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+import org.onap.dcae.apod.analytics.dmaap.module.AnalyticsDMaaPModule;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherFactory;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+/**
+ * Creates pre injected implementations for {@link DMaaPMRPublisher} and {@link DMaaPMRSubscriber}
+ * <p>
+ * Usage:
+ * <p>Create an instance of DMaaP MR Factory</p>
+ * <pre>
+ * DMaaPFactory dmaapFactory = DMaaPFactory.initalize()
+ * </pre>
+ * <p>Create a new DMaaP MR Publisher</p>
+ * <pre>
+ * DMaaPMRPublisher publisher = dmaapFactory.createPublisher(publisherConfig)
+ * </pre>
+ * <p>Create new DMaaP MR Subscriber</p>
+ * <pre>
+ * DMaaPMRSubscriber subscriber = dmaapFactory.createSubscriber(subscriberConfig)
+ * </pre>
+ * <p>
+ * <strong>All Clients must use this Factory to initalize DMaaP Message Router Publishers and Subscribers</strong>
+ * </p>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/20/2016.
+ */
+public class DMaaPMRFactory {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRFactory.class);
+
+ private final Injector injector;
+
+ public DMaaPMRFactory(AbstractModule guiceModule) {
+ injector = Guice.createInjector(guiceModule);
+ }
+
+ /**
+ * Returns configured instance of {@link DMaaPMRPublisher}
+ *
+ * @param publisherConfig Publisher Config
+ * @return configured instance of DMaaP MR Publisher
+ */
+ public DMaaPMRPublisher createPublisher(@Nonnull DMaaPMRPublisherConfig publisherConfig) {
+ final DMaaPMRPublisherFactory publisherFactory = injector.getInstance(DMaaPMRPublisherFactory.class);
+ LOG.debug("Creating new DMaaP MR Publisher Instance with configuration: {}", publisherConfig);
+ final DMaaPMRPublisher dMaaPMRPublisher = publisherFactory.create(publisherConfig);
+ LOG.info("Created new DMaaP MR Publisher Instance. Publisher creation time: {}",
+ dMaaPMRPublisher.getPublisherCreationTime());
+ return dMaaPMRPublisher;
+ }
+
+ /**
+ * Returns configured instance of {@link DMaaPMRSubscriber}
+ *
+ * @param subscriberConfig Subscriber Config
+ * @return configured instance of DMaaP MR Subscriber
+ */
+ public DMaaPMRSubscriber createSubscriber(@Nonnull DMaaPMRSubscriberConfig subscriberConfig) {
+ final DMaaPMRSubscriberFactory subscriberFactory = injector.getInstance(DMaaPMRSubscriberFactory.class);
+ LOG.debug("Creating new DMaaP MR Subscriber Instance with configuration: {}", subscriberConfig);
+ final DMaaPMRSubscriber dMaaPMRSubscriber = subscriberFactory.create(subscriberConfig);
+ LOG.info("Created new DMaaP MR Subscriber Instance. Subscriber creation time: {}",
+ dMaaPMRSubscriber.getSubscriberCreationTime());
+ return dMaaPMRSubscriber;
+ }
+
+ /**
+ * Creates an instance of {@link DMaaPMRFactory}
+ *
+ * @return {@link DMaaPMRFactory} factory instance
+ */
+ public static DMaaPMRFactory create() {
+ final DMaaPMRFactory dMaaPMRFactory = new DMaaPMRFactory(new AnalyticsDMaaPModule());
+ LOG.info("Created new instance of DMaaP MR Factory");
+ return dMaaPMRFactory;
+ }
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java
index 32a4357..3f74091 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java
@@ -1,193 +1,193 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.config;
-
-import com.google.common.base.Objects;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Locale;
-
-import static org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils.JSON_APPLICATION_TYPE;
-
-/**
- * <p>
- * Contains common parameters for both DMaaP Message Router Publisher and Subscriber Configs
- * <p>
- * @author Rajiv Singla . Creation Date: 10/12/2016.
- */
-public abstract class DMaaPMRBaseConfig implements DMaaPMRConfig {
-
- protected static final Logger LOG = LoggerFactory.getLogger(DMaaPMRBaseConfig.class);
-
- protected String hostName;
- protected Integer portNumber;
- protected String topicName;
- protected String protocol;
- protected String userName;
- protected String userPassword;
- protected String contentType;
-
- /**
- * Provides host name e.g. mrlocal-mtnjftle01.homer.com
- *
- * @return host name
- */
- public String getHostName() {
- return hostName;
- }
-
-
- /**
- * Provides Port Number of DMaaP MR Topic Host. Defaults to 80
- *
- * @return host port number
- */
- public Integer getPortNumber() {
- return portNumber;
- }
-
- /**
- * Provides topic name e.g. com.dcae.dmaap.mtnje2.DcaeTestVES
- *
- * @return topic name
- */
- public String getTopicName() {
- return topicName;
- }
-
- /**
- * Provides protocol type e.g. http or https
- *
- * @return protocol type
- */
- public String getProtocol() {
- return protocol;
- }
-
- /**
- * Provides content type e.g. application/json
- *
- * @return content type
- */
- public String getContentType() {
- return contentType;
- }
-
-
- /**
- * Provides User name for the DMaaP MR Topic authentication
- *
- * @return user name
- */
- public String getUserName() {
- return userName;
- }
-
- /**
- * Provides User password for the DMaaP MR Topic authentication
- *
- * @return user Password
- */
- public String getUserPassword() {
- return userPassword;
- }
-
-
- /**
- * Trims, adjusts casing and validates user input String for protocol selection
- *
- * @param protocol - User input for protocol String
- * @return - network protocol e.g http or https
- */
- protected static String normalizeValidateProtocol(final String protocol) {
- // validate that only http and https are supported protocols are Supported for DMaaP MR
- String normalizedProtocolString = protocol.trim().toLowerCase(Locale.ENGLISH);
- if (normalizedProtocolString.isEmpty() ||
- !("http".equals(normalizedProtocolString) || "https".equals(normalizedProtocolString))) {
-
- final String errorMessage =
- "Unsupported protocol selection. Only HTTPS and HTTPS are currently supported for DMaaP MR";
-
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
- }
- return normalizedProtocolString;
- }
-
-
- /**
- * Trims, adjust casing and validates content type is supported by DMaaP.
- *
- * NOTE: DMaaP currently only support application/json content type
- *
- * @param contentType content type that needs to checked for DMaaP MR support
- * @return true if content type is supported by DMaaP MR
- */
- protected static String normalizeValidateContentType(final String contentType) {
- // Current DMaaP MR is only supporting "application/json" content type
- String normalizedContentType = contentType.trim().toLowerCase(Locale.ENGLISH);
- final boolean isSupported = contentType.equals(JSON_APPLICATION_TYPE);
- if (!isSupported) {
- final String errorMessage =
- "Unsupported content type selection. Only application/json is currently supported for DMaaP MR";
-
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
- }
- return normalizedContentType;
- }
-
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof DMaaPMRBaseConfig)) {
- return false;
- }
- DMaaPMRBaseConfig that = (DMaaPMRBaseConfig) o;
- return Objects.equal(hostName, that.hostName) &&
- Objects.equal(portNumber, that.portNumber) &&
- Objects.equal(topicName, that.topicName) &&
- Objects.equal(protocol, that.protocol) &&
- Objects.equal(userName, that.userName) &&
- Objects.equal(userPassword, that.userPassword) &&
- Objects.equal(contentType, that.contentType);
- }
-
- @Override
- public int hashCode() {
- return Objects.hashCode(hostName, portNumber, topicName, protocol, userName, userPassword, contentType);
- }
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this)
- .add("hostName", hostName)
- .add("portNumber", portNumber)
- .add("topicName", topicName)
- .add("protocol", protocol)
- .add("userName", userName)
- .add("contentType", contentType)
- .toString();
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.config;
+
+import com.google.common.base.Objects;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Locale;
+
+import static org.onap.dcae.apod.analytics.common.utils.HTTPUtils.JSON_APPLICATION_TYPE;
+
+/**
+ * <p>
+ * Contains common parameters for both DMaaP Message Router Publisher and Subscriber Configs
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/12/2016.
+ */
+public abstract class DMaaPMRBaseConfig implements DMaaPMRConfig {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(DMaaPMRBaseConfig.class);
+
+ protected String hostName;
+ protected Integer portNumber;
+ protected String topicName;
+ protected String protocol;
+ protected String userName;
+ protected String userPassword;
+ protected String contentType;
+
+ /**
+ * Provides host name e.g. mrlocal-mtnjftle01.homer.com
+ *
+ * @return host name
+ */
+ public String getHostName() {
+ return hostName;
+ }
+
+
+ /**
+ * Provides Port Number of DMaaP MR Topic Host. Defaults to 80
+ *
+ * @return host port number
+ */
+ public Integer getPortNumber() {
+ return portNumber;
+ }
+
+ /**
+ * Provides topic name e.g. com.dcae.dmaap.mtnje2.DcaeTestVES
+ *
+ * @return topic name
+ */
+ public String getTopicName() {
+ return topicName;
+ }
+
+ /**
+ * Provides protocol type e.g. http or https
+ *
+ * @return protocol type
+ */
+ public String getProtocol() {
+ return protocol;
+ }
+
+ /**
+ * Provides content type e.g. application/json
+ *
+ * @return content type
+ */
+ public String getContentType() {
+ return contentType;
+ }
+
+
+ /**
+ * Provides User name for the DMaaP MR Topic authentication
+ *
+ * @return user name
+ */
+ public String getUserName() {
+ return userName;
+ }
+
+ /**
+ * Provides User password for the DMaaP MR Topic authentication
+ *
+ * @return user Password
+ */
+ public String getUserPassword() {
+ return userPassword;
+ }
+
+
+ /**
+ * Trims, adjusts casing and validates user input String for protocol selection
+ *
+ * @param protocol - User input for protocol String
+ * @return - network protocol e.g http or https
+ */
+ protected static String normalizeValidateProtocol(final String protocol) {
+ // validate that only http and https are supported protocols are Supported for DMaaP MR
+ String normalizedProtocolString = protocol.trim().toLowerCase(Locale.ENGLISH);
+ if (normalizedProtocolString.isEmpty() ||
+ !("http".equals(normalizedProtocolString) || "https".equals(normalizedProtocolString))) {
+
+ final String errorMessage =
+ "Unsupported protocol selection. Only HTTPS and HTTPS are currently supported for DMaaP MR";
+
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
+ }
+ return normalizedProtocolString;
+ }
+
+
+ /**
+ * Trims, adjust casing and validates content type is supported by DMaaP.
+ *
+ * NOTE: DMaaP currently only support application/json content type
+ *
+ * @param contentType content type that needs to checked for DMaaP MR support
+ * @return true if content type is supported by DMaaP MR
+ */
+ protected static String normalizeValidateContentType(final String contentType) {
+ // Current DMaaP MR is only supporting "application/json" content type
+ String normalizedContentType = contentType.trim().toLowerCase(Locale.ENGLISH);
+ final boolean isSupported = contentType.equals(JSON_APPLICATION_TYPE);
+ if (!isSupported) {
+ final String errorMessage =
+ "Unsupported content type selection. Only application/json is currently supported for DMaaP MR";
+
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
+ }
+ return normalizedContentType;
+ }
+
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof DMaaPMRBaseConfig)) {
+ return false;
+ }
+ DMaaPMRBaseConfig that = (DMaaPMRBaseConfig) o;
+ return Objects.equal(hostName, that.hostName) &&
+ Objects.equal(portNumber, that.portNumber) &&
+ Objects.equal(topicName, that.topicName) &&
+ Objects.equal(protocol, that.protocol) &&
+ Objects.equal(userName, that.userName) &&
+ Objects.equal(userPassword, that.userPassword) &&
+ Objects.equal(contentType, that.contentType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(hostName, portNumber, topicName, protocol, userName, userPassword, contentType);
+ }
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(this)
+ .add("hostName", hostName)
+ .add("portNumber", portNumber)
+ .add("topicName", topicName)
+ .add("protocol", protocol)
+ .add("userName", userName)
+ .add("contentType", contentType)
+ .toString();
+ }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java
index 687d489..2ff3c4e 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java
@@ -1,32 +1,32 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.config;
-
-/**
- * <p>
- * Marker Interface for all DMaaP MR Configs.
- * Holds Default configuration parameters for all Configs
- * <p>
- * @author Rajiv Singla . Creation Date: 10/12/2016.
- */
-public interface DMaaPMRConfig {
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.config;
+
+/**
+ * <p>
+ * Marker Interface for all DMaaP MR Configs.
+ * Holds Default configuration parameters for all Configs
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/12/2016.
+ */
+public interface DMaaPMRConfig {
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java
index 6fbebf2..8dfc289 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java
@@ -1,248 +1,248 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.config;
-
-import com.google.common.base.Objects;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-
-import javax.annotation.Nonnull;
-
-/**
- * <p>
- * Immutable DMaaP MR Configuration for DMaaP MR Publisher.
- * <p>
- * Use {@link DMaaPMRPublisherConfig.Builder} to construct Subscriber Configuration
- * </p>
- * <p>
- * @author Rajiv Singla . Creation Date: 10/12/2016.
- */
-public class DMaaPMRPublisherConfig extends DMaaPMRBaseConfig {
-
- /**
- * Publisher batching queue size
- */
- private int maxBatchSize;
-
- /**
- * Publisher Recovery Queue Size
- */
- private int maxRecoveryQueueSize;
-
-
- private DMaaPMRPublisherConfig(@Nonnull String hostName,
- @Nonnull Integer portNumber,
- @Nonnull String topicName,
- @Nonnull String protocol,
- String userName,
- String userPassword,
- @Nonnull String contentType,
- int maxBatchSize,
- int maxRecoveryQueueSize) {
- this.hostName = hostName;
- this.portNumber = portNumber;
- this.topicName = topicName;
- this.protocol = protocol;
- this.userName = userName;
- this.userPassword = userPassword;
- this.contentType = contentType;
- this.maxBatchSize = maxBatchSize;
- this.maxRecoveryQueueSize = maxRecoveryQueueSize;
- }
-
-
- /**
- * Builder to initialize immutable {@link DMaaPMRPublisherConfig} object
- */
- public static class Builder {
-
- private String hostName;
- private Integer portNumber;
- private String topicName;
- private String userName;
- private String userPassword;
- private String protocol;
- private String contentType;
- private int maxBatchSize;
- private int maxRecoveryQueueSize;
-
- public Builder(@Nonnull String hostName, @Nonnull String topicName) {
- // required values
- this.hostName = hostName;
- this.topicName = topicName;
- // Default values
- this.portNumber = AnalyticsConstants.DEFAULT_PORT_NUMBER;
- this.userName = AnalyticsConstants.DEFAULT_USER_NAME;
- this.userPassword = AnalyticsConstants.DEFAULT_USER_PASSWORD;
- this.protocol = AnalyticsConstants.DEFAULT_PROTOCOL;
- this.contentType = AnalyticsConstants.DEFAULT_CONTENT_TYPE;
- this.maxBatchSize = AnalyticsConstants.DEFAULT_PUBLISHER_MAX_BATCH_SIZE;
- this.maxRecoveryQueueSize = AnalyticsConstants.DEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE;
- }
-
- /**
- * Setup for custom host port number - Defaults to 80.
- *
- * @param portNumber custom port number
- * @return Builder object itself for chaining
- */
- public Builder setPortNumber(@Nonnull Integer portNumber) {
- this.portNumber = portNumber;
- return this;
- }
-
-
- /**
- * Setup user name for authentication. If no username is provided authentication will be disabled
- *
- * @param userName user name for DMaaP Topic Authentication
- * @return Builder object itself for chaining
- */
- public Builder setUserName(@Nonnull String userName) {
- this.userName = userName;
- return this;
- }
-
-
- /**
- * Setup user password for authentication. If no password is provided authentication will be disabled
- *
- * @param userPassword user password for DMaaP Topic Authentication
- * @return Builder object itself for chaining
- */
- public Builder setUserPassword(@Nonnull String userPassword) {
- this.userPassword = userPassword;
- return this;
- }
-
-
- /**
- * Setup custom Publisher protocol - Defaults to https.
- * Note: Only http and https are currently supported.
- *
- * @param protocol protocol e.g. https
- * @return Builder object itself for chaining
- */
- public Builder setProtocol(@Nonnull String protocol) {
- this.protocol = normalizeValidateProtocol(protocol);
- return this;
- }
-
-
- /**
- * Setup custom Publisher content-type - Defaults to application/json
- *
- * @param contentType content type e.g. application/json
- * @return Builder object itself for chaining
- */
- public Builder setContentType(@Nonnull String contentType) {
- final String normalizedContentType = normalizeValidateContentType(contentType);
- this.contentType = normalizedContentType;
- return this;
- }
-
-
- /**
- * Setup custom Publisher Max Batch Size - Defaults to 100
- *
- * @param maxBatchSize max Batch Size
- * @return Builder object itself for chaining
- */
- public Builder setMaxBatchSize(int maxBatchSize) {
- this.maxBatchSize = maxBatchSize;
- return this;
- }
-
-
- /**
- * Setup custom Maximum Recovery Queue Size. Recovery Queue is used to hold messages temporarily in case
- * DMaaP MR Publisher topic is not responding for any reason. Defaults to 100,000
- *
- * @param maxRecoveryQueueSize max recovery queue size
- * @return Builder object itself for chaining
- */
- public Builder setMaxRecoveryQueueSize(int maxRecoveryQueueSize) {
- this.maxRecoveryQueueSize = maxRecoveryQueueSize;
- return this;
- }
-
- /**
- * Creates immutable instance of {@link DMaaPMRPublisherConfig}
- *
- * @return Builds and returns thread safe, immutable {@link DMaaPMRPublisherConfig} object
- */
- public DMaaPMRPublisherConfig build() {
- return new DMaaPMRPublisherConfig(hostName, portNumber, topicName, protocol, userName, userPassword,
- contentType, maxBatchSize, maxRecoveryQueueSize);
- }
-
- }
-
-
- /**
- * Returns max Publisher Batch Queue Size
- *
- * @return max Publisher Batch Queue size
- */
- public int getMaxBatchSize() {
- return maxBatchSize;
- }
-
- /**
- * Returns max Publisher Recovery Queue Size
- *
- * @return max Recovery Queue size
- */
- public int getMaxRecoveryQueueSize() {
- return maxRecoveryQueueSize;
- }
-
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- DMaaPMRPublisherConfig that = (DMaaPMRPublisherConfig) o;
- return maxBatchSize == that.maxBatchSize &&
- maxRecoveryQueueSize == that.maxRecoveryQueueSize;
- }
-
- @Override
- public int hashCode() {
- return Objects.hashCode(super.hashCode(), maxBatchSize, maxRecoveryQueueSize);
- }
-
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this)
- .add("baseConfig", super.toString())
- .add("maxBatchSize", maxBatchSize)
- .add("maxRecoveryQueueSize", maxRecoveryQueueSize)
- .toString();
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.config;
+
+import com.google.common.base.Objects;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+
+import javax.annotation.Nonnull;
+
+/**
+ * <p>
+ * Immutable DMaaP MR Configuration for DMaaP MR Publisher.
+ * <p>
+ * Use {@link DMaaPMRPublisherConfig.Builder} to construct Subscriber Configuration
+ * </p>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/12/2016.
+ */
+public class DMaaPMRPublisherConfig extends DMaaPMRBaseConfig {
+
+ /**
+ * Publisher batching queue size
+ */
+ private int maxBatchSize;
+
+ /**
+ * Publisher Recovery Queue Size
+ */
+ private int maxRecoveryQueueSize;
+
+
+ private DMaaPMRPublisherConfig(@Nonnull String hostName,
+ @Nonnull Integer portNumber,
+ @Nonnull String topicName,
+ @Nonnull String protocol,
+ String userName,
+ String userPassword,
+ @Nonnull String contentType,
+ int maxBatchSize,
+ int maxRecoveryQueueSize) {
+ this.hostName = hostName;
+ this.portNumber = portNumber;
+ this.topicName = topicName;
+ this.protocol = protocol;
+ this.userName = userName;
+ this.userPassword = userPassword;
+ this.contentType = contentType;
+ this.maxBatchSize = maxBatchSize;
+ this.maxRecoveryQueueSize = maxRecoveryQueueSize;
+ }
+
+
+ /**
+ * Builder to initialize immutable {@link DMaaPMRPublisherConfig} object
+ */
+ public static class Builder {
+
+ private String hostName;
+ private Integer portNumber;
+ private String topicName;
+ private String userName;
+ private String userPassword;
+ private String protocol;
+ private String contentType;
+ private int maxBatchSize;
+ private int maxRecoveryQueueSize;
+
+ public Builder(@Nonnull String hostName, @Nonnull String topicName) {
+ // required values
+ this.hostName = hostName;
+ this.topicName = topicName;
+ // Default values
+ this.portNumber = AnalyticsConstants.DEFAULT_PORT_NUMBER;
+ this.userName = AnalyticsConstants.DEFAULT_USER_NAME;
+ this.userPassword = AnalyticsConstants.DEFAULT_USER_PASSWORD;
+ this.protocol = AnalyticsConstants.DEFAULT_PROTOCOL;
+ this.contentType = AnalyticsConstants.DEFAULT_CONTENT_TYPE;
+ this.maxBatchSize = AnalyticsConstants.DEFAULT_PUBLISHER_MAX_BATCH_SIZE;
+ this.maxRecoveryQueueSize = AnalyticsConstants.DEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE;
+ }
+
+ /**
+ * Setup for custom host port number - Defaults to 80.
+ *
+ * @param portNumber custom port number
+ * @return Builder object itself for chaining
+ */
+ public Builder setPortNumber(@Nonnull Integer portNumber) {
+ this.portNumber = portNumber;
+ return this;
+ }
+
+
+ /**
+ * Setup user name for authentication. If no username is provided authentication will be disabled
+ *
+ * @param userName user name for DMaaP Topic Authentication
+ * @return Builder object itself for chaining
+ */
+ public Builder setUserName(@Nonnull String userName) {
+ this.userName = userName;
+ return this;
+ }
+
+
+ /**
+ * Setup user password for authentication. If no password is provided authentication will be disabled
+ *
+ * @param userPassword user password for DMaaP Topic Authentication
+ * @return Builder object itself for chaining
+ */
+ public Builder setUserPassword(@Nonnull String userPassword) {
+ this.userPassword = userPassword;
+ return this;
+ }
+
+
+ /**
+ * Setup custom Publisher protocol - Defaults to https.
+ * Note: Only http and https are currently supported.
+ *
+ * @param protocol protocol e.g. https
+ * @return Builder object itself for chaining
+ */
+ public Builder setProtocol(@Nonnull String protocol) {
+ this.protocol = normalizeValidateProtocol(protocol);
+ return this;
+ }
+
+
+ /**
+ * Setup custom Publisher content-type - Defaults to application/json
+ *
+ * @param contentType content type e.g. application/json
+ * @return Builder object itself for chaining
+ */
+ public Builder setContentType(@Nonnull String contentType) {
+ final String normalizedContentType = normalizeValidateContentType(contentType);
+ this.contentType = normalizedContentType;
+ return this;
+ }
+
+
+ /**
+ * Setup custom Publisher Max Batch Size - Defaults to 100
+ *
+ * @param maxBatchSize max Batch Size
+ * @return Builder object itself for chaining
+ */
+ public Builder setMaxBatchSize(int maxBatchSize) {
+ this.maxBatchSize = maxBatchSize;
+ return this;
+ }
+
+
+ /**
+ * Setup custom Maximum Recovery Queue Size. Recovery Queue is used to hold messages temporarily in case
+ * DMaaP MR Publisher topic is not responding for any reason. Defaults to 100,000
+ *
+ * @param maxRecoveryQueueSize max recovery queue size
+ * @return Builder object itself for chaining
+ */
+ public Builder setMaxRecoveryQueueSize(int maxRecoveryQueueSize) {
+ this.maxRecoveryQueueSize = maxRecoveryQueueSize;
+ return this;
+ }
+
+ /**
+ * Creates immutable instance of {@link DMaaPMRPublisherConfig}
+ *
+ * @return Builds and returns thread safe, immutable {@link DMaaPMRPublisherConfig} object
+ */
+ public DMaaPMRPublisherConfig build() {
+ return new DMaaPMRPublisherConfig(hostName, portNumber, topicName, protocol, userName, userPassword,
+ contentType, maxBatchSize, maxRecoveryQueueSize);
+ }
+
+ }
+
+
+ /**
+ * Returns max Publisher Batch Queue Size
+ *
+ * @return max Publisher Batch Queue size
+ */
+ public int getMaxBatchSize() {
+ return maxBatchSize;
+ }
+
+ /**
+ * Returns max Publisher Recovery Queue Size
+ *
+ * @return max Recovery Queue size
+ */
+ public int getMaxRecoveryQueueSize() {
+ return maxRecoveryQueueSize;
+ }
+
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ DMaaPMRPublisherConfig that = (DMaaPMRPublisherConfig) o;
+ return maxBatchSize == that.maxBatchSize &&
+ maxRecoveryQueueSize == that.maxRecoveryQueueSize;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), maxBatchSize, maxRecoveryQueueSize);
+ }
+
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(this)
+ .add("baseConfig", super.toString())
+ .add("maxBatchSize", maxBatchSize)
+ .add("maxRecoveryQueueSize", maxRecoveryQueueSize)
+ .toString();
+ }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java
index 7252ce0..8158b6b 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java
@@ -1,295 +1,295 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.config;
-
-import com.google.common.base.Objects;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-
-import java.util.UUID;
-
-import javax.annotation.Nonnull;
-
-/**
- * <p>
- * Immutable DMaaP MR Configuration for Subscriber.
- * <p>
- * Use {@link DMaaPMRSubscriberConfig.Builder} to construct Subscriber Configuration
- * <p>
- *
- * @author Rajiv Singla . Creation Date: 10/12/2016.
- */
-public final class DMaaPMRSubscriberConfig extends DMaaPMRBaseConfig {
-
- private final String consumerId;
- private final String consumerGroup;
- private final Integer timeoutMS;
- private final Integer messageLimit;
-
- private DMaaPMRSubscriberConfig(@Nonnull String hostName,
- @Nonnull Integer portNumber,
- @Nonnull String topicName,
- @Nonnull String protocol,
- String userName,
- String userPassword,
- @Nonnull String contentType,
- @Nonnull String consumerId,
- @Nonnull String consumerGroup,
- @Nonnull Integer timeoutMS,
- @Nonnull Integer messageLimit) {
- this.hostName = hostName;
- this.portNumber = portNumber;
- this.topicName = topicName;
- this.protocol = protocol;
- this.userName = userName;
- this.userPassword = userPassword;
- this.contentType = contentType;
- this.consumerId = consumerId;
- this.consumerGroup = consumerGroup;
- this.timeoutMS = timeoutMS;
- this.messageLimit = messageLimit;
- }
-
- /**
- * Builder to initialize immutable {@link DMaaPMRSubscriberConfig} object
- */
- public static class Builder {
-
- private String hostName;
- private Integer portNumber;
- private String topicName;
- private String userName;
- private String userPassword;
- private String protocol;
- private String contentType;
- private String consumerId;
- private String consumerGroup;
- private Integer timeoutMS;
- private Integer messageLimit;
-
- public Builder(@Nonnull String hostName,
- @Nonnull String topicName) {
- // Required Values
- this.hostName = hostName;
- this.topicName = topicName;
-
- // Default values
- this.portNumber = AnalyticsConstants.DEFAULT_PORT_NUMBER;
- this.userName = AnalyticsConstants.DEFAULT_USER_NAME;
- this.userPassword = AnalyticsConstants.DEFAULT_USER_PASSWORD;
- this.protocol = AnalyticsConstants.DEFAULT_PROTOCOL;
- this.contentType = AnalyticsConstants.DEFAULT_CONTENT_TYPE;
- this.consumerId = UUID.randomUUID().toString(); // consumer is assigned a random id by default
- this.consumerGroup = AnalyticsConstants.DEFAULT_SUBSCRIBER_GROUP_PREFIX + consumerId; // random group
- this.timeoutMS = AnalyticsConstants.DEFAULT_SUBSCRIBER_TIMEOUT_MS; // defaults to 10ms timeout
- this.messageLimit = AnalyticsConstants.DEFAULT_SUBSCRIBER_MESSAGE_LIMIT; // defaults to 1000 message limit
- }
-
-
- /**
- * Setup for custom host port number - Defaults to 80.
- *
- * @param portNumber custom port number
- * @return Builder object itself for chaining
- */
- public Builder setPortNumber(@Nonnull Integer portNumber) {
- this.portNumber = portNumber;
- return this;
- }
-
-
- /**
- * Setup user name for authentication. If no username is provided authentication will be disabled
- *
- * @param userName user name for DMaaP Topic Authentication
- * @return Builder object itself for chaining
- */
- public Builder setUserName(@Nonnull String userName) {
- this.userName = userName;
- return this;
- }
-
-
- /**
- * Setup user password for authentication. If no password is provided authentication will be disabled
- *
- * @param userPassword user password for DMaaP Topic Authentication
- * @return Builder object itself for chaining
- */
- public Builder setUserPassword(@Nonnull String userPassword) {
- this.userPassword = userPassword;
- return this;
- }
-
-
- /**
- * Setup custom Subscriber protocol - Defaults to https.
- * Note: Only http and https are currently supported.
- *
- * @param protocol protocol e.g. https or http
- * @return Builder object itself for chaining
- */
- public Builder setProtocol(@Nonnull String protocol) {
-
- this.protocol = normalizeValidateProtocol(protocol);
- return this;
- }
-
- /**
- * Setup custom Subscriber content-type - Defaults to application/json
- *
- * @param contentType content type e.g. application/json
- * @return Builder object itself for chaining
- */
- public Builder setContentType(@Nonnull String contentType) {
- final String normalizedContentType = normalizeValidateContentType(contentType);
- this.contentType = normalizedContentType;
- return this;
- }
-
-
- /**
- * Setup custom Consumer Id - Defaults to random Id
- *
- * @param consumerId - custom consumer ID
- * @return Builder object itself for chaining
- */
- public Builder setConsumerId(@Nonnull String consumerId) {
- this.consumerId = consumerId;
- return this;
- }
-
- /**
- * Setup custom Consumer Group - Default to OpenDCAE-DMaaPSub-ConsumerID
- *
- * @param consumerGroup - custom Consumer Group
- * @return Builder object itself for chaining
- */
- public Builder setConsumerGroup(@Nonnull String consumerGroup) {
- this.consumerGroup = consumerGroup;
- return this;
- }
-
- /**
- * Setup Custom Subscriber timeout in ms - Default to no timeout limit
- *
- * @param timeoutMS timeout in milliseconds
- * @return Builder object itself for chaining
- */
- public Builder setTimeoutMS(@Nonnull Integer timeoutMS) {
- this.timeoutMS = timeoutMS;
- return this;
- }
-
- /**
- * Setup custom Subscriber Message Limit - Default to no limit
- *
- * @param messageLimit message Limit
- * @return Builder object itself for chaining
- */
- public Builder setMessageLimit(@Nonnull Integer messageLimit) {
- this.messageLimit = messageLimit;
- return this;
- }
-
- /**
- * Builds Immutable instance of {@link DMaaPMRSubscriberConfig}
- *
- * @return immutable DMaaP Subscriber Config Object
- */
- public DMaaPMRSubscriberConfig build() {
- return new DMaaPMRSubscriberConfig(hostName, portNumber, topicName, protocol, userName, userPassword,
- contentType, consumerId, consumerGroup, timeoutMS, messageLimit);
- }
-
- }
-
-
- /**
- * DMaaP MR Subscriber Consumer Id
- *
- * @return consumer Id
- */
- public String getConsumerId() {
- return consumerId;
- }
-
- /**
- * DMaaP MR Subscriber Consumer Group
- *
- * @return consumer group
- */
- public String getConsumerGroup() {
- return consumerGroup;
- }
-
- /**
- * DMaaP MR Subscriber Timeout in ms
- *
- * @return subscriber timeout ms
- */
- public Integer getTimeoutMS() {
- return timeoutMS;
- }
-
- /**
- * DMaaP MR Subscriber message limit
- *
- * @return subscriber message limit
- */
- public Integer getMessageLimit() {
- return messageLimit;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- DMaaPMRSubscriberConfig that = (DMaaPMRSubscriberConfig) o;
- return Objects.equal(consumerId, that.consumerId) &&
- Objects.equal(consumerGroup, that.consumerGroup) &&
- Objects.equal(timeoutMS, that.timeoutMS) &&
- Objects.equal(messageLimit, that.messageLimit);
- }
-
- @Override
- public int hashCode() {
- return Objects.hashCode(super.hashCode(), consumerId, consumerGroup, timeoutMS, messageLimit);
- }
-
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this)
- .add("baseConfig", super.toString())
- .add("consumerId", consumerId)
- .add("consumerGroup", consumerGroup)
- .add("timeoutMS", timeoutMS)
- .add("messageLimit", messageLimit)
- .toString();
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.config;
+
+import com.google.common.base.Objects;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+
+import java.util.UUID;
+
+import javax.annotation.Nonnull;
+
+/**
+ * <p>
+ * Immutable DMaaP MR Configuration for Subscriber.
+ * <p>
+ * Use {@link DMaaPMRSubscriberConfig.Builder} to construct Subscriber Configuration
+ * <p>
+ *
+ * @author Rajiv Singla . Creation Date: 10/12/2016.
+ */
+public final class DMaaPMRSubscriberConfig extends DMaaPMRBaseConfig {
+
+ private final String consumerId;
+ private final String consumerGroup;
+ private final Integer timeoutMS;
+ private final Integer messageLimit;
+
+ private DMaaPMRSubscriberConfig(@Nonnull String hostName,
+ @Nonnull Integer portNumber,
+ @Nonnull String topicName,
+ @Nonnull String protocol,
+ String userName,
+ String userPassword,
+ @Nonnull String contentType,
+ @Nonnull String consumerId,
+ @Nonnull String consumerGroup,
+ @Nonnull Integer timeoutMS,
+ @Nonnull Integer messageLimit) {
+ this.hostName = hostName;
+ this.portNumber = portNumber;
+ this.topicName = topicName;
+ this.protocol = protocol;
+ this.userName = userName;
+ this.userPassword = userPassword;
+ this.contentType = contentType;
+ this.consumerId = consumerId;
+ this.consumerGroup = consumerGroup;
+ this.timeoutMS = timeoutMS;
+ this.messageLimit = messageLimit;
+ }
+
+ /**
+ * Builder to initialize immutable {@link DMaaPMRSubscriberConfig} object
+ */
+ public static class Builder {
+
+ private String hostName;
+ private Integer portNumber;
+ private String topicName;
+ private String userName;
+ private String userPassword;
+ private String protocol;
+ private String contentType;
+ private String consumerId;
+ private String consumerGroup;
+ private Integer timeoutMS;
+ private Integer messageLimit;
+
+ public Builder(@Nonnull String hostName,
+ @Nonnull String topicName) {
+ // Required Values
+ this.hostName = hostName;
+ this.topicName = topicName;
+
+ // Default values
+ this.portNumber = AnalyticsConstants.DEFAULT_PORT_NUMBER;
+ this.userName = AnalyticsConstants.DEFAULT_USER_NAME;
+ this.userPassword = AnalyticsConstants.DEFAULT_USER_PASSWORD;
+ this.protocol = AnalyticsConstants.DEFAULT_PROTOCOL;
+ this.contentType = AnalyticsConstants.DEFAULT_CONTENT_TYPE;
+ this.consumerId = UUID.randomUUID().toString(); // consumer is assigned a random id by default
+ this.consumerGroup = AnalyticsConstants.DEFAULT_SUBSCRIBER_GROUP_PREFIX + consumerId; // random group
+ this.timeoutMS = AnalyticsConstants.DEFAULT_SUBSCRIBER_TIMEOUT_MS; // defaults to 10ms timeout
+ this.messageLimit = AnalyticsConstants.DEFAULT_SUBSCRIBER_MESSAGE_LIMIT; // defaults to 1000 message limit
+ }
+
+
+ /**
+ * Setup for custom host port number - Defaults to 80.
+ *
+ * @param portNumber custom port number
+ * @return Builder object itself for chaining
+ */
+ public Builder setPortNumber(@Nonnull Integer portNumber) {
+ this.portNumber = portNumber;
+ return this;
+ }
+
+
+ /**
+ * Setup user name for authentication. If no username is provided authentication will be disabled
+ *
+ * @param userName user name for DMaaP Topic Authentication
+ * @return Builder object itself for chaining
+ */
+ public Builder setUserName(@Nonnull String userName) {
+ this.userName = userName;
+ return this;
+ }
+
+
+ /**
+ * Setup user password for authentication. If no password is provided authentication will be disabled
+ *
+ * @param userPassword user password for DMaaP Topic Authentication
+ * @return Builder object itself for chaining
+ */
+ public Builder setUserPassword(@Nonnull String userPassword) {
+ this.userPassword = userPassword;
+ return this;
+ }
+
+
+ /**
+ * Setup custom Subscriber protocol - Defaults to https.
+ * Note: Only http and https are currently supported.
+ *
+ * @param protocol protocol e.g. https or http
+ * @return Builder object itself for chaining
+ */
+ public Builder setProtocol(@Nonnull String protocol) {
+
+ this.protocol = normalizeValidateProtocol(protocol);
+ return this;
+ }
+
+ /**
+ * Setup custom Subscriber content-type - Defaults to application/json
+ *
+ * @param contentType content type e.g. application/json
+ * @return Builder object itself for chaining
+ */
+ public Builder setContentType(@Nonnull String contentType) {
+ final String normalizedContentType = normalizeValidateContentType(contentType);
+ this.contentType = normalizedContentType;
+ return this;
+ }
+
+
+ /**
+ * Setup custom Consumer Id - Defaults to random Id
+ *
+ * @param consumerId - custom consumer ID
+ * @return Builder object itself for chaining
+ */
+ public Builder setConsumerId(@Nonnull String consumerId) {
+ this.consumerId = consumerId;
+ return this;
+ }
+
+ /**
+ * Setup custom Consumer Group - Default to OpenDCAE-DMaaPSub-ConsumerID
+ *
+ * @param consumerGroup - custom Consumer Group
+ * @return Builder object itself for chaining
+ */
+ public Builder setConsumerGroup(@Nonnull String consumerGroup) {
+ this.consumerGroup = consumerGroup;
+ return this;
+ }
+
+ /**
+ * Setup Custom Subscriber timeout in ms - Default to no timeout limit
+ *
+ * @param timeoutMS timeout in milliseconds
+ * @return Builder object itself for chaining
+ */
+ public Builder setTimeoutMS(@Nonnull Integer timeoutMS) {
+ this.timeoutMS = timeoutMS;
+ return this;
+ }
+
+ /**
+ * Setup custom Subscriber Message Limit - Default to no limit
+ *
+ * @param messageLimit message Limit
+ * @return Builder object itself for chaining
+ */
+ public Builder setMessageLimit(@Nonnull Integer messageLimit) {
+ this.messageLimit = messageLimit;
+ return this;
+ }
+
+ /**
+ * Builds Immutable instance of {@link DMaaPMRSubscriberConfig}
+ *
+ * @return immutable DMaaP Subscriber Config Object
+ */
+ public DMaaPMRSubscriberConfig build() {
+ return new DMaaPMRSubscriberConfig(hostName, portNumber, topicName, protocol, userName, userPassword,
+ contentType, consumerId, consumerGroup, timeoutMS, messageLimit);
+ }
+
+ }
+
+
+ /**
+ * DMaaP MR Subscriber Consumer Id
+ *
+ * @return consumer Id
+ */
+ public String getConsumerId() {
+ return consumerId;
+ }
+
+ /**
+ * DMaaP MR Subscriber Consumer Group
+ *
+ * @return consumer group
+ */
+ public String getConsumerGroup() {
+ return consumerGroup;
+ }
+
+ /**
+ * DMaaP MR Subscriber Timeout in ms
+ *
+ * @return subscriber timeout ms
+ */
+ public Integer getTimeoutMS() {
+ return timeoutMS;
+ }
+
+ /**
+ * DMaaP MR Subscriber message limit
+ *
+ * @return subscriber message limit
+ */
+ public Integer getMessageLimit() {
+ return messageLimit;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ DMaaPMRSubscriberConfig that = (DMaaPMRSubscriberConfig) o;
+ return Objects.equal(consumerId, that.consumerId) &&
+ Objects.equal(consumerGroup, that.consumerGroup) &&
+ Objects.equal(timeoutMS, that.timeoutMS) &&
+ Objects.equal(messageLimit, that.messageLimit);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(super.hashCode(), consumerId, consumerGroup, timeoutMS, messageLimit);
+ }
+
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(this)
+ .add("baseConfig", super.toString())
+ .add("consumerId", consumerId)
+ .add("consumerGroup", consumerGroup)
+ .add("timeoutMS", timeoutMS)
+ .add("messageLimit", messageLimit)
+ .toString();
+ }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java
index 845f6df..b4a29a1 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java
@@ -1,38 +1,38 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-/**
- * <p>
- * Contract for all DMaaPMR Publisher Response
- * <p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRPublisherResponse extends DMaaPMRResponse {
-
-
- /**
- * Gets number of pending messages
- *
- * @return pending messages in the batch queue
- */
- int getPendingMessagesCount();
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.response;
+
+/**
+ * <p>
+ * Contract for all DMaaPMR Publisher Response
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRPublisherResponse extends DMaaPMRResponse {
+
+
+ /**
+ * Gets number of pending messages
+ *
+ * @return pending messages in the batch queue
+ */
+ int getPendingMessagesCount();
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java
index 92996b4..879feac 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java
@@ -1,70 +1,70 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-import com.google.common.base.Objects;
-
-import javax.annotation.Nonnull;
-
-/**
- * <p>
- * An simple implementation of {@link DMaaPMRPublisherResponse}
- * <p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public class DMaaPMRPublisherResponseImpl implements DMaaPMRPublisherResponse {
-
- private final Integer responseCode;
- private final String responseMessage;
- private final int pendingMessagesCount;
-
- public DMaaPMRPublisherResponseImpl(@Nonnull Integer responseCode,
- @Nonnull String responseMessage,
- int pendingMessagesCount) {
- this.responseCode = responseCode;
- this.responseMessage = responseMessage;
- this.pendingMessagesCount = pendingMessagesCount;
- }
-
- @Override
- public Integer getResponseCode() {
- return responseCode;
- }
-
- @Override
- public String getResponseMessage() {
- return responseMessage;
- }
-
- @Override
- public int getPendingMessagesCount() {
- return pendingMessagesCount;
- }
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this)
- .add("responseCode", responseCode)
- .add("responseMessage", responseMessage)
- .add("pendingMessagesCount", pendingMessagesCount)
- .toString();
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.response;
+
+import com.google.common.base.Objects;
+
+import javax.annotation.Nonnull;
+
+/**
+ * <p>
+ * An simple implementation of {@link DMaaPMRPublisherResponse}
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public class DMaaPMRPublisherResponseImpl implements DMaaPMRPublisherResponse {
+
+ private final Integer responseCode;
+ private final String responseMessage;
+ private final int pendingMessagesCount;
+
+ public DMaaPMRPublisherResponseImpl(@Nonnull Integer responseCode,
+ @Nonnull String responseMessage,
+ int pendingMessagesCount) {
+ this.responseCode = responseCode;
+ this.responseMessage = responseMessage;
+ this.pendingMessagesCount = pendingMessagesCount;
+ }
+
+ @Override
+ public Integer getResponseCode() {
+ return responseCode;
+ }
+
+ @Override
+ public String getResponseMessage() {
+ return responseMessage;
+ }
+
+ @Override
+ public int getPendingMessagesCount() {
+ return pendingMessagesCount;
+ }
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(this)
+ .add("responseCode", responseCode)
+ .add("responseMessage", responseMessage)
+ .add("pendingMessagesCount", pendingMessagesCount)
+ .toString();
+ }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java
index 661a7e5..4ac5cf8 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java
@@ -1,46 +1,46 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-/**
- * <p>
- * Contract for DMaaP MR Responses
- * <p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRResponse {
-
- /**
- * Gets HTTP Response Code
- *
- * @return HTTP Response code as String
- */
- Integer getResponseCode();
-
- /**
- * Gets Response Message
- *
- * @return Response Message
- */
- String getResponseMessage();
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.response;
+
+/**
+ * <p>
+ * Contract for DMaaP MR Responses
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRResponse {
+
+ /**
+ * Gets HTTP Response Code
+ *
+ * @return HTTP Response code as String
+ */
+ Integer getResponseCode();
+
+ /**
+ * Gets Response Message
+ *
+ * @return Response Message
+ */
+ String getResponseMessage();
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java
index 228e2d3..5e90e24 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java
@@ -1,40 +1,40 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-import java.util.List;
-
-/**
- * <p>
- * Contract for all DMaaP MR Subscriber Responses
- * </p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRSubscriberResponse extends DMaaPMRResponse {
-
- /**
- * Returns message fetched from DMaaP MR Topic
- *
- * @return collection of actual message retrieved from DMaaP MR Topic
- */
- List<String> getFetchedMessages();
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.response;
+
+import java.util.List;
+
+/**
+ * <p>
+ * Contract for all DMaaP MR Subscriber Responses
+ * </p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRSubscriberResponse extends DMaaPMRResponse {
+
+ /**
+ * Returns message fetched from DMaaP MR Topic
+ *
+ * @return collection of actual message retrieved from DMaaP MR Topic
+ */
+ List<String> getFetchedMessages();
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java
index 08f0dab..160491c 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java
@@ -1,80 +1,80 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-import com.google.common.base.Objects;
-import com.google.common.collect.ImmutableList;
-
-import java.util.List;
-
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-
-import static java.util.Collections.unmodifiableList;
-
-/**
- * <p>
- * A simple implementation for {@link DMaaPMRSubscriberResponse}
- * <p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public class DMaaPMRSubscriberResponseImpl implements DMaaPMRSubscriberResponse {
-
- private final Integer responseCode;
- private final String responseMessage;
- private final List<String> fetchedMessages;
-
- public DMaaPMRSubscriberResponseImpl(@Nonnull Integer responseCode,
- @Nonnull String responseMessage,
- @Nullable List<String> fetchedMessages) {
- this.responseCode = responseCode;
- this.responseMessage = responseMessage;
- this.fetchedMessages = fetchedMessages != null ? fetchedMessages : ImmutableList.<String>of();
- }
-
- public DMaaPMRSubscriberResponseImpl(Integer responseCode, String responseMessage) {
- this(responseCode, responseMessage, null);
- }
-
- @Override
- public Integer getResponseCode() {
- return responseCode;
- }
-
- @Override
- public String getResponseMessage() {
- return responseMessage;
- }
-
- @Override
- public List<String> getFetchedMessages() {
- return unmodifiableList(fetchedMessages);
- }
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this)
- .add("responseCode", responseCode)
- .add("responseMessage", responseMessage)
- .add("fetchedMessages(size)", fetchedMessages.size())
- .toString();
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.domain.response;
+
+import com.google.common.base.Objects;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import static java.util.Collections.unmodifiableList;
+
+/**
+ * <p>
+ * A simple implementation for {@link DMaaPMRSubscriberResponse}
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public class DMaaPMRSubscriberResponseImpl implements DMaaPMRSubscriberResponse {
+
+ private final Integer responseCode;
+ private final String responseMessage;
+ private final List<String> fetchedMessages;
+
+ public DMaaPMRSubscriberResponseImpl(@Nonnull Integer responseCode,
+ @Nonnull String responseMessage,
+ @Nullable List<String> fetchedMessages) {
+ this.responseCode = responseCode;
+ this.responseMessage = responseMessage;
+ this.fetchedMessages = fetchedMessages != null ? fetchedMessages : ImmutableList.<String>of();
+ }
+
+ public DMaaPMRSubscriberResponseImpl(Integer responseCode, String responseMessage) {
+ this(responseCode, responseMessage, null);
+ }
+
+ @Override
+ public Integer getResponseCode() {
+ return responseCode;
+ }
+
+ @Override
+ public String getResponseMessage() {
+ return responseMessage;
+ }
+
+ @Override
+ public List<String> getFetchedMessages() {
+ return unmodifiableList(fetchedMessages);
+ }
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(this)
+ .add("responseCode", responseCode)
+ .add("responseMessage", responseMessage)
+ .add("fetchedMessages(size)", fetchedMessages.size())
+ .toString();
+ }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java
index e85b377..ab08642 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java
@@ -1,62 +1,62 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.module;
-
-import com.google.inject.AbstractModule;
-import com.google.inject.assistedinject.FactoryModuleBuilder;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClients;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherImpl;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueue;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueueFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueueImpl;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberImpl;
-
-/**
- * Guice Module to wire concrete implementations with interfaces
- * <p>
- * @author Rajiv Singla . Creation Date: 10/20/2016.
- */
-public class AnalyticsDMaaPModule extends AbstractModule {
-
-
- @Override
- protected void configure() {
-
- // Bind Http Client
- bind(CloseableHttpClient.class).toInstance(HttpClients.createDefault());
-
- // Bind Publishing queue
- install(new FactoryModuleBuilder().implement(DMaaPMRPublisherQueue.class, DMaaPMRPublisherQueueImpl.class)
- .build(DMaaPMRPublisherQueueFactory.class));
-
- install(new FactoryModuleBuilder().implement(DMaaPMRPublisher.class, DMaaPMRPublisherImpl.class)
- .build(DMaaPMRPublisherFactory.class));
-
- install(new FactoryModuleBuilder().implement(DMaaPMRSubscriber.class, DMaaPMRSubscriberImpl.class)
- .build(DMaaPMRSubscriberFactory.class));
-
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.module;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.assistedinject.FactoryModuleBuilder;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherFactory;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherImpl;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueue;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueueFactory;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueueImpl;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberFactory;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberImpl;
+
+/**
+ * Guice Module to wire concrete implementations with interfaces
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/20/2016.
+ */
+public class AnalyticsDMaaPModule extends AbstractModule {
+
+
+ @Override
+ protected void configure() {
+
+ // Bind Http Client
+ bind(CloseableHttpClient.class).toInstance(HttpClients.createDefault());
+
+ // Bind Publishing queue
+ install(new FactoryModuleBuilder().implement(DMaaPMRPublisherQueue.class, DMaaPMRPublisherQueueImpl.class)
+ .build(DMaaPMRPublisherQueueFactory.class));
+
+ install(new FactoryModuleBuilder().implement(DMaaPMRPublisher.class, DMaaPMRPublisherImpl.class)
+ .build(DMaaPMRPublisherFactory.class));
+
+ install(new FactoryModuleBuilder().implement(DMaaPMRSubscriber.class, DMaaPMRSubscriberImpl.class)
+ .build(DMaaPMRSubscriberFactory.class));
+
+ }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java
index 22f25d1..f381b5f 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java
@@ -1,378 +1,378 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Optional;
-import org.apache.commons.codec.binary.Base64;
-import org.apache.commons.lang3.StringEscapeUtils;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.ResponseHandler;
-import org.apache.http.client.utils.URIBuilder;
-import org.apache.http.util.EntityUtils;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponseImpl;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponseImpl;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.charset.Charset;
-import java.util.LinkedList;
-import java.util.List;
-
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-
-import static java.lang.String.format;
-
-/**
- * Base class for DMaaP MR Publishers and Subscriber Implementations containing various utility methods
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public abstract class BaseDMaaPMRComponent implements DMaaPMRComponent {
-
- private static final Logger LOG = LoggerFactory.getLogger(BaseDMaaPMRComponent.class);
-
- private static final ObjectMapper objectMapper = new ObjectMapper();
-
- /**
- * Creates Base64 encoded Auth Header for given userName and Password
- * If either user name of password are null return absent
- *
- * @param userName username
- * @param userPassword user password
- * @return base64 encoded auth header if username or password are both non null
- */
- protected static Optional<String> getAuthHeader(@Nullable final String userName,
- @Nullable final String userPassword) {
- if (userName == null || userPassword == null) {
- return Optional.absent();
- } else {
- final String auth = userName + ":" + userPassword;
- final Charset isoCharset = Charset.forName("ISO-8859-1");
- byte[] encodedAuth = Base64.encodeBase64(auth.getBytes(isoCharset));
- return Optional.of("Basic " + new String(encodedAuth, isoCharset));
- }
- }
-
-
- /**
- * Creates Publisher URI for given {@link DMaaPMRPublisherConfig}
- *
- * @param publisherConfig publisher settings
- *
- * @return DMaaP MR Publisher Topic URI that can be used to post messages to MR Topic
- */
- protected static URI createPublisherURI(final DMaaPMRPublisherConfig publisherConfig) {
- final String hostName = publisherConfig.getHostName();
- final Integer portNumber = publisherConfig.getPortNumber();
- final String getProtocol = publisherConfig.getProtocol();
- final String topicName = publisherConfig.getTopicName();
- URI publisherURI = null;
- try {
- publisherURI = new URIBuilder().setScheme(getProtocol).setHost(hostName).setPort(portNumber)
- .setPath(AnalyticsConstants.DMAAP_URI_PATH_PREFIX + topicName).build();
- } catch (URISyntaxException e) {
- final String errorMessage = format("Error while creating publisher URI: %s", e);
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
- LOG.info("Created DMaaP MR Publisher URI: {}", publisherURI);
- return publisherURI;
- }
-
-
- /**
- * Creates Subscriber URI for given {@link DMaaPMRSubscriberConfig}
- *
- * @param subscriberConfig subscriber settings
- *
- * @return DMaaP MR Subscriber Topic URI that can be used to fetch messages from MR topic
- */
- protected static URI createSubscriberURI(final DMaaPMRSubscriberConfig subscriberConfig) {
- final String hostName = subscriberConfig.getHostName();
- final Integer portNumber = subscriberConfig.getPortNumber();
- final String getProtocol = subscriberConfig.getProtocol();
- final String topicName = subscriberConfig.getTopicName();
- final String consumerId = subscriberConfig.getConsumerId();
- final String consumerGroup = subscriberConfig.getConsumerGroup();
- final Integer timeoutMS = subscriberConfig.getTimeoutMS();
- final Integer messageLimit = subscriberConfig.getMessageLimit();
- URI subscriberURI = null;
- try {
- URIBuilder uriBuilder = new URIBuilder().setScheme(getProtocol).setHost(hostName).setPort(portNumber)
- .setPath(AnalyticsConstants.DMAAP_URI_PATH_PREFIX
- + topicName + "/"
- + consumerGroup + "/" +
- consumerId);
- // add query params if present
- if (timeoutMS > 0) {
- uriBuilder.addParameter(AnalyticsConstants.SUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME, timeoutMS.toString());
- }
- if (messageLimit > 0) {
- uriBuilder.addParameter(AnalyticsConstants.SUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME,
- messageLimit.toString());
- }
- subscriberURI = uriBuilder.build();
-
- } catch (URISyntaxException e) {
- final String errorMessage = format("Error while creating subscriber URI: %s", e);
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
-
- LOG.info("Created DMaaP MR Subscriber URI: {}", subscriberURI);
- return subscriberURI;
- }
-
-
- /**
- * Creates 202 (Accepted) Response code message
- *
- * @param batchQueueSize batch Queue size
- *
- * @return response with 202 message code
- */
- protected static DMaaPMRPublisherResponse createPublisherAcceptedResponse(int batchQueueSize) {
- return createPublisherResponse(HTTPUtils.HTTP_ACCEPTED_RESPONSE_CODE,
- "Accepted - Messages queued for batch publishing to MR Topic", batchQueueSize);
- }
-
-
- /**
- * Creates 204 (No Content) Response code message
- *
- * @return response with 204 message code
- */
- protected static DMaaPMRPublisherResponse createPublisherNoContentResponse() {
- return createPublisherResponse(HTTPUtils.HTTP_NO_CONTENT_RESPONSE_CODE,
- "No Content - No Messages in batch queue for flushing to MR Topic", 0);
- }
-
-
- /**
- * Creates Publisher Response for given response code, response Message and pending Message Count
- *
- * @param responseCode HTTP Status Code
- * @param responseMessage response message
- * @param pendingMessages pending messages in batch queue
- *
- * @return DMaaP MR Publisher Response
- */
- protected static DMaaPMRPublisherResponse createPublisherResponse(int responseCode, String
- responseMessage, int pendingMessages) {
- return new DMaaPMRPublisherResponseImpl(responseCode, responseMessage, pendingMessages);
- }
-
-
- /**
- * Returns weekly consistent pending messages in batch queue
- *
- * @param publisherQueue batch queue
- * @param publisherConfig publisher settings
- *
- * @return pending messages to be published
- */
- protected static int getPendingMessages(@Nonnull final DMaaPMRPublisherQueue publisherQueue,
- @Nonnull final DMaaPMRPublisherConfig publisherConfig) {
- return publisherConfig.getMaxBatchSize() - publisherQueue.getBatchQueueRemainingSize();
- }
-
-
- /**
- * Creates Subscriber Response for give response Code, response Message and fetch messages
- *
- * @param responseCode response Code
- * @param responseMessage response Message
- * @param fetchedMessages fetched messages
- *
- * @return DMaaP MR Subscriber Response
- */
- protected static DMaaPMRSubscriberResponse createSubscriberResponse(int responseCode, String
- responseMessage, List<String> fetchedMessages) {
- if (fetchedMessages == null) {
- return new DMaaPMRSubscriberResponseImpl(responseCode, responseMessage);
- } else {
- return new DMaaPMRSubscriberResponseImpl(responseCode, responseMessage, fetchedMessages);
- }
- }
-
-
- /**
- * Custom response handler which extract status code and response body
- *
- * @return Pair containing Response code and response body
- */
- protected static ResponseHandler<Pair<Integer, String>> responseHandler() {
- return new ResponseHandler<Pair<Integer, String>>() {
- @Override
- public Pair<Integer, String> handleResponse(HttpResponse response) throws IOException {
- // Get Response status code
- final int status = response.getStatusLine().getStatusCode();
- final HttpEntity responseEntity = response.getEntity();
- // If response entity is not null - extract response body as string
- String responseEntityString = "";
- if (responseEntity != null) {
- responseEntityString = EntityUtils.toString(responseEntity);
- }
- return new ImmutablePair<>(status, responseEntityString);
- }
- };
- }
-
-
- /**
- * Adds message to Publisher recovery queue. If recovery queue is full throws an error as messages will
- * be lost
- *
- * @param publisherQueue publisher queue
- * @param messages recoverable messages to be published to recovery queue
- */
- protected static void addMessagesToRecoveryQueue(DMaaPMRPublisherQueue publisherQueue,
- List<String> messages) {
- try {
- publisherQueue.addRecoverableMessages(messages);
-
- LOG.debug("Messages Added to Recovery Queue. Messages Size: {}, Recovery Queue Remaining Size: {}",
- messages.size(), publisherQueue.getBatchQueueRemainingSize());
-
- } catch (IllegalStateException e) {
- final String errorMessage = format("Unable to put messages in recovery queue. Messages will be lost. " +
- "Recovery Queue might be full. Message Size: %d, Recovery Queue Remaining Capacity: %d",
- messages.size(), publisherQueue.getRecoveryQueueRemainingSize());
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
- }
-
-
- /**
- * Converts List of messages to Json String Array which can be published to DMaaP MR topic.
- *
- * @param messages messages that need to parsed to Json Array representation
- * @return json string representation of message
- */
- protected static String convertToJsonString(@Nullable final List<String> messages) {
- // If messages are null or empty just return empty array
- if (messages == null || messages.isEmpty()) {
- return "[]";
- }
-
-
- List<JsonNode> jsonMessageObjectsList = new LinkedList<>();
-
- try {
- for (String message : messages) {
- final JsonNode jsonNode = objectMapper.readTree(message);
- jsonMessageObjectsList.add(jsonNode);
- }
- return objectMapper.writeValueAsString(jsonMessageObjectsList);
- } catch (JsonProcessingException e) {
- final String errorMessage =
- format("Unable to convert publisher messages to Json. Messages: %s, Json Error: %s",
- messages, e);
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-
- } catch (IOException e) {
- final String errorMessage =
- format("IO Exception while converting publisher messages to Json. Messages: %s, Json Error: %s",
- messages, e);
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
- }
-
-
- /**
- * Converts subscriber messages json string to List of messages. If message Json String is empty
- * or null
- *
- * @param messagesJsonString json messages String
- *
- * @return List containing DMaaP MR Messages
- */
- protected static List<String> convertJsonToStringMessages(@Nullable final String messagesJsonString) {
-
- final LinkedList<String> messages = new LinkedList<>();
-
- // If message string is not null or not empty parse json message array to List of string messages
- if (messagesJsonString != null && !messagesJsonString.trim().isEmpty()
- && !("[]").equals(messagesJsonString.trim())) {
-
- try {
- // get root node
- final JsonNode rootNode = objectMapper.readTree(messagesJsonString);
- // iterate over root node and parse arrays messages
- for (JsonNode jsonNode : rootNode) {
- // if array parse it is array of messages
- final String incomingMessageString = jsonNode.toString();
- if (jsonNode.isArray()) {
- final List messageList = objectMapper.readValue(incomingMessageString, List.class);
- for (Object message : messageList) {
- final String jsonMessageString = objectMapper.writeValueAsString(message);
- addUnescapedJsonToMessage(messages, jsonMessageString);
- }
- } else {
- // parse it as object
- addUnescapedJsonToMessage(messages, incomingMessageString);
- }
- }
-
- } catch (IOException e) {
- final String errorMessage =
- format("Unable to convert subscriber Json String to Messages. Subscriber Response String: %s," +
- " Json Error: %s", messagesJsonString, e);
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
-
- }
- return messages;
- }
-
- /**
- * Adds unescaped Json messages to given messages list
- *
- * @param messages message list in which unescaped messages will be added
- * @param incomingMessageString incoming message string that may need to be escaped
- */
- private static void addUnescapedJsonToMessage(List<String> messages, String incomingMessageString) {
- if (incomingMessageString.startsWith("\"") && incomingMessageString.endsWith("\"")) {
- messages.add(StringEscapeUtils.unescapeJson(
- incomingMessageString.substring(1, incomingMessageString.length() - 1)));
- } else {
- messages.add(StringEscapeUtils.unescapeJson(incomingMessageString));
- }
- }
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.lang3.StringEscapeUtils;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.ResponseHandler;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.util.EntityUtils;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.common.utils.HTTPUtils;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponseImpl;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponseImpl;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.Charset;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import static java.lang.String.format;
+
+/**
+ * Base class for DMaaP MR Publishers and Subscriber Implementations containing various utility methods
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public abstract class BaseDMaaPMRComponent implements DMaaPMRComponent {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BaseDMaaPMRComponent.class);
+
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ /**
+ * Creates Base64 encoded Auth Header for given userName and Password
+ * If either user name of password are null return absent
+ *
+ * @param userName username
+ * @param userPassword user password
+ * @return base64 encoded auth header if username or password are both non null
+ */
+ protected static Optional<String> getAuthHeader(@Nullable final String userName,
+ @Nullable final String userPassword) {
+ if (userName == null || userPassword == null) {
+ return Optional.absent();
+ } else {
+ final String auth = userName + ":" + userPassword;
+ final Charset isoCharset = Charset.forName("ISO-8859-1");
+ byte[] encodedAuth = Base64.encodeBase64(auth.getBytes(isoCharset));
+ return Optional.of("Basic " + new String(encodedAuth, isoCharset));
+ }
+ }
+
+
+ /**
+ * Creates Publisher URI for given {@link DMaaPMRPublisherConfig}
+ *
+ * @param publisherConfig publisher settings
+ *
+ * @return DMaaP MR Publisher Topic URI that can be used to post messages to MR Topic
+ */
+ protected static URI createPublisherURI(final DMaaPMRPublisherConfig publisherConfig) {
+ final String hostName = publisherConfig.getHostName();
+ final Integer portNumber = publisherConfig.getPortNumber();
+ final String getProtocol = publisherConfig.getProtocol();
+ final String topicName = publisherConfig.getTopicName();
+ URI publisherURI = null;
+ try {
+ publisherURI = new URIBuilder().setScheme(getProtocol).setHost(hostName).setPort(portNumber)
+ .setPath(AnalyticsConstants.DMAAP_URI_PATH_PREFIX + topicName).build();
+ } catch (URISyntaxException e) {
+ final String errorMessage = format("Error while creating publisher URI: %s", e);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+ LOG.info("Created DMaaP MR Publisher URI: {}", publisherURI);
+ return publisherURI;
+ }
+
+
+ /**
+ * Creates Subscriber URI for given {@link DMaaPMRSubscriberConfig}
+ *
+ * @param subscriberConfig subscriber settings
+ *
+ * @return DMaaP MR Subscriber Topic URI that can be used to fetch messages from MR topic
+ */
+ protected static URI createSubscriberURI(final DMaaPMRSubscriberConfig subscriberConfig) {
+ final String hostName = subscriberConfig.getHostName();
+ final Integer portNumber = subscriberConfig.getPortNumber();
+ final String getProtocol = subscriberConfig.getProtocol();
+ final String topicName = subscriberConfig.getTopicName();
+ final String consumerId = subscriberConfig.getConsumerId();
+ final String consumerGroup = subscriberConfig.getConsumerGroup();
+ final Integer timeoutMS = subscriberConfig.getTimeoutMS();
+ final Integer messageLimit = subscriberConfig.getMessageLimit();
+ URI subscriberURI = null;
+ try {
+ URIBuilder uriBuilder = new URIBuilder().setScheme(getProtocol).setHost(hostName).setPort(portNumber)
+ .setPath(AnalyticsConstants.DMAAP_URI_PATH_PREFIX
+ + topicName + "/"
+ + consumerGroup + "/" +
+ consumerId);
+ // add query params if present
+ if (timeoutMS > 0) {
+ uriBuilder.addParameter(AnalyticsConstants.SUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME, timeoutMS.toString());
+ }
+ if (messageLimit > 0) {
+ uriBuilder.addParameter(AnalyticsConstants.SUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME,
+ messageLimit.toString());
+ }
+ subscriberURI = uriBuilder.build();
+
+ } catch (URISyntaxException e) {
+ final String errorMessage = format("Error while creating subscriber URI: %s", e);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+
+ LOG.info("Created DMaaP MR Subscriber URI: {}", subscriberURI);
+ return subscriberURI;
+ }
+
+
+ /**
+ * Creates 202 (Accepted) Response code message
+ *
+ * @param batchQueueSize batch Queue size
+ *
+ * @return response with 202 message code
+ */
+ protected static DMaaPMRPublisherResponse createPublisherAcceptedResponse(int batchQueueSize) {
+ return createPublisherResponse(HTTPUtils.HTTP_ACCEPTED_RESPONSE_CODE,
+ "Accepted - Messages queued for batch publishing to MR Topic", batchQueueSize);
+ }
+
+
+ /**
+ * Creates 204 (No Content) Response code message
+ *
+ * @return response with 204 message code
+ */
+ protected static DMaaPMRPublisherResponse createPublisherNoContentResponse() {
+ return createPublisherResponse(HTTPUtils.HTTP_NO_CONTENT_RESPONSE_CODE,
+ "No Content - No Messages in batch queue for flushing to MR Topic", 0);
+ }
+
+
+ /**
+ * Creates Publisher Response for given response code, response Message and pending Message Count
+ *
+ * @param responseCode HTTP Status Code
+ * @param responseMessage response message
+ * @param pendingMessages pending messages in batch queue
+ *
+ * @return DMaaP MR Publisher Response
+ */
+ protected static DMaaPMRPublisherResponse createPublisherResponse(int responseCode, String
+ responseMessage, int pendingMessages) {
+ return new DMaaPMRPublisherResponseImpl(responseCode, responseMessage, pendingMessages);
+ }
+
+
+ /**
+ * Returns weekly consistent pending messages in batch queue
+ *
+ * @param publisherQueue batch queue
+ * @param publisherConfig publisher settings
+ *
+ * @return pending messages to be published
+ */
+ protected static int getPendingMessages(@Nonnull final DMaaPMRPublisherQueue publisherQueue,
+ @Nonnull final DMaaPMRPublisherConfig publisherConfig) {
+ return publisherConfig.getMaxBatchSize() - publisherQueue.getBatchQueueRemainingSize();
+ }
+
+
+ /**
+ * Creates Subscriber Response for give response Code, response Message and fetch messages
+ *
+ * @param responseCode response Code
+ * @param responseMessage response Message
+ * @param fetchedMessages fetched messages
+ *
+ * @return DMaaP MR Subscriber Response
+ */
+ protected static DMaaPMRSubscriberResponse createSubscriberResponse(int responseCode, String
+ responseMessage, List<String> fetchedMessages) {
+ if (fetchedMessages == null) {
+ return new DMaaPMRSubscriberResponseImpl(responseCode, responseMessage);
+ } else {
+ return new DMaaPMRSubscriberResponseImpl(responseCode, responseMessage, fetchedMessages);
+ }
+ }
+
+
+ /**
+ * Custom response handler which extract status code and response body
+ *
+ * @return Pair containing Response code and response body
+ */
+ protected static ResponseHandler<Pair<Integer, String>> responseHandler() {
+ return new ResponseHandler<Pair<Integer, String>>() {
+ @Override
+ public Pair<Integer, String> handleResponse(HttpResponse response) throws IOException {
+ // Get Response status code
+ final int status = response.getStatusLine().getStatusCode();
+ final HttpEntity responseEntity = response.getEntity();
+ // If response entity is not null - extract response body as string
+ String responseEntityString = "";
+ if (responseEntity != null) {
+ responseEntityString = EntityUtils.toString(responseEntity);
+ }
+ return new ImmutablePair<>(status, responseEntityString);
+ }
+ };
+ }
+
+
+ /**
+ * Adds message to Publisher recovery queue. If recovery queue is full throws an error as messages will
+ * be lost
+ *
+ * @param publisherQueue publisher queue
+ * @param messages recoverable messages to be published to recovery queue
+ */
+ protected static void addMessagesToRecoveryQueue(DMaaPMRPublisherQueue publisherQueue,
+ List<String> messages) {
+ try {
+ publisherQueue.addRecoverableMessages(messages);
+
+ LOG.debug("Messages Added to Recovery Queue. Messages Size: {}, Recovery Queue Remaining Size: {}",
+ messages.size(), publisherQueue.getBatchQueueRemainingSize());
+
+ } catch (IllegalStateException e) {
+ final String errorMessage = format("Unable to put messages in recovery queue. Messages will be lost. " +
+ "Recovery Queue might be full. Message Size: %d, Recovery Queue Remaining Capacity: %d",
+ messages.size(), publisherQueue.getRecoveryQueueRemainingSize());
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+ }
+
+
+ /**
+ * Converts List of messages to Json String Array which can be published to DMaaP MR topic.
+ *
+ * @param messages messages that need to parsed to Json Array representation
+ * @return json string representation of message
+ */
+ protected static String convertToJsonString(@Nullable final List<String> messages) {
+ // If messages are null or empty just return empty array
+ if (messages == null || messages.isEmpty()) {
+ return "[]";
+ }
+
+
+ List<JsonNode> jsonMessageObjectsList = new LinkedList<>();
+
+ try {
+ for (String message : messages) {
+ final JsonNode jsonNode = objectMapper.readTree(message);
+ jsonMessageObjectsList.add(jsonNode);
+ }
+ return objectMapper.writeValueAsString(jsonMessageObjectsList);
+ } catch (JsonProcessingException e) {
+ final String errorMessage =
+ format("Unable to convert publisher messages to Json. Messages: %s, Json Error: %s",
+ messages, e);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+
+ } catch (IOException e) {
+ final String errorMessage =
+ format("IO Exception while converting publisher messages to Json. Messages: %s, Json Error: %s",
+ messages, e);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+ }
+
+
+ /**
+ * Converts subscriber messages json string to List of messages. If message Json String is empty
+ * or null
+ *
+ * @param messagesJsonString json messages String
+ *
+ * @return List containing DMaaP MR Messages
+ */
+ protected static List<String> convertJsonToStringMessages(@Nullable final String messagesJsonString) {
+
+ final LinkedList<String> messages = new LinkedList<>();
+
+ // If message string is not null or not empty parse json message array to List of string messages
+ if (messagesJsonString != null && !messagesJsonString.trim().isEmpty()
+ && !("[]").equals(messagesJsonString.trim())) {
+
+ try {
+ // get root node
+ final JsonNode rootNode = objectMapper.readTree(messagesJsonString);
+ // iterate over root node and parse arrays messages
+ for (JsonNode jsonNode : rootNode) {
+ // if array parse it is array of messages
+ final String incomingMessageString = jsonNode.toString();
+ if (jsonNode.isArray()) {
+ final List messageList = objectMapper.readValue(incomingMessageString, List.class);
+ for (Object message : messageList) {
+ final String jsonMessageString = objectMapper.writeValueAsString(message);
+ addUnescapedJsonToMessage(messages, jsonMessageString);
+ }
+ } else {
+ // parse it as object
+ addUnescapedJsonToMessage(messages, incomingMessageString);
+ }
+ }
+
+ } catch (IOException e) {
+ final String errorMessage =
+ format("Unable to convert subscriber Json String to Messages. Subscriber Response String: %s," +
+ " Json Error: %s", messagesJsonString, e);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+
+ }
+ return messages;
+ }
+
+ /**
+ * Adds unescaped Json messages to given messages list
+ *
+ * @param messages message list in which unescaped messages will be added
+ * @param incomingMessageString incoming message string that may need to be escaped
+ */
+ private static void addUnescapedJsonToMessage(List<String> messages, String incomingMessageString) {
+ if (incomingMessageString.startsWith("\"") && incomingMessageString.endsWith("\"")) {
+ messages.add(StringEscapeUtils.unescapeJson(
+ incomingMessageString.substring(1, incomingMessageString.length() - 1)));
+ } else {
+ messages.add(StringEscapeUtils.unescapeJson(incomingMessageString));
+ }
+ }
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java
index 5c0df09..872c4fe 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java
@@ -1,29 +1,29 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service;
-
-/**
- * Marker interface for all DMaaP MR Components e.g. MR Publishers, MR Subscribers
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public interface DMaaPMRComponent {
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service;
+
+/**
+ * Marker interface for all DMaaP MR Components e.g. MR Publishers, MR Subscribers
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public interface DMaaPMRComponent {
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java
index 48b0a70..04186c1 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java
@@ -1,95 +1,95 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
-
-import java.util.Date;
-import java.util.List;
-
-/**
- * <p>
- * DMaaP MR Publisher can be used to publish messages to DMaaP MR Topics.
- * <p>
- *
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRPublisher extends AutoCloseable {
-
-
- /**
- * <p>
- * Adds collection of messages to DMaaP MR Topic Publishing Queue.
- * <p>
- * Note: Invoking this method may or may not cause publishing immediately
- * as publishing in done is batch mode by default. Parameter maxBatchSize
- * in {@link DMaaPMRPublisherConfig} is used to determine max batch queue size.
- * If the maxBatchSize is reached all message will be published automatically
- * during subsequent call.
- * </p>
- *
- * @param messages messages to publish to DMaaP MR Publisher
- * @return response which may contain Http Response code 202 (Accepted) as publishing
- * will proceed when max batch size is reached. Throws {@link DCAEAnalyticsRuntimeException}
- * if publishing fails
- */
- DMaaPMRPublisherResponse publish(List<String> messages);
-
-
- /**
- * <p>
- * Forces publishing of messages to DMaaP MR Topic and returns {@link DMaaPMRPublisherResponse}
- * which can be inspected for HTTP status code of publishing call to DMaaP MR Topic.
- * </p>
- *
- * @param messages messages to publish to DMaaP MR Publisher
- * @return DMaaP Message Router Publisher Response. Throws {@link DCAEAnalyticsRuntimeException}
- * if force publishing fails
- *
- */
- DMaaPMRPublisherResponse forcePublish(List<String> messages);
-
-
- /**
- * <p>
- * Forces publishing of messages in Publisher queue to DMaaP MR Topic and returns
- * {@link DMaaPMRPublisherResponse}.If there are no messages were in the queue to
- * be flushed response code 304 (Not Modified) will be returned
- * </p>
- *
- * @return DMaaP Message Router Publisher Response
- */
- DMaaPMRPublisherResponse flush();
-
-
- /**
- * <p>
- * Returns the creation time when Publisher instance was created.
- * <p>
- *
- * @return creation time of Subscriber instance
- */
- Date getPublisherCreationTime();
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.publisher;
+
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
+
+import java.util.Date;
+import java.util.List;
+
+/**
+ * <p>
+ * DMaaP MR Publisher can be used to publish messages to DMaaP MR Topics.
+ * <p>
+ *
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRPublisher extends AutoCloseable {
+
+
+ /**
+ * <p>
+ * Adds collection of messages to DMaaP MR Topic Publishing Queue.
+ * <p>
+ * Note: Invoking this method may or may not cause publishing immediately
+ * as publishing in done is batch mode by default. Parameter maxBatchSize
+ * in {@link DMaaPMRPublisherConfig} is used to determine max batch queue size.
+ * If the maxBatchSize is reached all message will be published automatically
+ * during subsequent call.
+ * </p>
+ *
+ * @param messages messages to publish to DMaaP MR Publisher
+ * @return response which may contain Http Response code 202 (Accepted) as publishing
+ * will proceed when max batch size is reached. Throws {@link DCAEAnalyticsRuntimeException}
+ * if publishing fails
+ */
+ DMaaPMRPublisherResponse publish(List<String> messages);
+
+
+ /**
+ * <p>
+ * Forces publishing of messages to DMaaP MR Topic and returns {@link DMaaPMRPublisherResponse}
+ * which can be inspected for HTTP status code of publishing call to DMaaP MR Topic.
+ * </p>
+ *
+ * @param messages messages to publish to DMaaP MR Publisher
+ * @return DMaaP Message Router Publisher Response. Throws {@link DCAEAnalyticsRuntimeException}
+ * if force publishing fails
+ *
+ */
+ DMaaPMRPublisherResponse forcePublish(List<String> messages);
+
+
+ /**
+ * <p>
+ * Forces publishing of messages in Publisher queue to DMaaP MR Topic and returns
+ * {@link DMaaPMRPublisherResponse}.If there are no messages were in the queue to
+ * be flushed response code 304 (Not Modified) will be returned
+ * </p>
+ *
+ * @return DMaaP Message Router Publisher Response
+ */
+ DMaaPMRPublisherResponse flush();
+
+
+ /**
+ * <p>
+ * Returns the creation time when Publisher instance was created.
+ * <p>
+ *
+ * @return creation time of Subscriber instance
+ */
+ Date getPublisherCreationTime();
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java
index 5d6dfa1..3b55b57 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java
@@ -1,49 +1,49 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-
-/**
- * <p>
- * Factory to initialize instance of {@link DMaaPMRPublisher} for Guice DI injection purposes.
- * <p>
- * <strong>
- * NOTE: Client should not use this Factory to initialize {@link DMaaPMRPublisher} unless they
- * are wiring dependencies using Guice. Client must use {@link DMaaPMRFactory} to initialize
- * guice injected Publisher instances
- * </strong>
- * <p>
- * @author Rajiv Singla . Creation Date: 10/20/2016.
- */
-public interface DMaaPMRPublisherFactory {
-
- /**
- * Guice Factory to create DMaaP MR Publisher
- *
- * @param publisherConfig publisher config
- *
- * @return DMaaP MR Publisher instance
- */
- DMaaPMRPublisher create(DMaaPMRPublisherConfig publisherConfig);
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.publisher;
+
+import org.onap.dcae.apod.analytics.dmaap.DMaaPMRFactory;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+
+/**
+ * <p>
+ * Factory to initialize instance of {@link DMaaPMRPublisher} for Guice DI injection purposes.
+ * <p>
+ * <strong>
+ * NOTE: Client should not use this Factory to initialize {@link DMaaPMRPublisher} unless they
+ * are wiring dependencies using Guice. Client must use {@link DMaaPMRFactory} to initialize
+ * guice injected Publisher instances
+ * </strong>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/20/2016.
+ */
+public interface DMaaPMRPublisherFactory {
+
+ /**
+ * Guice Factory to create DMaaP MR Publisher
+ *
+ * @param publisherConfig publisher config
+ *
+ * @return DMaaP MR Publisher instance
+ */
+ DMaaPMRPublisher create(DMaaPMRPublisherConfig publisherConfig);
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java
index 32e09fa..16ea5fa 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java
@@ -1,210 +1,210 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import com.google.common.base.Optional;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import com.google.inject.Inject;
-import com.google.inject.assistedinject.Assisted;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.http.HttpHeaders;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.Date;
-import java.util.List;
-
-import static org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
-import static java.lang.String.format;
-
-/**
- * Concrete Implementation of {@link DMaaPMRPublisher} which uses {@link HttpClient}
- *
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public class DMaaPMRPublisherImpl extends BaseDMaaPMRComponent implements DMaaPMRPublisher {
-
- private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherImpl.class);
-
- private final DMaaPMRPublisherConfig publisherConfig;
- private final CloseableHttpClient closeableHttpClient;
- private final DMaaPMRPublisherQueue publisherQueue;
- private final Date publisherCreationTime;
- private URI publisherUri;
-
- @Inject
- public DMaaPMRPublisherImpl(@Assisted DMaaPMRPublisherConfig publisherConfig,
- DMaaPMRPublisherQueueFactory dMaaPMRPublisherQueueFactory,
- CloseableHttpClient closeableHttpClient) {
-
- this.publisherConfig = publisherConfig;
- final int maxBatchSize = publisherConfig.getMaxBatchSize() > 0 ? publisherConfig.getMaxBatchSize() : 1;
- this.publisherQueue = dMaaPMRPublisherQueueFactory.create(
- maxBatchSize, publisherConfig.getMaxRecoveryQueueSize());
- this.closeableHttpClient = closeableHttpClient;
- this.publisherUri = createPublisherURI(publisherConfig);
- this.publisherCreationTime = new Date();
- }
-
-
- @Override
- public DMaaPMRPublisherResponse publish(List<String> messages) {
-
- final int batchQueueRemainingSize = publisherQueue.getBatchQueueRemainingSize();
-
- // if messages size is less than batch queue size - just queue them for batch publishing
- if (batchQueueRemainingSize > messages.size()) {
- LOG.debug("Adding messages to batch Queue. No flushing required. Messages Size:{}. Batch Queue Size:{}",
- messages.size(), batchQueueRemainingSize);
- final int batchQueueSize = publisherQueue.addBatchMessages(messages);
- return createPublisherAcceptedResponse(batchQueueSize);
-
- } else {
-
- // grab all already queued messages, append current messages and force publish them to DMaaP MR topic
- final List<String> queueMessages = publisherQueue.getMessageForPublishing();
- LOG.debug("Batch Queue capacity exceeds messages size. Flushing of all pending messages to DMaaP MR " +
- "Publisher Topic.");
- return forcePublish(Lists.newLinkedList(Iterables.concat(queueMessages, messages)));
- }
-
- }
-
- @Override
- public DMaaPMRPublisherResponse forcePublish(List<String> messages) {
-
- LOG.debug("Force publishing messages to DMaaP MR Topic. Messages Size: {}", messages.size());
-
- final String contentType = publisherConfig.getContentType();
- final String userName = publisherConfig.getUserName();
- final String userPassword = publisherConfig.getUserPassword();
- final HttpPost postRequest = new HttpPost(publisherUri);
-
- // add Authorization Header if username and password are present
- final Optional<String> authHeader = getAuthHeader(userName, userPassword);
- if (authHeader.isPresent()) {
- postRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
- } else {
- LOG.debug("DMaaP MR Publisher Authentication is disabled as username or password is not present.");
- }
-
- // Create post string entity
- final String messagesJson = convertToJsonString(messages);
- final StringEntity requestEntity =
- new StringEntity(messagesJson, ContentType.create(contentType, "UTF-8"));
- postRequest.setEntity(requestEntity);
-
- try {
- final Pair<Integer, String> responsePair = closeableHttpClient.execute(postRequest, responseHandler());
- final Integer responseCode = responsePair.getLeft();
- final String responseBody = responsePair.getRight();
- // if messages were published successfully, return successful response
- if (isSuccessfulResponseCode(responseCode)) {
- LOG.debug("DMaaP MR Messages published successfully. DMaaP Response Code: {}. DMaaP Response " +
- "Body: {}, Number of Messages published: {}",
- responseCode, responseBody, messages.size());
-
- } else {
- LOG.warn("Unable to publish messages to DMaaP MR Topic. DMaaP Response Code: {}, DMaaP Response " +
- "Body: {}. Messages will be queued in recovery queue", responseCode, responseBody);
- addMessagesToRecoveryQueue(publisherQueue, messages);
- }
-
- return createPublisherResponse(responseCode, responseBody,
- getPendingMessages(publisherQueue, publisherConfig));
-
- } catch (IOException e) {
- // If IO Error then we need to also put messages in recovery queue
- addMessagesToRecoveryQueue(publisherQueue, messages);
- final String errorMessage = format("IO Exception while publishing messages to DMaaP Topic. " +
- "Messages will be queued in recovery queue. Messages Size: %d", messages.size());
-
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
-
- }
-
-
- @Override
- public DMaaPMRPublisherResponse flush() {
- final List<String> queueMessages = publisherQueue.getMessageForPublishing();
- // If there are no message return 204 (No Content) response code
- if (queueMessages.isEmpty()) {
- LOG.debug("No messages to publish to batch queue. Returning 204 status code");
- return createPublisherNoContentResponse();
- } else {
- // force publish messages in queue
- return forcePublish(queueMessages);
- }
- }
-
- @Override
- public Date getPublisherCreationTime() {
- return new Date(publisherCreationTime.getTime());
- }
-
- @Override
- public void close() throws Exception {
-
- // flush current message in the queue
- int retrialNumber = 0;
- int flushResponseCode;
-
- // automatic retries if messages cannot be flushed
- do {
- retrialNumber++;
- DMaaPMRPublisherResponse flushResponse = flush();
- flushResponseCode = flushResponse.getResponseCode();
-
- if (!isSuccessfulResponseCode(flushResponseCode)) {
- LOG.warn("Unable to flush batch messages to publisher due to DMaaP MR invalid Response: {}. " +
- "Retrial No: {} of Max {} Retries", flushResponseCode, retrialNumber,
- AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE);
-
- Thread.sleep(AnalyticsConstants.PUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE);
- }
- } while (retrialNumber <= AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE &&
- !isSuccessfulResponseCode(flushResponseCode));
-
- if (!isSuccessfulResponseCode(flushResponseCode)) {
- LOG.error("Unable to flush batch messages to publisher. Messages loss cannot be prevented");
- } else {
- LOG.info("Successfully published all batched messages to publisher.");
- }
-
- // close http client
- closeableHttpClient.close();
-
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.publisher;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHeaders;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
+import org.onap.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Date;
+import java.util.List;
+
+import static org.onap.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
+import static java.lang.String.format;
+
+/**
+ * Concrete Implementation of {@link DMaaPMRPublisher} which uses {@link HttpClient}
+ *
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public class DMaaPMRPublisherImpl extends BaseDMaaPMRComponent implements DMaaPMRPublisher {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherImpl.class);
+
+ private final DMaaPMRPublisherConfig publisherConfig;
+ private final CloseableHttpClient closeableHttpClient;
+ private final DMaaPMRPublisherQueue publisherQueue;
+ private final Date publisherCreationTime;
+ private URI publisherUri;
+
+ @Inject
+ public DMaaPMRPublisherImpl(@Assisted DMaaPMRPublisherConfig publisherConfig,
+ DMaaPMRPublisherQueueFactory dMaaPMRPublisherQueueFactory,
+ CloseableHttpClient closeableHttpClient) {
+
+ this.publisherConfig = publisherConfig;
+ final int maxBatchSize = publisherConfig.getMaxBatchSize() > 0 ? publisherConfig.getMaxBatchSize() : 1;
+ this.publisherQueue = dMaaPMRPublisherQueueFactory.create(
+ maxBatchSize, publisherConfig.getMaxRecoveryQueueSize());
+ this.closeableHttpClient = closeableHttpClient;
+ this.publisherUri = createPublisherURI(publisherConfig);
+ this.publisherCreationTime = new Date();
+ }
+
+
+ @Override
+ public DMaaPMRPublisherResponse publish(List<String> messages) {
+
+ final int batchQueueRemainingSize = publisherQueue.getBatchQueueRemainingSize();
+
+ // if messages size is less than batch queue size - just queue them for batch publishing
+ if (batchQueueRemainingSize > messages.size()) {
+ LOG.debug("Adding messages to batch Queue. No flushing required. Messages Size:{}. Batch Queue Size:{}",
+ messages.size(), batchQueueRemainingSize);
+ final int batchQueueSize = publisherQueue.addBatchMessages(messages);
+ return createPublisherAcceptedResponse(batchQueueSize);
+
+ } else {
+
+ // grab all already queued messages, append current messages and force publish them to DMaaP MR topic
+ final List<String> queueMessages = publisherQueue.getMessageForPublishing();
+ LOG.debug("Batch Queue capacity exceeds messages size. Flushing of all pending messages to DMaaP MR " +
+ "Publisher Topic.");
+ return forcePublish(Lists.newLinkedList(Iterables.concat(queueMessages, messages)));
+ }
+
+ }
+
+ @Override
+ public DMaaPMRPublisherResponse forcePublish(List<String> messages) {
+
+ LOG.debug("Force publishing messages to DMaaP MR Topic. Messages Size: {}", messages.size());
+
+ final String contentType = publisherConfig.getContentType();
+ final String userName = publisherConfig.getUserName();
+ final String userPassword = publisherConfig.getUserPassword();
+ final HttpPost postRequest = new HttpPost(publisherUri);
+
+ // add Authorization Header if username and password are present
+ final Optional<String> authHeader = getAuthHeader(userName, userPassword);
+ if (authHeader.isPresent()) {
+ postRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
+ } else {
+ LOG.debug("DMaaP MR Publisher Authentication is disabled as username or password is not present.");
+ }
+
+ // Create post string entity
+ final String messagesJson = convertToJsonString(messages);
+ final StringEntity requestEntity =
+ new StringEntity(messagesJson, ContentType.create(contentType, "UTF-8"));
+ postRequest.setEntity(requestEntity);
+
+ try {
+ final Pair<Integer, String> responsePair = closeableHttpClient.execute(postRequest, responseHandler());
+ final Integer responseCode = responsePair.getLeft();
+ final String responseBody = responsePair.getRight();
+ // if messages were published successfully, return successful response
+ if (isSuccessfulResponseCode(responseCode)) {
+ LOG.debug("DMaaP MR Messages published successfully. DMaaP Response Code: {}. DMaaP Response " +
+ "Body: {}, Number of Messages published: {}",
+ responseCode, responseBody, messages.size());
+
+ } else {
+ LOG.warn("Unable to publish messages to DMaaP MR Topic. DMaaP Response Code: {}, DMaaP Response " +
+ "Body: {}. Messages will be queued in recovery queue", responseCode, responseBody);
+ addMessagesToRecoveryQueue(publisherQueue, messages);
+ }
+
+ return createPublisherResponse(responseCode, responseBody,
+ getPendingMessages(publisherQueue, publisherConfig));
+
+ } catch (IOException e) {
+ // If IO Error then we need to also put messages in recovery queue
+ addMessagesToRecoveryQueue(publisherQueue, messages);
+ final String errorMessage = format("IO Exception while publishing messages to DMaaP Topic. " +
+ "Messages will be queued in recovery queue. Messages Size: %d", messages.size());
+
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+
+ }
+
+
+ @Override
+ public DMaaPMRPublisherResponse flush() {
+ final List<String> queueMessages = publisherQueue.getMessageForPublishing();
+ // If there are no message return 204 (No Content) response code
+ if (queueMessages.isEmpty()) {
+ LOG.debug("No messages to publish to batch queue. Returning 204 status code");
+ return createPublisherNoContentResponse();
+ } else {
+ // force publish messages in queue
+ return forcePublish(queueMessages);
+ }
+ }
+
+ @Override
+ public Date getPublisherCreationTime() {
+ return new Date(publisherCreationTime.getTime());
+ }
+
+ @Override
+ public void close() throws Exception {
+
+ // flush current message in the queue
+ int retrialNumber = 0;
+ int flushResponseCode;
+
+ // automatic retries if messages cannot be flushed
+ do {
+ retrialNumber++;
+ DMaaPMRPublisherResponse flushResponse = flush();
+ flushResponseCode = flushResponse.getResponseCode();
+
+ if (!isSuccessfulResponseCode(flushResponseCode)) {
+ LOG.warn("Unable to flush batch messages to publisher due to DMaaP MR invalid Response: {}. " +
+ "Retrial No: {} of Max {} Retries", flushResponseCode, retrialNumber,
+ AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE);
+
+ Thread.sleep(AnalyticsConstants.PUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE);
+ }
+ } while (retrialNumber <= AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE &&
+ !isSuccessfulResponseCode(flushResponseCode));
+
+ if (!isSuccessfulResponseCode(flushResponseCode)) {
+ LOG.error("Unable to flush batch messages to publisher. Messages loss cannot be prevented");
+ } else {
+ LOG.info("Successfully published all batched messages to publisher.");
+ }
+
+ // close http client
+ closeableHttpClient.close();
+
+ }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java
index 3877f0b..e625fba 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java
@@ -1,87 +1,87 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import java.util.List;
-
-/**
- * <p>
- * DMaaP MR Publisher Queue handles back pressure in case DMaaP MR Publisher topic
- * is offline for some reason. It does so by having a recovery queue which keeps
- * messages in order in case there is temporary interruption in DMaaP Publisher
- * </p>
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public interface DMaaPMRPublisherQueue {
-
- /**
- * <p>
- * Add batchMessages to Batch Queue
- * </p>
- *
- * @param batchMessages messages that needs to be added to batch queue
- * @return current size of batch queue. Throws {@link IllegalStateException}
- * if batch queue does not have enough space
- */
- int addBatchMessages(List<String> batchMessages);
-
-
- /**
- * <p>
- * Add recoverable messages to Recoverable Queue
- * </p>
- *
- * @param recoverableMessages messages that needs to be added to recoverable queue
- * @return current size of the recoverable queue. Throws {@link IllegalStateException}
- * if recoverable queue does not have enough space
- */
- int addRecoverableMessages(List<String> recoverableMessages);
-
- /**
- * <p>
- * Get messages that need to be published to DMaaP topic. Messages in recoverable
- * queue are appended if present.
- * </p>
- *
- * @return List of messages from both batch and recovery queue
- */
- List<String> getMessageForPublishing();
-
- /**
- * <p>
- * Remaining capacity of Batch Queue
- * </p>
- *
- * @return Remaining Batch Queue Size
- */
- int getBatchQueueRemainingSize();
-
- /**
- * <p>
- * Remaining capacity of Recovery Queue
- * </p>
- *
- * @return Remaining Recovery Queue Size
- */
- int getRecoveryQueueRemainingSize();
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.publisher;
+
+import java.util.List;
+
+/**
+ * <p>
+ * DMaaP MR Publisher Queue handles back pressure in case DMaaP MR Publisher topic
+ * is offline for some reason. It does so by having a recovery queue which keeps
+ * messages in order in case there is temporary interruption in DMaaP Publisher
+ * </p>
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public interface DMaaPMRPublisherQueue {
+
+ /**
+ * <p>
+ * Add batchMessages to Batch Queue
+ * </p>
+ *
+ * @param batchMessages messages that needs to be added to batch queue
+ * @return current size of batch queue. Throws {@link IllegalStateException}
+ * if batch queue does not have enough space
+ */
+ int addBatchMessages(List<String> batchMessages);
+
+
+ /**
+ * <p>
+ * Add recoverable messages to Recoverable Queue
+ * </p>
+ *
+ * @param recoverableMessages messages that needs to be added to recoverable queue
+ * @return current size of the recoverable queue. Throws {@link IllegalStateException}
+ * if recoverable queue does not have enough space
+ */
+ int addRecoverableMessages(List<String> recoverableMessages);
+
+ /**
+ * <p>
+ * Get messages that need to be published to DMaaP topic. Messages in recoverable
+ * queue are appended if present.
+ * </p>
+ *
+ * @return List of messages from both batch and recovery queue
+ */
+ List<String> getMessageForPublishing();
+
+ /**
+ * <p>
+ * Remaining capacity of Batch Queue
+ * </p>
+ *
+ * @return Remaining Batch Queue Size
+ */
+ int getBatchQueueRemainingSize();
+
+ /**
+ * <p>
+ * Remaining capacity of Recovery Queue
+ * </p>
+ *
+ * @return Remaining Recovery Queue Size
+ */
+ int getRecoveryQueueRemainingSize();
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java
index 8d44f93..aeacbb0 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java
@@ -1,45 +1,45 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import com.google.inject.assistedinject.Assisted;
-
-/**
- * <p>
- * Factory to initialize instance of {@link DMaaPMRPublisherQueue} for Guice DI injection purposes.
- * <p>
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public interface DMaaPMRPublisherQueueFactory {
-
- /**
- * Guice Factory to create DMaaP MR Publisher Queue
- *
- * @param batchQueueSize batch queue size
- * @param recoveryQueueSize recovery queue size
- *
- * @return instance of DMaaP MR Publisher Queue
- */
- DMaaPMRPublisherQueue create(@Assisted("batchQueueSize") int batchQueueSize,
- @Assisted("recoveryQueueSize") int recoveryQueueSize);
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.publisher;
+
+import com.google.inject.assistedinject.Assisted;
+
+/**
+ * <p>
+ * Factory to initialize instance of {@link DMaaPMRPublisherQueue} for Guice DI injection purposes.
+ * <p>
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public interface DMaaPMRPublisherQueueFactory {
+
+ /**
+ * Guice Factory to create DMaaP MR Publisher Queue
+ *
+ * @param batchQueueSize batch queue size
+ * @param recoveryQueueSize recovery queue size
+ *
+ * @return instance of DMaaP MR Publisher Queue
+ */
+ DMaaPMRPublisherQueue create(@Assisted("batchQueueSize") int batchQueueSize,
+ @Assisted("recoveryQueueSize") int recoveryQueueSize);
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java
index e42b6b0..343c31a 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java
@@ -1,126 +1,126 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import com.google.inject.Inject;
-import com.google.inject.assistedinject.Assisted;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.LinkedBlockingDeque;
-
-import static com.google.common.collect.Iterables.concat;
-import static com.google.common.collect.Lists.newLinkedList;
-import static java.util.Collections.unmodifiableList;
-
-/**
- * <p>
- * An implementation of {@link DMaaPMRPublisherQueue} which uses {@link java.util.concurrent.BlockingDeque}
- * for batch and recovery queues
- * </p>
- *
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public class DMaaPMRPublisherQueueImpl implements DMaaPMRPublisherQueue {
-
- private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherQueueImpl.class);
-
- private final LinkedBlockingDeque<String> batchQueue;
- private final LinkedBlockingDeque<String> recoveryQueue;
-
- @Inject
- public DMaaPMRPublisherQueueImpl(@Assisted("batchQueueSize") int batchQueueSize,
- @Assisted("recoveryQueueSize") int recoveryQueueSize) {
- batchQueue = new LinkedBlockingDeque<>(batchQueueSize);
- recoveryQueue = new LinkedBlockingDeque<>(recoveryQueueSize);
- LOG.debug("Creating Instance of DMaaP Publisher Queue. BatchQueueSize: {}, RecoveryQueueSize: {}",
- batchQueueSize, recoveryQueueSize);
- }
-
- @Override
- public synchronized int addBatchMessages(List<String> batchMessages) {
-
- // checks if batchMessages size does not exceed batch queue capacity
- if (batchMessages.size() > batchQueue.remainingCapacity()) {
- throw new IllegalStateException("Not enough capacity to add batchMessages in batch queue");
- }
-
- // Add batchMessages to batch queue
- for (String message : batchMessages) {
- batchQueue.add(message);
- }
-
- // returns current elements size in batch queue
- return batchQueue.size();
- }
-
- @Override
- public synchronized int addRecoverableMessages(List<String> recoverableMessages) {
-
- // checks if messages size does not exceed recovery queue size
- if (recoverableMessages.size() > recoveryQueue.remainingCapacity()) {
- throw new IllegalStateException("Not enough capacity to add messages in recovery queue");
- }
-
- // add messages to recovery queue
- for (String recoverableMessage : recoverableMessages) {
- recoveryQueue.add(recoverableMessage);
- }
-
- // returns current size of recovery queue
- return recoveryQueue.size();
- }
-
- @Override
- public synchronized List<String> getMessageForPublishing() {
-
- final List<String> recoveryMessageList = new LinkedList<>();
- final List<String> batchMessagesList = new LinkedList<>();
-
- // get messages from recovery queue if present
- if (!recoveryQueue.isEmpty()) {
- final int recoveryQueueSize = recoveryQueue.drainTo(recoveryMessageList);
- LOG.debug("Drained Recovery Queue elements for flushing: {}", recoveryQueueSize);
- }
-
- // get messages from batch queue if present
- if (!batchQueue.isEmpty()) {
- final int batchQueueSize = batchQueue.drainTo(batchMessagesList);
- LOG.debug("Drained Batch Queue elements for flushing: {}", batchQueueSize);
- }
-
- // concat recovery and batch queue elements
- return unmodifiableList(newLinkedList(concat(recoveryMessageList, batchMessagesList)));
- }
-
- @Override
- public synchronized int getBatchQueueRemainingSize() {
- return batchQueue.remainingCapacity();
- }
-
- @Override
- public synchronized int getRecoveryQueueRemainingSize() {
- return recoveryQueue.remainingCapacity();
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.publisher;
+
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import static com.google.common.collect.Iterables.concat;
+import static com.google.common.collect.Lists.newLinkedList;
+import static java.util.Collections.unmodifiableList;
+
+/**
+ * <p>
+ * An implementation of {@link DMaaPMRPublisherQueue} which uses {@link java.util.concurrent.BlockingDeque}
+ * for batch and recovery queues
+ * </p>
+ *
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public class DMaaPMRPublisherQueueImpl implements DMaaPMRPublisherQueue {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherQueueImpl.class);
+
+ private final LinkedBlockingDeque<String> batchQueue;
+ private final LinkedBlockingDeque<String> recoveryQueue;
+
+ @Inject
+ public DMaaPMRPublisherQueueImpl(@Assisted("batchQueueSize") int batchQueueSize,
+ @Assisted("recoveryQueueSize") int recoveryQueueSize) {
+ batchQueue = new LinkedBlockingDeque<>(batchQueueSize);
+ recoveryQueue = new LinkedBlockingDeque<>(recoveryQueueSize);
+ LOG.debug("Creating Instance of DMaaP Publisher Queue. BatchQueueSize: {}, RecoveryQueueSize: {}",
+ batchQueueSize, recoveryQueueSize);
+ }
+
+ @Override
+ public synchronized int addBatchMessages(List<String> batchMessages) {
+
+ // checks if batchMessages size does not exceed batch queue capacity
+ if (batchMessages.size() > batchQueue.remainingCapacity()) {
+ throw new IllegalStateException("Not enough capacity to add batchMessages in batch queue");
+ }
+
+ // Add batchMessages to batch queue
+ for (String message : batchMessages) {
+ batchQueue.add(message);
+ }
+
+ // returns current elements size in batch queue
+ return batchQueue.size();
+ }
+
+ @Override
+ public synchronized int addRecoverableMessages(List<String> recoverableMessages) {
+
+ // checks if messages size does not exceed recovery queue size
+ if (recoverableMessages.size() > recoveryQueue.remainingCapacity()) {
+ throw new IllegalStateException("Not enough capacity to add messages in recovery queue");
+ }
+
+ // add messages to recovery queue
+ for (String recoverableMessage : recoverableMessages) {
+ recoveryQueue.add(recoverableMessage);
+ }
+
+ // returns current size of recovery queue
+ return recoveryQueue.size();
+ }
+
+ @Override
+ public synchronized List<String> getMessageForPublishing() {
+
+ final List<String> recoveryMessageList = new LinkedList<>();
+ final List<String> batchMessagesList = new LinkedList<>();
+
+ // get messages from recovery queue if present
+ if (!recoveryQueue.isEmpty()) {
+ final int recoveryQueueSize = recoveryQueue.drainTo(recoveryMessageList);
+ LOG.debug("Drained Recovery Queue elements for flushing: {}", recoveryQueueSize);
+ }
+
+ // get messages from batch queue if present
+ if (!batchQueue.isEmpty()) {
+ final int batchQueueSize = batchQueue.drainTo(batchMessagesList);
+ LOG.debug("Drained Batch Queue elements for flushing: {}", batchQueueSize);
+ }
+
+ // concat recovery and batch queue elements
+ return unmodifiableList(newLinkedList(concat(recoveryMessageList, batchMessagesList)));
+ }
+
+ @Override
+ public synchronized int getBatchQueueRemainingSize() {
+ return batchQueue.remainingCapacity();
+ }
+
+ @Override
+ public synchronized int getRecoveryQueueRemainingSize() {
+ return recoveryQueue.remainingCapacity();
+ }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java
index f193b2c..6a0aa63 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java
@@ -1,57 +1,57 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.subscriber;
-
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
-
-import java.util.Date;
-
-/**
- * <p>
- * DMaaP MR Subscriber can be used to subscribe messages from DMaaP MR Topics.
- * <p>
- *
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRSubscriber extends AutoCloseable {
-
- /**
- * Fetches Messages from DMaaP MR Topic. {@link DMaaPMRPublisherConfig} settings parameters
- * for messageLimit and message timeout are used
- *
- * @return DMaaP Message Router Subscriber Response
- */
- DMaaPMRSubscriberResponse fetchMessages();
-
-
- /**
- * Returns the Subscriber instance creation time
- * <p>
- * NOTE: Due to DMaaP API Design - Subscribers can only fetch messages which
- * are published to the topic after the creation of the Subscriber.
- *
- * @return creation time of Subscriber instance
- */
- Date getSubscriberCreationTime();
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.subscriber;
+
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
+
+import java.util.Date;
+
+/**
+ * <p>
+ * DMaaP MR Subscriber can be used to subscribe messages from DMaaP MR Topics.
+ * <p>
+ *
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRSubscriber extends AutoCloseable {
+
+ /**
+ * Fetches Messages from DMaaP MR Topic. {@link DMaaPMRPublisherConfig} settings parameters
+ * for messageLimit and message timeout are used
+ *
+ * @return DMaaP Message Router Subscriber Response
+ */
+ DMaaPMRSubscriberResponse fetchMessages();
+
+
+ /**
+ * Returns the Subscriber instance creation time
+ * <p>
+ * NOTE: Due to DMaaP API Design - Subscribers can only fetch messages which
+ * are published to the topic after the creation of the Subscriber.
+ *
+ * @return creation time of Subscriber instance
+ */
+ Date getSubscriberCreationTime();
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java
index b7f5ada..e06e2bd 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java
@@ -1,47 +1,47 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.subscriber;
-
-import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
-
-/**
- * Factory to initialize instance of {@link DMaaPMRSubscriber} for Guice DI injection purposes.
- * <p>
- * <strong>
- * NOTE: Client should not use this Factory to initialize {@link DMaaPMRSubscriber} unless they
- * are wiring dependencies using Guice. Client must use {@link DMaaPMRFactory} to initialize
- * guice injected Subscriber instances
- * </strong>
- * <p>
- * @author Rajiv Singla . Creation Date: 10/20/2016.
- */
-public interface DMaaPMRSubscriberFactory {
-
- /**
- * Guice Factory to create DMaaP MR Subscriber Instance
- *
- * @param subscriberConfig subscriber config
- *
- * @return DMaaP MR Subscriber instance
- */
- DMaaPMRSubscriber create(DMaaPMRSubscriberConfig subscriberConfig);
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.subscriber;
+
+import org.onap.dcae.apod.analytics.dmaap.DMaaPMRFactory;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+
+/**
+ * Factory to initialize instance of {@link DMaaPMRSubscriber} for Guice DI injection purposes.
+ * <p>
+ * <strong>
+ * NOTE: Client should not use this Factory to initialize {@link DMaaPMRSubscriber} unless they
+ * are wiring dependencies using Guice. Client must use {@link DMaaPMRFactory} to initialize
+ * guice injected Subscriber instances
+ * </strong>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/20/2016.
+ */
+public interface DMaaPMRSubscriberFactory {
+
+ /**
+ * Guice Factory to create DMaaP MR Subscriber Instance
+ *
+ * @param subscriberConfig subscriber config
+ *
+ * @return DMaaP MR Subscriber instance
+ */
+ DMaaPMRSubscriber create(DMaaPMRSubscriberConfig subscriberConfig);
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java
index 678ff76..9d75ffc 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java
@@ -1,129 +1,129 @@
-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.dmaap.service.subscriber;
-
-import com.google.common.base.Optional;
-import com.google.inject.Inject;
-import com.google.inject.assistedinject.Assisted;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.http.HttpHeaders;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.Date;
-import java.util.LinkedList;
-import java.util.List;
-
-import static org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
-import static java.lang.String.format;
-
-/**
- * Concrete Implementation of {@link DMaaPMRSubscriber} which uses {@link HttpClient}
- *
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public class DMaaPMRSubscriberImpl extends BaseDMaaPMRComponent implements DMaaPMRSubscriber {
-
- private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRSubscriberImpl.class);
-
- private final DMaaPMRSubscriberConfig subscriberConfig;
- private final CloseableHttpClient closeableHttpClient;
- private final URI subscriberUri;
- private final Date subscriberCreationTime;
-
- @Inject
- public DMaaPMRSubscriberImpl(@Assisted DMaaPMRSubscriberConfig subscriberConfig,
- CloseableHttpClient closeableHttpClient) {
- this.subscriberConfig = subscriberConfig;
- this.closeableHttpClient = closeableHttpClient;
- this.subscriberUri = createSubscriberURI(subscriberConfig);
- this.subscriberCreationTime = new Date();
- }
-
- @Override
- public DMaaPMRSubscriberResponse fetchMessages() {
-
- final String userName = subscriberConfig.getUserName();
- final String userPassword = subscriberConfig.getUserPassword();
-
- final HttpGet getRequest = new HttpGet(subscriberUri);
-
- // add Authorization Header if username and password are present
- final Optional<String> authHeader = getAuthHeader(userName, userPassword);
- if (authHeader.isPresent()) {
- getRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
- } else {
- LOG.debug("DMaaP MR Subscriber Authentication is disabled as username or password is not present.");
- }
-
- try {
-
- final Pair<Integer, String> responsePair = closeableHttpClient.execute(getRequest, responseHandler());
- final Integer responseCode = responsePair.getLeft();
- final String responseBody = responsePair.getRight();
-
- List<String> fetchedMessages = new LinkedList<>();
- String responseMessage = responseBody;
-
- // if messages were published successfully, return successful response
- if (isSuccessfulResponseCode(responseCode)) {
- if (responseBody != null) {
- fetchedMessages = convertJsonToStringMessages(responseBody);
- responseMessage = "Messages Fetched Successfully";
- } else {
- responseMessage = "DMaaP Response Body had no messages";
- }
- } else {
- LOG.error("Unable to fetch messages to DMaaP MR Topic. DMaaP MR unsuccessful Response Code: {}, " +
- "DMaaP Response Body: {}", responseCode, responseBody);
- }
-
- return createSubscriberResponse(responseCode, responseMessage, fetchedMessages);
-
- } catch (IOException e) {
-
- final String errorMessage =
- format("IO Exception while fetching messages from DMaaP Topic. Exception %s", e);
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
- }
-
-
- }
-
- @Override
- public Date getSubscriberCreationTime() {
- return new Date(subscriberCreationTime.getTime());
- }
-
- @Override
- public void close() throws Exception {
- closeableHttpClient.close();
- }
-}
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.dmaap.service.subscriber;
+
+import com.google.common.base.Optional;
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHeaders;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
+import org.onap.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Date;
+import java.util.LinkedList;
+import java.util.List;
+
+import static org.onap.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
+import static java.lang.String.format;
+
+/**
+ * Concrete Implementation of {@link DMaaPMRSubscriber} which uses {@link HttpClient}
+ *
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public class DMaaPMRSubscriberImpl extends BaseDMaaPMRComponent implements DMaaPMRSubscriber {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRSubscriberImpl.class);
+
+ private final DMaaPMRSubscriberConfig subscriberConfig;
+ private final CloseableHttpClient closeableHttpClient;
+ private final URI subscriberUri;
+ private final Date subscriberCreationTime;
+
+ @Inject
+ public DMaaPMRSubscriberImpl(@Assisted DMaaPMRSubscriberConfig subscriberConfig,
+ CloseableHttpClient closeableHttpClient) {
+ this.subscriberConfig = subscriberConfig;
+ this.closeableHttpClient = closeableHttpClient;
+ this.subscriberUri = createSubscriberURI(subscriberConfig);
+ this.subscriberCreationTime = new Date();
+ }
+
+ @Override
+ public DMaaPMRSubscriberResponse fetchMessages() {
+
+ final String userName = subscriberConfig.getUserName();
+ final String userPassword = subscriberConfig.getUserPassword();
+
+ final HttpGet getRequest = new HttpGet(subscriberUri);
+
+ // add Authorization Header if username and password are present
+ final Optional<String> authHeader = getAuthHeader(userName, userPassword);
+ if (authHeader.isPresent()) {
+ getRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
+ } else {
+ LOG.debug("DMaaP MR Subscriber Authentication is disabled as username or password is not present.");
+ }
+
+ try {
+
+ final Pair<Integer, String> responsePair = closeableHttpClient.execute(getRequest, responseHandler());
+ final Integer responseCode = responsePair.getLeft();
+ final String responseBody = responsePair.getRight();
+
+ List<String> fetchedMessages = new LinkedList<>();
+ String responseMessage = responseBody;
+
+ // if messages were published successfully, return successful response
+ if (isSuccessfulResponseCode(responseCode)) {
+ if (responseBody != null) {
+ fetchedMessages = convertJsonToStringMessages(responseBody);
+ responseMessage = "Messages Fetched Successfully";
+ } else {
+ responseMessage = "DMaaP Response Body had no messages";
+ }
+ } else {
+ LOG.error("Unable to fetch messages to DMaaP MR Topic. DMaaP MR unsuccessful Response Code: {}, " +
+ "DMaaP Response Body: {}", responseCode, responseBody);
+ }
+
+ return createSubscriberResponse(responseCode, responseMessage, fetchedMessages);
+
+ } catch (IOException e) {
+
+ final String errorMessage =
+ format("IO Exception while fetching messages from DMaaP Topic. Exception %s", e);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+
+
+ }
+
+ @Override
+ public Date getSubscriberCreationTime() {
+ return new Date(subscriberCreationTime.getTime());
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeableHttpClient.close();
+ }
+}