4df35e7f68279107fe70e8b31f6b1d31a23371d7
[dcaegen2/analytics/tca.git] / dcae-analytics-cdap-tca / src / main / java / org / openecomp / dcae / apod / analytics / cdap / tca / flow / TCAVESCollectorFlow.java
1 /*
2  * ===============================LICENSE_START======================================
3  *  dcae-analytics
4  * ================================================================================
5  *    Copyright © 2017 AT&T 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 package org.openecomp.dcae.apod.analytics.cdap.tca.flow;
22
23 import co.cask.cdap.api.flow.AbstractFlow;
24 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
25 import org.openecomp.dcae.apod.analytics.cdap.tca.flowlet.TCAVESAlertsAbatementFlowlet;
26 import org.openecomp.dcae.apod.analytics.cdap.tca.flowlet.TCAVESAlertsSinkFlowlet;
27 import org.openecomp.dcae.apod.analytics.cdap.tca.flowlet.TCAVESMessageRouterFlowlet;
28 import org.openecomp.dcae.apod.analytics.cdap.tca.flowlet.TCAVESThresholdViolationCalculatorFlowlet;
29 import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppConfig;
30
31 /**
32  * TCA Flow for VES (Virtual Event Streaming) Collector Flow
33  *
34  * @author Rajiv Singla . Creation Date: 11/3/2016.
35  */
36 public class TCAVESCollectorFlow extends AbstractFlow {
37
38     private final TCAAppConfig tcaAppConfig;
39
40     public TCAVESCollectorFlow(TCAAppConfig tcaAppConfig) {
41         this.tcaAppConfig = tcaAppConfig;
42     }
43
44     @Override
45     protected void configure() {
46
47         setName(CDAPComponentsConstants.TCA_FIXED_VES_COLLECTOR_NAME_FLOW);
48         setDescription(CDAPComponentsConstants.TCA_FIXED_VES_COLLECTOR_DESCRIPTION_FLOW);
49
50         final TCAVESMessageRouterFlowlet messageRouterFlowlet = new TCAVESMessageRouterFlowlet();
51         addFlowlet(messageRouterFlowlet);
52
53         final TCAVESThresholdViolationCalculatorFlowlet thresholdViolationCalculatorFlowlet =
54                 new TCAVESThresholdViolationCalculatorFlowlet(tcaAppConfig.getTcaVESMessageStatusTableName());
55         addFlowlet(thresholdViolationCalculatorFlowlet, tcaAppConfig.getThresholdCalculatorFlowletInstances());
56
57         final TCAVESAlertsAbatementFlowlet tcavesAlertsAbatementFlowlet =
58                 new TCAVESAlertsAbatementFlowlet(tcaAppConfig.getTcaAlertsAbatementTableName());
59         addFlowlet(tcavesAlertsAbatementFlowlet);
60
61         final TCAVESAlertsSinkFlowlet alertsSinkFlowlet =
62                 new TCAVESAlertsSinkFlowlet(tcaAppConfig.getTcaVESAlertsTableName());
63         addFlowlet(alertsSinkFlowlet);
64
65
66         // connect DMaaP MR VES Subscriber output stream to VES Message Router Flowlet
67         connectStream(tcaAppConfig.getTcaSubscriberOutputStreamName(), messageRouterFlowlet);
68         // connect message router to VES threshold calculator
69         connect(messageRouterFlowlet, thresholdViolationCalculatorFlowlet);
70         // connect VES threshold calculator flowlet to Alerts Abatement Flowlet
71         connect(thresholdViolationCalculatorFlowlet, tcavesAlertsAbatementFlowlet);
72         // connect Alerts Abatement flowlet to Alerts Sink Flowlet
73         connect(tcavesAlertsAbatementFlowlet, alertsSinkFlowlet);
74
75     }
76 }