dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.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.plugins.streaming.dmaap;\r
  22 \r
  23 import co.cask.cdap.api.data.format.StructuredRecord;\r
  24 import co.cask.cdap.api.metrics.Metrics;\r
  25 import com.google.common.base.Optional;\r
  26 import org.apache.spark.storage.StorageLevel;\r
  27 import org.apache.spark.streaming.receiver.Receiver;\r
  28 import org.openecomp.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils;\r
  29 import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;\r
  30 import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils;\r
  31 import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper;\r
  32 import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;\r
  33 import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;\r
  34 import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;\r
  35 import org.slf4j.Logger;\r
  36 import org.slf4j.LoggerFactory;\r
  37 \r
  38 import java.util.List;\r
  39 import java.util.concurrent.TimeUnit;\r
  40 \r
  41 /**\r
  42  * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records\r
  43  * <p>\r
  44  * @author Rajiv Singla . Creation Date: 1/19/2017.\r
  45  */\r
  46 public class DMaaPMRReceiver extends Receiver<StructuredRecord> {\r
  47 \r
  48     private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRReceiver.class);\r
  49     private static final long serialVersionUID = 1L;\r
  50 \r
  51     private final DMaaPMRSourcePluginConfig pluginConfig;\r
  52     private final Metrics metrics;\r
  53 \r
  54     public DMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig,\r
  55                            final Metrics metrics) {\r
  56         super(storageLevel);\r
  57         this.pluginConfig = pluginConfig;\r
  58         this.metrics = metrics;\r
  59         LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig);\r
  60     }\r
  61 \r
  62     @Override\r
  63     public void onStart() {\r
  64 \r
  65         // create DMaaP MR Subscriber\r
  66         final DMaaPMRSubscriber subscriber =\r
  67                 DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig));\r
  68 \r
  69         // Start a new thread with indefinite loop until receiver is stopped\r
  70         new Thread() {\r
  71             @Override\r
  72             public void run() {\r
  73                 while (!isStopped()) {\r
  74                     storeStructuredRecords(subscriber);\r
  75                     try {\r
  76                         final Integer pollingInterval = pluginConfig.getPollingInterval();\r
  77                         LOG.debug("DMaaP MR Receiver sleeping for polling interval: {}", pollingInterval);\r
  78                         TimeUnit.MILLISECONDS.sleep(pollingInterval);\r
  79                     } catch (InterruptedException e) {\r
  80                         final String errorMessage = String.format(\r
  81                                 "Interrupted Exception while DMaaP MR Receiver sleeping polling interval: %s", e);\r
  82                         throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);\r
  83                     }\r
  84                 }\r
  85             }\r
  86         }.start();\r
  87 \r
  88     }\r
  89 \r
  90     @Override\r
  91     public void onStop() {\r
  92         LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig);\r
  93     }\r
  94 \r
  95     /**\r
  96      * Fetches records from DMaaP MR Subscriber and store them as structured records\r
  97      *\r
  98      * @param subscriber DMaaP MR Subscriber Instance\r
  99      */\r
 100     public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) {\r
 101 \r
 102         LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic");\r
 103 \r
 104         // Fetch messages from DMaaP MR Topic\r
 105         final Optional<List<String>> subscriberMessagesOptional =\r
 106                 DMaaPMRUtils.getSubscriberMessages(subscriber, metrics);\r
 107 \r
 108         // store records\r
 109         if (subscriberMessagesOptional.isPresent()) {\r
 110             final List<String> messages = subscriberMessagesOptional.get();\r
 111             for (final String message : messages) {\r
 112                 store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(message));\r
 113             }\r
 114             LOG.debug("Stored DMaaP Subscriber messages as Structured Records. Message count {}", messages.size());\r
 115         }\r
 116     }\r
 117 \r
 118 }\r