[dcaegen2/analytics/tca.git] /

/*\r
 * ===============================LICENSE_START======================================\r
 *  dcae-analytics\r
 * ================================================================================\r
 *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
 * ================================================================================\r
 *  Licensed under the Apache License, Version 2.0 (the "License");\r
 *  you may not use this file except in compliance with the License.\r
 *   You may obtain a copy of the License at\r
 *\r
 *          http://www.apache.org/licenses/LICENSE-2.0\r
 *\r
 *  Unless required by applicable law or agreed to in writing, software\r
 *  distributed under the License is distributed on an "AS IS" BASIS,\r
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
 *  See the License for the specific language governing permissions and\r
 *  limitations under the License.\r
 *  ============================LICENSE_END===========================================\r
 */\r
\r
package org.openecomp.dcae.apod.analytics.cdap.tca.worker;\r
\r
import co.cask.cdap.api.annotation.Property;\r
import co.cask.cdap.api.worker.AbstractWorker;\r
import co.cask.cdap.api.worker.WorkerContext;\r
import com.fasterxml.jackson.core.type.TypeReference;\r
import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;\r
import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;\r
import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;\r
import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener;\r
import org.slf4j.Logger;\r
import org.slf4j.LoggerFactory;\r
\r
import java.io.FileNotFoundException;\r
import java.io.IOException;\r
import java.io.InputStream;\r
import java.util.List;\r
\r
import static org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils.readValue;\r
import static org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils.writeValueAsString;\r
\r
/**\r
 * CDAP Worker which mocks fetching VES Messages from DMaaP MR topic.\r
 * The mock instead of making DMaaP MR calls will actually take messages\r
 * from file and send them to stream at subscriber polling interval\r
 *\r
 * TODO: To be removed before going to production - only for testing purposes\r
 *\r
 * @author Rajiv Singla . Creation Date: 11/4/2016.\r
 */\r
public class TCADMaaPMockSubscriberWorker extends AbstractWorker {\r
\r
    private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPMockSubscriberWorker.class);\r
\r
    // TODO: Remove this file before going to production - only for mocking purposes\r
    private static final String MOCK_MESSAGE_FILE_LOCATION = "ves_mock_messages.json";\r
    private static final TypeReference<List<EventListener>> EVENT_LISTENER_TYPE_REFERENCE =\r
            new TypeReference<List<EventListener>>() {\r
            };\r
\r
    private TCAAppPreferences tcaAppPreferences;\r
    private boolean stopSendingMessages;\r
    @Property\r
    private final String tcaSubscriberOutputStreamName;\r
\r
    public TCADMaaPMockSubscriberWorker(final String tcaSubscriberOutputStreamName) {\r
        this.tcaSubscriberOutputStreamName = tcaSubscriberOutputStreamName;\r
    }\r
\r
    @Override\r
    public void configure() {\r
        setName("MockTCASubscriberWorker");\r
        setDescription("Writes Mocked VES messages to CDAP Stream");\r
        LOG.info("Configuring Mock TCA MR DMaaP Subscriber worker with name: {}", "MockTCASubscriberWorker");\r
    }\r
\r
    @Override\r
    public void initialize(WorkerContext context) throws Exception {\r
        super.initialize(context);\r
\r
        final TCAAppPreferences appPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(context);\r
        LOG.info("Initializing Mock TCA MR DMaaP Subscriber worker with preferences: {}", appPreferences);\r
        this.tcaAppPreferences = appPreferences;\r
        this.stopSendingMessages = false;\r
    }\r
\r
\r
    @Override\r
    public void run() {\r
        final Integer subscriberPollingInterval = tcaAppPreferences.getSubscriberPollingInterval();\r
        LOG.debug("Mock TCA Subscriber Polling interval: {}", subscriberPollingInterval);\r
\r
        final InputStream resourceAsStream = Thread.currentThread().getContextClassLoader().getResourceAsStream\r
                (MOCK_MESSAGE_FILE_LOCATION);\r
\r
        if (resourceAsStream == null) {\r
            LOG.error("Unable to find file at location: {}", MOCK_MESSAGE_FILE_LOCATION);\r
            throw new DCAEAnalyticsRuntimeException("Unable to find file", LOG, new FileNotFoundException());\r
        }\r
\r
\r
        try {\r
            List<EventListener> eventListeners = readValue(resourceAsStream, EVENT_LISTENER_TYPE_REFERENCE);\r
\r
            final int totalMessageCount = eventListeners.size();\r
            LOG.debug("Mock message count to be written to cdap stream: ()", totalMessageCount);\r
\r
            int i = 1;\r
            for (EventListener eventListener : eventListeners) {\r
                if (stopSendingMessages) {\r
                    LOG.debug("Stop sending messages......");\r
                    break;\r
                }\r
                final String eventListenerString = writeValueAsString(eventListener);\r
                LOG.debug("=======>> Writing message to cdap stream no: {} of {}", i, totalMessageCount);\r
                getContext().write(tcaSubscriberOutputStreamName, eventListenerString);\r
                i++;\r
\r
                try {\r
                    Thread.sleep(subscriberPollingInterval);\r
                } catch (InterruptedException e) {\r
                    LOG.error("Error while sleeping");\r
                    throw new DCAEAnalyticsRuntimeException("Error while sleeping", LOG, e);\r
                }\r
            }\r
\r
            LOG.debug("Finished writing mock messages to CDAP Stream");\r
\r
        } catch (IOException e) {\r
            LOG.error("Error while parsing json file");\r
            throw new DCAEAnalyticsRuntimeException("Error while parsing mock json file", LOG, e);\r
        }\r
\r
\r
    }\r
\r
    @Override\r
    public void stop() {\r
        stopSendingMessages = true;\r
    }\r
}\r