diff options
Diffstat (limited to 'appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/http/HttpDmaapProducerImpl.java')
-rw-r--r-- | appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/http/HttpDmaapProducerImpl.java | 79 |
1 files changed, 38 insertions, 41 deletions
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 74c0c26b2..fe22ea10b 100644 --- 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 - } } |