X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=dcae-analytics-cdap-tca%2Fsrc%2Fmain%2Fjava%2Forg%2Fopenecomp%2Fdcae%2Fapod%2Fanalytics%2Fcdap%2Ftca%2Fworker%2FTCADMaaPMockSubscriberWorker.java;h=12a52f2e04b6a76c457706515603052eea7246c8;hb=06044df56fb07f4b368888581752855595e7b147;hp=4d721e2910ef0eb8459787dabfbe82a1b86bfdeb;hpb=475cb8c867038acd73ff540173d54bac3947c610;p=dcaegen2%2Fanalytics%2Ftca.git diff --git a/dcae-analytics-cdap-tca/src/main/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMockSubscriberWorker.java b/dcae-analytics-cdap-tca/src/main/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMockSubscriberWorker.java index 4d721e2..12a52f2 100644 --- a/dcae-analytics-cdap-tca/src/main/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMockSubscriberWorker.java +++ b/dcae-analytics-cdap-tca/src/main/java/org/openecomp/dcae/apod/analytics/cdap/tca/worker/TCADMaaPMockSubscriberWorker.java @@ -1,141 +1,141 @@ -/* - * ===============================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.tca.worker; - -import co.cask.cdap.api.annotation.Property; -import co.cask.cdap.api.worker.AbstractWorker; -import co.cask.cdap.api.worker.WorkerContext; -import com.fasterxml.jackson.core.type.TypeReference; -import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences; -import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils; -import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException; -import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.FileNotFoundException; -import java.io.IOException; -import java.io.InputStream; -import java.util.List; - -import static org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils.readValue; -import static org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils.writeValueAsString; - -/** - * CDAP Worker which mocks fetching VES Messages from DMaaP MR topic. - * The mock instead of making DMaaP MR calls will actually take messages - * from file and send them to stream at subscriber polling interval - * - * TODO: To be removed before going to production - only for testing purposes - * - * @author Rajiv Singla . Creation Date: 11/4/2016. - */ -public class TCADMaaPMockSubscriberWorker extends AbstractWorker { - - private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPMockSubscriberWorker.class); - - // TODO: Remove this file before going to production - only for mocking purposes - private static final String MOCK_MESSAGE_FILE_LOCATION = "ves_mock_messages.json"; - private static final TypeReference> EVENT_LISTENER_TYPE_REFERENCE = - new TypeReference>() { - }; - - private TCAAppPreferences tcaAppPreferences; - private boolean stopSendingMessages; - @Property - private final String tcaSubscriberOutputStreamName; - - public TCADMaaPMockSubscriberWorker(final String tcaSubscriberOutputStreamName) { - this.tcaSubscriberOutputStreamName = tcaSubscriberOutputStreamName; - } - - @Override - public void configure() { - setName("MockTCASubscriberWorker"); - setDescription("Writes Mocked VES messages to CDAP Stream"); - LOG.info("Configuring Mock TCA MR DMaaP Subscriber worker with name: {}", "MockTCASubscriberWorker"); - } - - @Override - public void initialize(WorkerContext context) throws Exception { - super.initialize(context); - - final TCAAppPreferences appPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(context); - LOG.info("Initializing Mock TCA MR DMaaP Subscriber worker with preferences: {}", appPreferences); - this.tcaAppPreferences = appPreferences; - this.stopSendingMessages = false; - } - - - @Override - public void run() { - final Integer subscriberPollingInterval = tcaAppPreferences.getSubscriberPollingInterval(); - LOG.debug("Mock TCA Subscriber Polling interval: {}", subscriberPollingInterval); - - final InputStream resourceAsStream = Thread.currentThread().getContextClassLoader().getResourceAsStream - (MOCK_MESSAGE_FILE_LOCATION); - - if (resourceAsStream == null) { - LOG.error("Unable to find file at location: {}", MOCK_MESSAGE_FILE_LOCATION); - throw new DCAEAnalyticsRuntimeException("Unable to find file", LOG, new FileNotFoundException()); - } - - - try { - List eventListeners = readValue(resourceAsStream, EVENT_LISTENER_TYPE_REFERENCE); - - final int totalMessageCount = eventListeners.size(); - LOG.debug("Mock message count to be written to cdap stream: ()", totalMessageCount); - - int i = 1; - for (EventListener eventListener : eventListeners) { - if (stopSendingMessages) { - LOG.debug("Stop sending messages......"); - break; - } - final String eventListenerString = writeValueAsString(eventListener); - LOG.debug("=======>> Writing message to cdap stream no: {} of {}", i, totalMessageCount); - getContext().write(tcaSubscriberOutputStreamName, eventListenerString); - i++; - - try { - Thread.sleep(subscriberPollingInterval); - } catch (InterruptedException e) { - LOG.error("Error while sleeping"); - throw new DCAEAnalyticsRuntimeException("Error while sleeping", LOG, e); - } - } - - LOG.debug("Finished writing mock messages to CDAP Stream"); - - } catch (IOException e) { - LOG.error("Error while parsing json file"); - throw new DCAEAnalyticsRuntimeException("Error while parsing mock json file", LOG, e); - } - - - } - - @Override - public void stop() { - stopSendingMessages = true; - } -} +/* + * ===============================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.tca.worker; + +import co.cask.cdap.api.annotation.Property; +import co.cask.cdap.api.worker.AbstractWorker; +import co.cask.cdap.api.worker.WorkerContext; +import com.fasterxml.jackson.core.type.TypeReference; +import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences; +import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils; +import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException; +import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.List; + +import static org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils.readValue; +import static org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils.writeValueAsString; + +/** + * CDAP Worker which mocks fetching VES Messages from DMaaP MR topic. + * The mock instead of making DMaaP MR calls will actually take messages + * from file and send them to stream at subscriber polling interval + * + * TODO: To be removed before going to production - only for testing purposes + * + * @author Rajiv Singla . Creation Date: 11/4/2016. + */ +public class TCADMaaPMockSubscriberWorker extends AbstractWorker { + + private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPMockSubscriberWorker.class); + + // TODO: Remove this file before going to production - only for mocking purposes + private static final String MOCK_MESSAGE_FILE_LOCATION = "ves_mock_messages.json"; + private static final TypeReference> EVENT_LISTENER_TYPE_REFERENCE = + new TypeReference>() { + }; + + private TCAAppPreferences tcaAppPreferences; + private boolean stopSendingMessages; + @Property + private final String tcaSubscriberOutputStreamName; + + public TCADMaaPMockSubscriberWorker(final String tcaSubscriberOutputStreamName) { + this.tcaSubscriberOutputStreamName = tcaSubscriberOutputStreamName; + } + + @Override + public void configure() { + setName("MockTCASubscriberWorker"); + setDescription("Writes Mocked VES messages to CDAP Stream"); + LOG.info("Configuring Mock TCA MR DMaaP Subscriber worker with name: {}", "MockTCASubscriberWorker"); + } + + @Override + public void initialize(WorkerContext context) throws Exception { + super.initialize(context); + + final TCAAppPreferences appPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(context); + LOG.info("Initializing Mock TCA MR DMaaP Subscriber worker with preferences: {}", appPreferences); + this.tcaAppPreferences = appPreferences; + this.stopSendingMessages = false; + } + + + @Override + public void run() { + final Integer subscriberPollingInterval = tcaAppPreferences.getSubscriberPollingInterval(); + LOG.debug("Mock TCA Subscriber Polling interval: {}", subscriberPollingInterval); + + final InputStream resourceAsStream = Thread.currentThread().getContextClassLoader().getResourceAsStream + (MOCK_MESSAGE_FILE_LOCATION); + + if (resourceAsStream == null) { + LOG.error("Unable to find file at location: {}", MOCK_MESSAGE_FILE_LOCATION); + throw new DCAEAnalyticsRuntimeException("Unable to find file", LOG, new FileNotFoundException()); + } + + + try { + List eventListeners = readValue(resourceAsStream, EVENT_LISTENER_TYPE_REFERENCE); + + final int totalMessageCount = eventListeners.size(); + LOG.debug("Mock message count to be written to cdap stream: ()", totalMessageCount); + + int i = 1; + for (EventListener eventListener : eventListeners) { + if (stopSendingMessages) { + LOG.debug("Stop sending messages......"); + break; + } + final String eventListenerString = writeValueAsString(eventListener); + LOG.debug("=======>> Writing message to cdap stream no: {} of {}", i, totalMessageCount); + getContext().write(tcaSubscriberOutputStreamName, eventListenerString); + i++; + + try { + Thread.sleep(subscriberPollingInterval); + } catch (InterruptedException e) { + LOG.error("Error while sleeping"); + throw new DCAEAnalyticsRuntimeException("Error while sleeping", LOG, e); + } + } + + LOG.debug("Finished writing mock messages to CDAP Stream"); + + } catch (IOException e) { + LOG.error("Error while parsing json file"); + throw new DCAEAnalyticsRuntimeException("Error while parsing mock json file", LOG, e); + } + + + } + + @Override + public void stop() { + stopSendingMessages = true; + } +}