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

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