X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=dcae-analytics-cdap-plugins%2Fsrc%2Fmain%2Fjava%2Forg%2Fonap%2Fdcae%2Fapod%2Fanalytics%2Fcdap%2Fplugins%2Fstreaming%2Fdmaap%2FDMaaPMRReceiver.java;fp=dcae-analytics-cdap-plugins%2Fsrc%2Fmain%2Fjava%2Forg%2Fopenecomp%2Fdcae%2Fapod%2Fanalytics%2Fcdap%2Fplugins%2Fstreaming%2Fdmaap%2FDMaaPMRReceiver.java;h=593d5d154326222d571906ff855009290b6613ff;hb=c489a2eb22484e798c39a978bc8b61821b92322f;hp=98227685b47c1cb3ed9447ae238bdf196aa39b01;hpb=baaecfc19132034168beb64a0c1e7813c7c17c4a;p=dcaegen2%2Fanalytics%2Ftca.git diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java similarity index 85% rename from dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java rename to dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java index 9822768..593d5d1 100644 --- a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java +++ b/dcae-analytics-cdap-plugins/src/main/java/org/onap/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java @@ -1,118 +1,118 @@ -/* - * ===============================LICENSE_START====================================== - * dcae-analytics - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ============================LICENSE_END=========================================== - */ - -package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap; - -import co.cask.cdap.api.data.format.StructuredRecord; -import co.cask.cdap.api.metrics.Metrics; -import com.google.common.base.Optional; -import org.apache.spark.storage.StorageLevel; -import org.apache.spark.streaming.receiver.Receiver; -import org.openecomp.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils; -import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig; -import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils; -import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper; -import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException; -import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory; -import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.List; -import java.util.concurrent.TimeUnit; - -/** - * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records - *

- * @author Rajiv Singla . Creation Date: 1/19/2017. - */ -public class DMaaPMRReceiver extends Receiver { - - private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRReceiver.class); - private static final long serialVersionUID = 1L; - - private final DMaaPMRSourcePluginConfig pluginConfig; - private final transient Metrics metrics; - - public DMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig, - final Metrics metrics) { - super(storageLevel); - this.pluginConfig = pluginConfig; - this.metrics = metrics; - LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig); - } - - @Override - public void onStart() { - - // create DMaaP MR Subscriber - final DMaaPMRSubscriber subscriber = - DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig)); - - // Start a new thread with indefinite loop until receiver is stopped - new Thread() { - @Override - public void run() { - while (!isStopped()) { - storeStructuredRecords(subscriber); - try { - final Integer pollingInterval = pluginConfig.getPollingInterval(); - LOG.debug("DMaaP MR Receiver sleeping for polling interval: {}", pollingInterval); - TimeUnit.MILLISECONDS.sleep(pollingInterval); - } catch (InterruptedException e) { - final String errorMessage = String.format( - "Interrupted Exception while DMaaP MR Receiver sleeping polling interval: %s", e); - throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e); - } - } - } - }.start(); - - } - - @Override - public void onStop() { - LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig); - } - - /** - * Fetches records from DMaaP MR Subscriber and store them as structured records - * - * @param subscriber DMaaP MR Subscriber Instance - */ - public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) { - - LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic"); - - // Fetch messages from DMaaP MR Topic - final Optional> subscriberMessagesOptional = - DMaaPMRUtils.getSubscriberMessages(subscriber, metrics); - - // store records - if (subscriberMessagesOptional.isPresent()) { - final List messages = subscriberMessagesOptional.get(); - for (final String message : messages) { - store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(message)); - } - LOG.debug("Stored DMaaP Subscriber messages as Structured Records. Message count {}", messages.size()); - } - } - -} +/* + * ===============================LICENSE_START====================================== + * dcae-analytics + * ================================================================================ + * Copyright © 2017 AT&T Intellectual Property. All rights reserved. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============================LICENSE_END=========================================== + */ + +package org.onap.dcae.apod.analytics.cdap.plugins.streaming.dmaap; + +import co.cask.cdap.api.data.format.StructuredRecord; +import co.cask.cdap.api.metrics.Metrics; +import com.google.common.base.Optional; +import org.apache.spark.storage.StorageLevel; +import org.apache.spark.streaming.receiver.Receiver; +import org.onap.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils; +import org.onap.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig; +import org.onap.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils; +import org.onap.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper; +import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException; +import org.onap.dcae.apod.analytics.dmaap.DMaaPMRFactory; +import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records + *

+ * @author Rajiv Singla . Creation Date: 1/19/2017. + */ +public class DMaaPMRReceiver extends Receiver { + + private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRReceiver.class); + private static final long serialVersionUID = 1L; + + private final DMaaPMRSourcePluginConfig pluginConfig; + private final transient Metrics metrics; + + public DMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig, + final Metrics metrics) { + super(storageLevel); + this.pluginConfig = pluginConfig; + this.metrics = metrics; + LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig); + } + + @Override + public void onStart() { + + // create DMaaP MR Subscriber + final DMaaPMRSubscriber subscriber = + DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig)); + + // Start a new thread with indefinite loop until receiver is stopped + new Thread() { + @Override + public void run() { + while (!isStopped()) { + storeStructuredRecords(subscriber); + try { + final Integer pollingInterval = pluginConfig.getPollingInterval(); + LOG.debug("DMaaP MR Receiver sleeping for polling interval: {}", pollingInterval); + TimeUnit.MILLISECONDS.sleep(pollingInterval); + } catch (InterruptedException e) { + final String errorMessage = String.format( + "Interrupted Exception while DMaaP MR Receiver sleeping polling interval: %s", e); + throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e); + } + } + } + }.start(); + + } + + @Override + public void onStop() { + LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig); + } + + /** + * Fetches records from DMaaP MR Subscriber and store them as structured records + * + * @param subscriber DMaaP MR Subscriber Instance + */ + public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) { + + LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic"); + + // Fetch messages from DMaaP MR Topic + final Optional> subscriberMessagesOptional = + DMaaPMRUtils.getSubscriberMessages(subscriber, metrics); + + // store records + if (subscriberMessagesOptional.isPresent()) { + final List messages = subscriberMessagesOptional.get(); + for (final String message : messages) { + store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(message)); + } + LOG.debug("Stored DMaaP Subscriber messages as Structured Records. Message count {}", messages.size()); + } + } + +}