gerrit.onap Code Review - dcaegen2/analytics/tca-gen2.git/blob

   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.tca.web.config;
  21
  22 import static org.onap.dcae.analytics.model.AnalyticsHttpConstants.REQUEST_ID_HEADER_KEY;
  23 import static org.onap.dcae.analytics.model.AnalyticsHttpConstants.REQUEST_TRANSACTION_ID_HEADER_KEY;
  24
  25 import java.util.List;
  26 import java.util.Map;
  27
  28 import org.onap.dcae.analytics.model.AnalyticsProfile;
  29 import org.onap.dcae.analytics.model.DmaapMrConstants;
  30 import org.onap.dcae.analytics.tca.web.TcaAppProperties;
  31 import org.onap.dcae.analytics.tca.web.domain.TcaPolicyWrapper;
  32 import org.onap.dcae.analytics.tca.web.integration.TcaAlertTransformer;
  33 import org.onap.dcae.analytics.tca.web.integration.TcaPublisherResponseHandler;
  34 import org.onap.dcae.analytics.tca.web.service.TcaProcessingService;
  35 import org.onap.dcae.analytics.tca.web.util.function.TcaAppPropsToMrPublisherPrefsFunction;
  36 import org.onap.dcae.analytics.tca.web.util.function.TcaAppPropsToMrSubscriberPrefsFunction;
  37 import org.onap.dcae.analytics.web.dmaap.MrPublisherPreferences;
  38 import org.onap.dcae.analytics.web.dmaap.MrSubscriberPreferences;
  39 import org.springframework.context.annotation.Bean;
  40 import org.springframework.context.annotation.Configuration;
  41 import org.springframework.context.annotation.Profile;
  42 import org.springframework.integration.channel.DirectChannel;
  43 import org.springframework.integration.channel.NullChannel;
  44 import org.springframework.integration.channel.QueueChannel;
  45 import org.springframework.integration.dsl.IntegrationFlow;
  46 import org.springframework.integration.dsl.IntegrationFlows;
  47 import org.springframework.messaging.MessageHeaders;
  48
  49 /**
  50  * @author Rajiv Singla
  51  */
  52 @Profile({AnalyticsProfile.DMAAP_PROFILE_NAME})
  53 @Configuration
  54 public class TcaMrConfig {
  55
  56     @Bean
  57     public MrSubscriberPreferences mrSubscriberPreferences(final TcaAppProperties tcaAppProperties) {
  58         return new TcaAppPropsToMrSubscriberPrefsFunction().apply(tcaAppProperties);
  59     }
  60
  61     @Bean
  62     public MrPublisherPreferences mrPublisherPreferences(final TcaAppProperties tcaAppProperties) {
  63         return new TcaAppPropsToMrPublisherPrefsFunction().apply(tcaAppProperties);
  64     }
  65
  66     @Bean
  67     public Integer processingBatchSize(final TcaAppProperties tcaAppProperties) {
  68         return tcaAppProperties.getTca().getProcessingBatchSize();
  69     }
  70
  71     @Bean
  72     public TcaAlertTransformer tcaAlertTransformer(final TcaAppProperties tcaAppProperties) {
  73         return new TcaAlertTransformer(tcaAppProperties);
  74     }
  75
  76     @Bean
  77     public IntegrationFlow tcaMrFlow(final TcaPolicyWrapper tcaPolicyWrapper,
  78                                      final QueueChannel mrSubscriberOutputChannel,
  79                                      final DirectChannel mrPublisherInputChannel,
  80                                      final TcaProcessingService tcaProcessingService,
  81                                      final TcaAlertTransformer tcaAlertTransformer) {
  82         // get messages from dmaap subscriber channel
  83         return IntegrationFlows.from(mrSubscriberOutputChannel)
  84                 // handle incoming message from dmaap
  85                 .handle((List<String> cefMessages, Map<String, Object> headers) ->
  86                         tcaProcessingService.getTcaExecutionResults(
  87                                 headers.getOrDefault(REQUEST_ID_HEADER_KEY, headers.get(MessageHeaders.ID)).toString(),
  88                                 headers.getOrDefault(REQUEST_TRANSACTION_ID_HEADER_KEY, "").toString(),
  89                                 tcaPolicyWrapper.getTcaPolicy(), cefMessages))
  90                 // transform tca execution results to alerts - if not alerts are detected terminate further processing
  91                 .transform(tcaAlertTransformer, c -> c.requiresReply(false))
  92                 // post messages to dmaap publisher input channel
  93                 .channel(mrPublisherInputChannel)
  94                 .get();
  95     }
  96
  97
  98     @Bean
  99     public TcaPublisherResponseHandler tcaPublisherResponseHandler(final TcaAppProperties tcaAppProperties) {
 100         return new TcaPublisherResponseHandler(tcaAppProperties);
 101     }
 102
 103     @Bean
 104     public IntegrationFlow tcaPublisherResponseFlow(final TcaPublisherResponseHandler tcaPublisherResponseHandler) {
 105         return IntegrationFlows.from(DmaapMrConstants.DMAAP_MR_PUBLISHER_OUTPUT_CHANNEL)
 106                 // log response from dmaap publisher output channel
 107                 .handle(tcaPublisherResponseHandler)
 108                 // finish processing
 109                 .channel(new NullChannel())
 110                 .get();
 111     }
 112
 113 }