Fixes in "appc-dmaap-adapter-bundle"

[appc.git] / appc-adapters / appc-dmaap-adapter / appc-dmaap-adapter-bundle / src / main / java / org / onap / appc / adapter / messaging / dmaap / http / HttpDmaapProducerImpl.java

diff --git a/appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/http/HttpDmaapProducerImpl.java b/appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/http/HttpDmaapProducerImpl.java
index 74c0c26..fe22ea1 100644 (file)
--- a/appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/http/HttpDmaapProducerImpl.java
+++ b/appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/http/HttpDmaapProducerImpl.java
@@ -24,16 +24,14 @@
 
 package org.onap.appc.adapter.messaging.dmaap.http;
 
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
-
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.entity.StringEntity;
@@ -41,7 +39,7 @@ import org.onap.appc.adapter.message.Producer;
 
 public class HttpDmaapProducerImpl extends CommonHttpClient implements Producer {
 
-    private final EELFLogger LOG = EELFManager.getInstance().getLogger(HttpDmaapProducerImpl.class);
+    private static final EELFLogger LOG = EELFManager.getInstance().getLogger(HttpDmaapProducerImpl.class);
 
     private static final String CONTENT_TYPE = "application/cambria";
     private static final String URL_TEMPLATE = "%s/events/%s";
@@ -49,19 +47,20 @@ public class HttpDmaapProducerImpl extends CommonHttpClient implements Producer
     private List<String> hosts;
     private Set<String> topics;
 
-    private boolean useHttps = false;
+    public HttpDmaapProducerImpl() {
+        //for test purposes
+    }
 
     public HttpDmaapProducerImpl(Collection<String> urls, String topicName) {
-        hosts = new ArrayList<>();
         topics = new HashSet<>();
         topics.add(topicName);
 
+        hosts = new ArrayList<>();
         for (String host : urls) {
             hosts.add(formatHostString(host));
         }
     }
 
-
     @Override
     public void updateCredentials(String user, String pass) {
         LOG.debug(String.format("Setting auth to %s for %s", user, this.toString()));
@@ -70,32 +69,16 @@ public class HttpDmaapProducerImpl extends CommonHttpClient implements Producer
 
     @Override
     public boolean post(String partition, String data) {
-        int sent = 0;
+        long sent = 0;
         try {
             HttpPost request = postReq(null);
             request.setHeader("Content-Type", CONTENT_TYPE);
             request.setEntity(new StringEntity(bodyLine(partition, data)));
 
-            for (String topic : topics) {
-                String uriStr = String.format(URL_TEMPLATE, hosts.get(0), topic);
-                try {
-                    request.setURI(new URI(uriStr));
-                    CloseableHttpResponse response = getClient().execute(request);
-                    if (response.getStatusLine().getStatusCode() == 200) {
-                        sent++;
-                    }
-                    response.close();
-                } catch (Exception sendEx) {
-                    LOG.error(String.format("Failed to send message to %s. Reason: %s", uriStr, sendEx.getMessage()),
-                        sendEx);
-                    if (hosts.size() > 1) {
-                        String failedUrl = hosts.remove(0);
-                        hosts.add(failedUrl);
-                        LOG.debug(String.format("Moving host %s to the end of the pool. New primary host is %s",
-                            failedUrl, hosts.get(0)));
-                    }
-                }
-            }
+            sent = topics.stream()
+                .filter(topic -> sendRequest(request, topic))
+                .count();
+
         } catch (Exception buildEx) {
             LOG.error(
                 String.format("Failed to build request with string [%s]. Message not sent to any topic. Reason: %s",
@@ -105,26 +88,40 @@ public class HttpDmaapProducerImpl extends CommonHttpClient implements Producer
         return sent == topics.size();
     }
 
-    @Override
-    public void useHttps(boolean yes) {
-        useHttps = yes;
+    private boolean sendRequest(HttpPost request, String topic) {
+        boolean successful = false;
+        String uriStr = String.format(URL_TEMPLATE, hosts.get(0), topic);
+        try {
+            request.setURI(new URI(uriStr));
+            CloseableHttpResponse response = getClient().execute(request);
+            if (response.getStatusLine().getStatusCode() == 200) {
+                successful = true;
+            }
+            response.close();
+        } catch (Exception sendEx) {
+            LOG.error(String.format("Failed to send message to %s. Reason: %s", uriStr, sendEx.getMessage()),
+                sendEx);
+            if (hosts.size() > 1) {
+                String failedUrl = hosts.remove(0);
+                hosts.add(failedUrl);
+                LOG.debug(String.format("Moving host %s to the end of the pool. New primary host is %s",
+                    failedUrl, hosts.get(0)));
+            }
+        }
+        return successful;
     }
 
     /**
      * Format the body for the application/cambria content type with no partitioning. See
      *
-     * @param msg
+     * @param message
      *            The message body to format
      * @return A string in the application/cambria content type
      */
-    private String bodyLine(String partition, String msg) {
-        String p = (partition == null) ? "" : partition;
-        String m = (msg == null) ? "" : msg;
-        return String.format("%d.%d.%s%s", p.length(), m.length(), p, m);
+    private String bodyLine(String partition, String message) {
+        String prt = (partition == null) ? "" : partition;
+        String msg = (message == null) ? "" : message;
+        return String.format("%d.%d.%s%s", prt.length(), msg.length(), prt, msg);
     }
 
-    @Override
-    public void close() {
-        // Nothing to do        
-    }
 }