dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java

   1 /*
   2  * ===============================LICENSE_START======================================
   3  *  dcae-analytics
   4  * ================================================================================
   5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *   You may obtain a copy of the License at
  10  *
  11  *          http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  *  Unless required by applicable law or agreed to in writing, software
  14  *  distributed under the License is distributed on an "AS IS" BASIS,
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  *  See the License for the specific language governing permissions and
  17  *  limitations under the License.
  18  *  ============================LICENSE_END===========================================
  19  */
  20
  21 package org.openecomp.dcae.apod.analytics.dmaap.service.publisher;
  22
  23 import com.google.common.base.Optional;
  24 import com.google.common.collect.Iterables;
  25 import com.google.common.collect.Lists;
  26 import com.google.inject.Inject;
  27 import com.google.inject.assistedinject.Assisted;
  28 import org.apache.commons.lang3.tuple.Pair;
  29 import org.apache.http.HttpHeaders;
  30 import org.apache.http.client.HttpClient;
  31 import org.apache.http.client.methods.HttpPost;
  32 import org.apache.http.entity.ContentType;
  33 import org.apache.http.entity.StringEntity;
  34 import org.apache.http.impl.client.CloseableHttpClient;
  35 import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
  36 import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
  37 import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
  38 import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
  39 import org.openecomp.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
  40 import org.slf4j.Logger;
  41 import org.slf4j.LoggerFactory;
  42
  43 import java.io.IOException;
  44 import java.net.URI;
  45 import java.util.Date;
  46 import java.util.List;
  47
  48 import static org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
  49 import static java.lang.String.format;
  50
  51 /**
  52  * Concrete Implementation of {@link DMaaPMRPublisher} which uses {@link HttpClient}
  53  *
  54  * @author Rajiv Singla . Creation Date: 10/13/2016.
  55  */
  56 public class DMaaPMRPublisherImpl extends BaseDMaaPMRComponent implements DMaaPMRPublisher {
  57
  58     private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherImpl.class);
  59
  60     private final DMaaPMRPublisherConfig publisherConfig;
  61     private final CloseableHttpClient closeableHttpClient;
  62     private final DMaaPMRPublisherQueue publisherQueue;
  63     private final Date publisherCreationTime;
  64     private URI publisherUri;
  65
  66     @Inject
  67     public DMaaPMRPublisherImpl(@Assisted DMaaPMRPublisherConfig publisherConfig,
  68                                 DMaaPMRPublisherQueueFactory dMaaPMRPublisherQueueFactory,
  69                                 CloseableHttpClient closeableHttpClient) {
  70
  71         this.publisherConfig = publisherConfig;
  72         this.publisherQueue = dMaaPMRPublisherQueueFactory.create(
  73                 publisherConfig.getMaxBatchSize(), publisherConfig.getMaxRecoveryQueueSize());
  74         this.closeableHttpClient = closeableHttpClient;
  75         this.publisherUri = createPublisherURI(publisherConfig);
  76         this.publisherCreationTime = new Date();
  77     }
  78
  79
  80     @Override
  81     public DMaaPMRPublisherResponse publish(List<String> messages)  {
  82
  83         final int batchQueueRemainingSize = publisherQueue.getBatchQueueRemainingSize();
  84
  85         // if messages size is less than batch queue size - just queue them for batch publishing
  86         if (batchQueueRemainingSize > messages.size()) {
  87             LOG.debug("Adding messages to batch Queue. No flushing required. Messages Size:{}. Batch Queue Size:{}",
  88                     messages.size(), batchQueueRemainingSize);
  89             final int batchQueueSize = publisherQueue.addBatchMessages(messages);
  90             return createPublisherAcceptedResponse(batchQueueSize);
  91
  92         } else {
  93
  94             // grab all already queued messages, append current messages and force publish them to DMaaP MR topic
  95             final List<String> queueMessages = publisherQueue.getMessageForPublishing();
  96             LOG.debug("Batch Queue capacity exceeds messages size. Flushing of all pending messages to DMaaP MR " +
  97                     "Publisher Topic.");
  98             return forcePublish(Lists.newLinkedList(Iterables.concat(queueMessages, messages)));
  99         }
 100
 101     }
 102
 103     @Override
 104     public DMaaPMRPublisherResponse forcePublish(List<String> messages) {
 105
 106         LOG.debug("Force publishing messages to DMaaP MR Topic. Messages Size: {}", messages.size());
 107
 108         final String contentType = publisherConfig.getContentType();
 109         final String userName = publisherConfig.getUserName();
 110         final String userPassword = publisherConfig.getUserPassword();
 111         final HttpPost postRequest = new HttpPost(publisherUri);
 112
 113         // add Authorization Header if username and password are present
 114         final Optional<String> authHeader = getAuthHeader(userName, userPassword);
 115         if (authHeader.isPresent()) {
 116             postRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
 117         } else {
 118             LOG.debug("DMaaP MR Publisher Authentication is disabled as username or password is not present.");
 119         }
 120
 121         // Create post string entity
 122         final String messagesJson = convertToJsonString(messages);
 123         final StringEntity requestEntity =
 124                 new StringEntity(messagesJson, ContentType.create(contentType, "UTF-8"));
 125         postRequest.setEntity(requestEntity);
 126
 127         try {
 128             final Pair<Integer, String> responsePair = closeableHttpClient.execute(postRequest, responseHandler());
 129             final Integer responseCode = responsePair.getLeft();
 130             final String responseBody = responsePair.getRight();
 131             // if messages were published successfully, return successful response
 132             if (isSuccessfulResponseCode(responseCode)) {
 133                 LOG.debug("DMaaP MR Messages published successfully. DMaaP Response Code: {}. DMaaP Response " +
 134                                 "Body: {}, Number of Messages published: {}",
 135                         responseCode, responseBody, messages.size());
 136
 137             } else {
 138                 LOG.warn("Unable to publish messages to DMaaP MR Topic. DMaaP Response Code: {}, DMaaP Response " +
 139                         "Body: {}. Messages will be queued in recovery queue", responseCode, responseBody);
 140                 addMessagesToRecoveryQueue(publisherQueue, messages);
 141             }
 142
 143             return createPublisherResponse(responseCode, responseBody,
 144                     getPendingMessages(publisherQueue, publisherConfig));
 145
 146         } catch (IOException e) {
 147             // If IO Error then we need to also put messages in recovery queue
 148             addMessagesToRecoveryQueue(publisherQueue, messages);
 149             final String errorMessage = format("IO Exception while publishing messages to DMaaP Topic. " +
 150                     "Messages will be queued in recovery queue. Messages Size: %d", messages.size());
 151
 152             throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
 153         }
 154
 155     }
 156
 157
 158     @Override
 159     public DMaaPMRPublisherResponse flush() {
 160         final List<String> queueMessages = publisherQueue.getMessageForPublishing();
 161         // If there are no message return 204 (No Content) response code
 162         if (queueMessages.isEmpty()) {
 163             LOG.debug("No messages to publish to batch queue. Returning 204 status code");
 164             return createPublisherNoContentResponse();
 165         } else {
 166             // force publish messages in queue
 167             return forcePublish(queueMessages);
 168         }
 169     }
 170
 171     @Override
 172     public Date getPublisherCreationTime() {
 173         return new Date(publisherCreationTime.getTime());
 174     }
 175
 176     @Override
 177     public void close() throws Exception {
 178
 179         // flush current message in the queue
 180         int retrialNumber = 0;
 181         int flushResponseCode;
 182
 183         // automatic retries if messages cannot be flushed
 184         do {
 185             retrialNumber++;
 186             DMaaPMRPublisherResponse flushResponse = flush();
 187             flushResponseCode = flushResponse.getResponseCode();
 188
 189             if (!isSuccessfulResponseCode(flushResponseCode)) {
 190                 LOG.warn("Unable to flush batch messages to publisher due to DMaaP MR invalid Response: {}. " +
 191                                 "Retrial No: {} of Max {} Retries", flushResponseCode, retrialNumber,
 192                         AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE);
 193
 194                 Thread.sleep(AnalyticsConstants.PUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE);
 195             }
 196         } while (retrialNumber <= AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE &&
 197                 !isSuccessfulResponseCode(flushResponseCode));
 198
 199         if (!isSuccessfulResponseCode(flushResponseCode)) {
 200             LOG.error("Unable to flush batch messages to publisher. Messages loss cannot be prevented");
 201         } else {
 202             LOG.info("Successfully published all batched messages to publisher.");
 203         }
 204
 205         // close http client
 206         closeableHttpClient.close();
 207
 208     }
 209 }