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

   1 /*
   2  * ================================================================================
   3  * Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
   4  * Copyright (c) 2021 China Mobile Property. All rights reserved.
   5  * Copyright (c) 2021 Nokia 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
  22 package org.onap.dcae.analytics.web.config;
  23
  24 import org.onap.dcae.analytics.model.AnalyticsHttpConstants;
  25 import org.onap.dcae.analytics.model.AnalyticsProfile;
  26 import org.onap.dcae.analytics.model.DmaapMrConstants;
  27 import org.onap.dcae.analytics.web.dmaap.MrMessageSplitter;
  28 import org.onap.dcae.analytics.web.dmaap.MrPublisherPreferences;
  29 import org.onap.dcae.analytics.web.dmaap.MrSubscriberPollingAdvice;
  30 import org.onap.dcae.analytics.web.dmaap.MrSubscriberPreferences;
  31 import org.onap.dcae.analytics.web.dmaap.MrTriggerMessageProvider;
  32 import org.onap.dcae.analytics.web.http.HttpClientPreferencesCustomizer;
  33 import org.onap.dcae.analytics.web.util.AnalyticsWebUtils;
  34 import org.springframework.boot.web.client.RestTemplateBuilder;
  35 import org.springframework.context.annotation.Bean;
  36 import org.springframework.context.annotation.Configuration;
  37 import org.springframework.context.annotation.Import;
  38 import org.springframework.context.annotation.Profile;
  39 import org.springframework.http.HttpMethod;
  40 import org.springframework.integration.channel.DirectChannel;
  41 import org.springframework.integration.channel.QueueChannel;
  42 import org.springframework.integration.core.MessageSource;
  43 import org.springframework.integration.dsl.IntegrationFlow;
  44 import org.springframework.integration.dsl.IntegrationFlows;
  45 import org.springframework.integration.dsl.MessageChannels;
  46 import org.springframework.integration.endpoint.MethodInvokingMessageSource;
  47 import org.springframework.integration.handler.GenericHandler;
  48 import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
  49 import org.springframework.integration.http.dsl.Http;
  50 import org.springframework.integration.scheduling.PollerMetadata;
  51 import org.springframework.integration.store.BasicMessageGroupStore;
  52 import org.springframework.integration.store.MessageGroupQueue;
  53 import org.springframework.integration.support.MessageBuilder;
  54 import org.springframework.messaging.MessageHeaders;
  55 import org.springframework.web.client.RestTemplate;
  56
  57 import com.fasterxml.jackson.databind.ObjectMapper;
  58
  59 /**
  60  * @author Rajiv Singla
  61  */
  62 @Configuration
  63 @Import(value = {DmaapPollerConfig.class, DmaapRetryConfig.class})
  64 @Profile(AnalyticsProfile.DMAAP_PROFILE_NAME)
  65 public class DmaapMrConfig {
  66
  67     private static final String[] DMAAP_MAPPED_REQUEST_HEADERS =
  68             DmaapMrConstants.getDmaapmappedHeaders().toArray(new String[DmaapMrConstants.getDmaapmappedHeaders().size()]);
  69
  70     @Bean(name = DmaapMrConstants.DMAAP_MR_SUBSCRIBER_OUTPUT_CHANNEL_NAME)
  71     public QueueChannel mrSubscriberOutputChannel(final BasicMessageGroupStore basicMessageGroupStore) {
  72         return MessageChannels.queue(new MessageGroupQueue(basicMessageGroupStore,
  73                 DmaapMrConstants.DMAAP_MR_SUBSCRIBER_OUTPUT_MESSAGE_STORE_GROUP_ID)).get();
  74     }
  75
  76     @Bean(name = DmaapMrConstants.DMAAP_MR_PUBLISHER_INPUT_CHANNEL)
  77     public DirectChannel mrPublisherInputChannel() {
  78         return MessageChannels.direct().get();
  79     }
  80
  81
  82     @Bean
  83     public RestTemplate mrSubscriberRestTemplate(final MrSubscriberPreferences mrSubscriberPreferences,
  84                                                  final RestTemplateBuilder restTemplateBuilder) {
  85         return restTemplateBuilder
  86                 .additionalCustomizers(new HttpClientPreferencesCustomizer<>(mrSubscriberPreferences))
  87                 .build();
  88     }
  89
  90     @Bean
  91     public RestTemplate mrPublisherRestTemplate(final MrPublisherPreferences mrPublisherPreferences,
  92                                                 final RestTemplateBuilder restTemplateBuilder) {
  93         return restTemplateBuilder
  94                 .additionalCustomizers(new HttpClientPreferencesCustomizer<>(mrPublisherPreferences))
  95                 .build();
  96     }
  97
  98     @Bean
  99     public MrMessageSplitter mrMessageSplitter(final ObjectMapper objectMapper,
 100                                                final Integer processingBatchSize) {
 101         final Integer batchSize = processingBatchSize != null ? processingBatchSize :
 102                 DmaapMrConstants.SUBSCRIBER_DEFAULT_PROCESSING_BATCH_SIZE;
 103         return new MrMessageSplitter(objectMapper, batchSize);
 104     }
 105
 106
 107     @Bean
 108     public MrTriggerMessageProvider mrTriggerMessageProvider(
 109             final MrSubscriberPreferences mrSubscriberPreferences) {
 110         return new MrTriggerMessageProvider(mrSubscriberPreferences);
 111     }
 112
 113     @Bean
 114     public MessageSource mrMessageSource(final MrTriggerMessageProvider mrTriggerMessageProvider) {
 115         final MethodInvokingMessageSource source = new MethodInvokingMessageSource();
 116         source.setObject(mrTriggerMessageProvider);
 117         source.setMethodName(MrTriggerMessageProvider.TRIGGER_METHOD_NAME);
 118         return source;
 119     }
 120
 121     @Bean
 122     public IntegrationFlow mrSubscriberFlow(final PollerMetadata pollerMetadata,
 123                                             final RestTemplate mrSubscriberRestTemplate,
 124                                             final MessageSource mrMessageSource,
 125                                             final QueueChannel mrSubscriberOutputChannel,
 126                                             final MrMessageSplitter mrMessageSplitter,
 127                                             final MrSubscriberPollingAdvice mrSubscriberPollingAdvice) {
 128         return IntegrationFlows.from(mrMessageSource, c -> c.poller(pollerMetadata))
 129                 .handle(Http.outboundGateway(m -> String.class.cast(m.getPayload()), mrSubscriberRestTemplate)
 130                         .mappedRequestHeaders(DMAAP_MAPPED_REQUEST_HEADERS)
 131                         .httpMethod(HttpMethod.GET)
 132                         .expectedResponseType(String.class), c -> c.advice(mrSubscriberPollingAdvice))
 133                 .split(mrMessageSplitter)
 134                 .channel(mrSubscriberOutputChannel)
 135                 .get();
 136     }
 137
 138
 139     @Bean
 140     public IntegrationFlow mrPublisherFlow(final MrPublisherPreferences mrPublisherPreferences,
 141                                            final RestTemplate mrPublisherRestTemplate,
 142                                            final DirectChannel mrPublisherInputChannel,
 143                                            final RequestHandlerRetryAdvice requestHandlerRetryAdvice) {
 144         return IntegrationFlows.from(mrPublisherInputChannel)
 145                 .handle(Http.outboundGateway(mrPublisherPreferences.getRequestURL(), mrPublisherRestTemplate)
 146                         .mappedRequestHeaders(DMAAP_MAPPED_REQUEST_HEADERS)
 147                         .httpMethod(HttpMethod.POST)
 148                         .extractPayload(true)
 149                         .expectedResponseType(String.class), c -> c.advice(requestHandlerRetryAdvice))
 150                 .handle(new GenericHandler<String>() {
 151                     @Override
 152                     public Object handle(String payload, MessageHeaders headers) {
 153                           return MessageBuilder.withPayload(payload).copyHeaders(headers)
 154                           .setHeader(AnalyticsHttpConstants.REQUEST_END_TS_HEADER_KEY,
 155                                   AnalyticsWebUtils.CREATION_TIMESTAMP_SUPPLIER.get()).build();
 156                     }
 157                 })
 158                 .channel(DmaapMrConstants.DMAAP_MR_PUBLISHER_OUTPUT_CHANNEL)
 159                 .get();
 160     }
 161
 162 }
 163