2 * ================================================================================
3 * Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
4 * ================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 * ============LICENSE_END=========================================================
20 package org.onap.dcae.analytics.web.config;
22 import java.util.LinkedHashMap;
25 import org.onap.dcae.analytics.model.AnalyticsProfile;
26 import org.onap.dcae.analytics.model.DmaapMrConstants;
27 import org.springframework.context.annotation.Bean;
28 import org.springframework.context.annotation.Configuration;
29 import org.springframework.context.annotation.Import;
30 import org.springframework.context.annotation.Profile;
31 import org.springframework.integration.channel.QueueChannel;
32 import org.springframework.integration.dsl.IntegrationFlow;
33 import org.springframework.integration.dsl.IntegrationFlows;
34 import org.springframework.integration.dsl.channel.MessageChannels;
35 import org.springframework.integration.handler.LoggingHandler;
36 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
37 import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
38 import org.springframework.integration.store.BasicMessageGroupStore;
39 import org.springframework.integration.store.MessageGroupQueue;
40 import org.springframework.messaging.MessageHandlingException;
41 import org.springframework.messaging.PollableChannel;
42 import org.springframework.retry.RetryPolicy;
43 import org.springframework.retry.backoff.BackOffPolicy;
44 import org.springframework.retry.backoff.ExponentialBackOffPolicy;
45 import org.springframework.retry.policy.SimpleRetryPolicy;
46 import org.springframework.retry.support.RetryTemplate;
47 import org.springframework.web.client.HttpStatusCodeException;
48 import org.springframework.web.client.RestClientException;
51 * @author Rajiv Singla
54 @Profile(AnalyticsProfile.DMAAP_PROFILE_NAME)
55 @Import(MessageStoreConfig.class)
56 public class DmaapRetryConfig {
59 public QueueChannel errorChannel() {
60 return MessageChannels.queue().get();
64 public IntegrationFlow loggingFlow() {
65 return IntegrationFlows.from(errorChannel())
66 .log(LoggingHandler.Level.ERROR)
71 public ErrorMessageSendingRecoverer errorMessageSendingRecoverer(final PollableChannel recoveryChannel) {
72 final ErrorMessageSendingRecoverer errorMessageSendingRecoverer = new ErrorMessageSendingRecoverer();
73 errorMessageSendingRecoverer.setChannel(recoveryChannel);
74 return errorMessageSendingRecoverer;
78 public PollableChannel recoveryChannel(final BasicMessageGroupStore basicMessageGroupStore) {
79 return MessageChannels.queue(new MessageGroupQueue(basicMessageGroupStore,
80 DmaapMrConstants.DMAAP_MR_PUBLISHER_RECOVERY_MESSAGE_STORE_GROUP_ID)).get();
84 public RequestHandlerRetryAdvice requestHandlerRetryAdvice(final RetryTemplate retryTemplate,
85 final ErrorMessageSendingRecoverer
86 errorMessageSendingRecoverer) {
87 final RequestHandlerRetryAdvice requestHandlerRetryAdvice = new RequestHandlerRetryAdvice();
88 requestHandlerRetryAdvice.setRetryTemplate(retryTemplate);
89 requestHandlerRetryAdvice.setRecoveryCallback(errorMessageSendingRecoverer);
90 return requestHandlerRetryAdvice;
94 public RetryTemplate retryTemplate(final RetryPolicy retryPolicy,
95 final BackOffPolicy backOffPolicy) {
96 final RetryTemplate retryTemplate = new RetryTemplate();
97 retryTemplate.setRetryPolicy(retryPolicy);
98 retryTemplate.setBackOffPolicy(backOffPolicy);
103 public RetryPolicy retryPolicy() {
104 final Map<Class<? extends Throwable>, Boolean> retryableExceptions = new LinkedHashMap<>();
105 retryableExceptions.put(MessageHandlingException.class, true);
106 retryableExceptions.put(HttpStatusCodeException.class, true);
107 retryableExceptions.put(RestClientException.class, true);
108 return new SimpleRetryPolicy(DmaapMrConstants.DEFAULT_NUM_OF_RETRIES_ON_FAILURE, retryableExceptions);
112 public BackOffPolicy backOffPolicy() {
113 final ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
114 backOffPolicy.setInitialInterval(DmaapMrConstants.DEFAULT_RETRY_INITIAL_INTERVAL);
115 backOffPolicy.setMultiplier(DmaapMrConstants.DEFAULT_RETRY_MULTIPLIER);
116 backOffPolicy.setMaxInterval(DmaapMrConstants.DEFAULT_RETRY_MAX_INTERVAL);
117 return backOffPolicy;