2 * ===============================LICENSE_START======================================
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
11 * http://www.apache.org/licenses/LICENSE-2.0
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===========================================
21 package org.openecomp.dcae.apod.analytics.dmaap.service.publisher;
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;
43 import java.io.IOException;
45 import java.util.Date;
46 import java.util.List;
48 import static org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
49 import static java.lang.String.format;
52 * Concrete Implementation of {@link DMaaPMRPublisher} which uses {@link HttpClient}
54 * @author Rajiv Singla . Creation Date: 10/13/2016.
56 public class DMaaPMRPublisherImpl extends BaseDMaaPMRComponent implements DMaaPMRPublisher {
58 private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherImpl.class);
60 private final DMaaPMRPublisherConfig publisherConfig;
61 private final CloseableHttpClient closeableHttpClient;
62 private final DMaaPMRPublisherQueue publisherQueue;
63 private final Date publisherCreationTime;
64 private URI publisherUri;
67 public DMaaPMRPublisherImpl(@Assisted DMaaPMRPublisherConfig publisherConfig,
68 DMaaPMRPublisherQueueFactory dMaaPMRPublisherQueueFactory,
69 CloseableHttpClient closeableHttpClient) {
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();
81 public DMaaPMRPublisherResponse publish(List<String> messages) {
83 final int batchQueueRemainingSize = publisherQueue.getBatchQueueRemainingSize();
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);
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 " +
98 return forcePublish(Lists.newLinkedList(Iterables.concat(queueMessages, messages)));
104 public DMaaPMRPublisherResponse forcePublish(List<String> messages) {
106 LOG.debug("Force publishing messages to DMaaP MR Topic. Messages Size: {}", messages.size());
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);
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());
118 LOG.debug("DMaaP MR Publisher Authentication is disabled as username or password is not present.");
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);
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());
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);
143 return createPublisherResponse(responseCode, responseBody,
144 getPendingMessages(publisherQueue, publisherConfig));
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());
152 throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
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();
166 // force publish messages in queue
167 return forcePublish(queueMessages);
172 public Date getPublisherCreationTime() {
173 return new Date(publisherCreationTime.getTime());
177 public void close() throws Exception {
179 // flush current message in the queue
180 int retrialNumber = 0;
181 int flushResponseCode;
183 // automatic retries if messages cannot be flushed
186 DMaaPMRPublisherResponse flushResponse = flush();
187 flushResponseCode = flushResponse.getResponseCode();
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);
194 Thread.sleep(AnalyticsConstants.PUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE);
196 } while (retrialNumber <= AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE &&
197 !isSuccessfulResponseCode(flushResponseCode));
199 if (!isSuccessfulResponseCode(flushResponseCode)) {
200 LOG.error("Unable to flush batch messages to publisher. Messages loss cannot be prevented");
202 LOG.info("Successfully published all batched messages to publisher.");
206 closeableHttpClient.close();