dcae-analytics-cdap-tca/src/main/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPSubscriberWorker.java

   1 /*\r
   2  * ===============================LICENSE_START======================================\r
   3  *  dcae-analytics\r
   4  * ================================================================================\r
   5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
   6  * ================================================================================\r
   7  *  Licensed under the Apache License, Version 2.0 (the "License");\r
   8  *  you may not use this file except in compliance with the License.\r
   9  *   You may obtain a copy of the License at\r
  10  *\r
  11  *          http://www.apache.org/licenses/LICENSE-2.0\r
  12  *\r
  13  *  Unless required by applicable law or agreed to in writing, software\r
  14  *  distributed under the License is distributed on an "AS IS" BASIS,\r
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  16  *  See the License for the specific language governing permissions and\r
  17  *  limitations under the License.\r
  18  *  ============================LICENSE_END===========================================\r
  19  */\r
  20 \r
  21 package org.openecomp.dcae.apod.analytics.cdap.tca.worker;\r
  22 \r
  23 import co.cask.cdap.api.annotation.Property;\r
  24 import co.cask.cdap.api.metrics.Metrics;\r
  25 import co.cask.cdap.api.worker.WorkerContext;\r
  26 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;\r
  27 import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;\r
  28 import org.openecomp.dcae.apod.analytics.cdap.tca.utils.AppPreferencesToSubscriberConfigMapper;\r
  29 import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;\r
  30 import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;\r
  31 import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;\r
  32 import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;\r
  33 import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;\r
  34 import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;\r
  35 import org.quartz.JobDataMap;\r
  36 import org.quartz.SchedulerException;\r
  37 import org.quartz.impl.StdSchedulerFactory;\r
  38 import org.slf4j.Logger;\r
  39 import org.slf4j.LoggerFactory;\r
  40 \r
  41 import java.util.concurrent.atomic.AtomicBoolean;\r
  42 \r
  43 /**\r
  44  * TCA DMaaP Subscriber will read messages and post them to cdap stream at regular intervals\r
  45  * <p>\r
  46  * @author Rajiv Singla . Creation Date: 10/14/2016.\r
  47  */\r
  48 public class TCADMaaPSubscriberWorker extends BaseTCADMaaPMRWorker {\r
  49 \r
  50     private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPSubscriberWorker.class);\r
  51 \r
  52     private DMaaPMRSubscriber subscriber;\r
  53     private Metrics metrics;\r
  54     @Property\r
  55     private final String tcaSubscriberOutputStreamName;\r
  56 \r
  57     public TCADMaaPSubscriberWorker(final String tcaSubscriberOutputStreamName) {\r
  58         this.tcaSubscriberOutputStreamName = tcaSubscriberOutputStreamName;\r
  59     }\r
  60 \r
  61 \r
  62     @Override\r
  63     public void configure() {\r
  64         setName(CDAPComponentsConstants.TCA_FIXED_DMAAP_SUBSCRIBER_WORKER);\r
  65         setDescription(CDAPComponentsConstants.TCA_FIXED_DMAAP_SUBSCRIBER_DESCRIPTION_WORKER);\r
  66         LOG.debug("Configuring TCA MR DMaaP Subscriber worker with name: {}",\r
  67                 CDAPComponentsConstants.TCA_FIXED_DMAAP_SUBSCRIBER_WORKER);\r
  68     }\r
  69 \r
  70     @Override\r
  71     public void initialize(WorkerContext context) throws Exception {\r
  72         super.initialize(context);\r
  73 \r
  74         // Parse runtime arguments\r
  75         final TCAAppPreferences tcaAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(context);\r
  76 \r
  77         LOG.info("Initializing TCA MR DMaaP Subscriber worker with preferences: {}", tcaAppPreferences);\r
  78 \r
  79         // Map TCA App Preferences to DMaaP MR Subscriber Config\r
  80         final DMaaPMRSubscriberConfig subscriberConfig = AppPreferencesToSubscriberConfigMapper.map(tcaAppPreferences);\r
  81 \r
  82         LOG.info("TCA DMaaP MR Subscriber worker will be writing to CDAP Stream: {}", tcaSubscriberOutputStreamName);\r
  83 \r
  84         // Create an instance of DMaaP MR Subscriber\r
  85         LOG.debug("Creating an instance of DMaaP Subscriber");\r
  86         subscriber = DMaaPMRFactory.create().createSubscriber(subscriberConfig);\r
  87 \r
  88         // initialize a new Quartz scheduler\r
  89         initializeScheduler(tcaAppPreferences, new StdSchedulerFactory());\r
  90 \r
  91         // initialize scheduler state\r
  92         isSchedulerShutdown = new AtomicBoolean(true);\r
  93     }\r
  94 \r
  95     /**\r
  96      * Initializes a scheduler instance for DMaaP MR Subscriber Job\r
  97      *\r
  98      * @throws SchedulerException SchedulerException\r
  99      */\r
 100     private void initializeScheduler(final TCAAppPreferences tcaAppPreferences,\r
 101                                      final StdSchedulerFactory stdSchedulerFactory) throws SchedulerException {\r
 102 \r
 103         // Get Subscriber polling interval\r
 104         final Integer subscriberPollingInterval = tcaAppPreferences.getSubscriberPollingInterval();\r
 105 \r
 106         // Subscriber Quartz Properties file\r
 107         final String quartzSubscriberPropertiesFileName = AnalyticsConstants.TCA_QUARTZ_SUBSCRIBER_PROPERTIES_FILE_NAME;\r
 108 \r
 109         // Create a new JobDataMap containing information required by TCA DMaaP Subscriber Job\r
 110         final JobDataMap jobDataMap = new JobDataMap();\r
 111         jobDataMap.put(AnalyticsConstants.CDAP_STREAM_VARIABLE_NAME, tcaSubscriberOutputStreamName);\r
 112         jobDataMap.put(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME, getContext());\r
 113         jobDataMap.put(AnalyticsConstants.DMAAP_SUBSCRIBER_VARIABLE_NAME, subscriber);\r
 114         jobDataMap.put(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME, metrics);\r
 115 \r
 116         // Create new publisher scheduler\r
 117         scheduler = TCAUtils.createQuartzScheduler(subscriberPollingInterval, stdSchedulerFactory,\r
 118                 quartzSubscriberPropertiesFileName, jobDataMap, TCADMaaPMRSubscriberJob.class,\r
 119                 AnalyticsConstants.TCA_DMAAP_SUBSCRIBER_QUARTZ_JOB_NAME,\r
 120                 AnalyticsConstants.TCA_DMAAP_SUBSCRIBER_QUARTZ_TRIGGER_NAME);\r
 121     }\r
 122 \r
 123 \r
 124 }\r