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