dcae-analytics/dcae-analytics-web/src/main/java/org/onap/dcae/analytics/web/config/DmaapRetryConfig.java

   1 /*
   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
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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=========================================================
  17  *
  18  */
  19
  20 package org.onap.dcae.analytics.web.config;
  21
  22 import java.util.LinkedHashMap;
  23 import java.util.Map;
  24
  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;
  49
  50 /**
  51  * @author Rajiv Singla
  52  */
  53 @Configuration
  54 @Profile(AnalyticsProfile.DMAAP_PROFILE_NAME)
  55 @Import(MessageStoreConfig.class)
  56 public class DmaapRetryConfig {
  57
  58     @Bean
  59     public QueueChannel errorChannel() {
  60         return MessageChannels.queue().get();
  61     }
  62
  63     @Bean
  64     public IntegrationFlow loggingFlow() {
  65         return IntegrationFlows.from(errorChannel())
  66                 .log(LoggingHandler.Level.ERROR)
  67                 .get();
  68     }
  69
  70     @Bean
  71     public ErrorMessageSendingRecoverer errorMessageSendingRecoverer(final PollableChannel recoveryChannel) {
  72         final ErrorMessageSendingRecoverer errorMessageSendingRecoverer = new ErrorMessageSendingRecoverer();
  73         errorMessageSendingRecoverer.setChannel(recoveryChannel);
  74         return errorMessageSendingRecoverer;
  75     }
  76
  77     @Bean
  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();
  81     }
  82
  83     @Bean
  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;
  91     }
  92
  93     @Bean
  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);
  99         return retryTemplate;
 100     }
 101
 102     @Bean
 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);
 109     }
 110
 111     @Bean
 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;
 118     }
 119
 120 }