path: root/app-c/appc/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java

/*-
 * ============LICENSE_START=======================================================
 * openECOMP : APP-C
 * ================================================================================
 * Copyright (C) 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.appc.listener.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.openecomp.appc.adapter.dmaap.Consumer;
import org.openecomp.appc.adapter.dmaap.DmaapConsumer;
import org.openecomp.appc.adapter.dmaap.DmaapProducer;
import org.openecomp.appc.adapter.dmaap.Producer;
import org.openecomp.appc.adapter.dmaap.DmaapConsumer;
import org.openecomp.appc.adapter.dmaap.DmaapProducer;
import org.openecomp.appc.listener.EventHandler;
import org.openecomp.appc.listener.ListenerProperties;
import org.openecomp.appc.listener.ListenerProperties.MessageService;
import org.openecomp.appc.listener.util.Mapper;
import org.openecomp.appc.logging.LoggingConstants;

import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;
import org.slf4j.MDC;

/**
 * This class is a wrapper for the DMaaP client provided in appc-dmaap-adapter. Its aim is to ensure that only well formed
 * messages are sent and received on DMaaP.
 * 
 */
public class EventHandlerImpl implements EventHandler {

    private final EELFLogger LOG = EELFManager.getInstance().getLogger(EventHandlerImpl.class);

    /*
     * The amount of time in seconds to keep a connection to a topic open while waiting for data
     */
    private int READ_TIMEOUT = 60;

    /*
     * The pool of hosts to query against
     */
    private Collection<String> pool;

    /*
     * The topic to read messages from
     */
    private String readTopic;

    /*
     * The topic to write messages to
     */
    private Set<String> writeTopics;

    /*
     * The client (group) name to use for reading messages
     */
    private String clientName;

    /*
     * The id of the client (group) that is reading messages
     */
    private String clientId;

    /*
     * The api public key to use for authentication
     */
    private String apiKey;

    /*
     * The api secret key to use for authentication
     */
    private String apiSecret;

    /*
     * A json object containing filter arguments.
     */
    private String filter_json;

    private MessageService messageService;

    private Consumer reader = null;
    private Producer producer = null;
    
    public EventHandlerImpl(ListenerProperties props) {
        pool = new HashSet<String>();
        writeTopics = new HashSet<String>();

        if (props != null) {
            readTopic = props.getProperty(ListenerProperties.KEYS.TOPIC_READ);
            clientName = props.getProperty(ListenerProperties.KEYS.CLIENT_NAME, "APP-C");
            clientId = props.getProperty(ListenerProperties.KEYS.CLIENT_ID, "0");
            apiKey = props.getProperty(ListenerProperties.KEYS.AUTH_USER_KEY);
            apiSecret = props.getProperty(ListenerProperties.KEYS.AUTH_SECRET_KEY);

            filter_json = props.getProperty(ListenerProperties.KEYS.TOPIC_READ_FILTER);

            READ_TIMEOUT = Integer
                .valueOf(props.getProperty(ListenerProperties.KEYS.TOPIC_READ_TIMEOUT, String.valueOf(READ_TIMEOUT)));

            String hostnames = props.getProperty(ListenerProperties.KEYS.HOSTS);
            if (hostnames != null && !hostnames.isEmpty()) {
                for (String name : hostnames.split(",")) {
                    pool.add(name);
                }
            }

            String writeTopicStr = props.getProperty(ListenerProperties.KEYS.TOPIC_WRITE);
            if (writeTopicStr != null) {
                for (String topic : writeTopicStr.split(",")) {
                    writeTopics.add(topic);
                }
            }

            messageService = MessageService.parse(props.getProperty(ListenerProperties.KEYS.MESSAGE_SERVICE));

            LOG.info(String.format(
                "Configured to use %s client on host pool [%s]. Reading from [%s] filtered by %s. Wriring to [%s]. Authenticated using %s",
                messageService, hostnames, readTopic, filter_json, writeTopics, apiKey));
        }
    }

    @Override
    public List<String> getIncomingEvents() {
        return getIncomingEvents(1000);
    }

    @Override
    public List<String> getIncomingEvents(int limit) {
        List<String> out = new ArrayList<String>();
        LOG.info(String.format("Getting up to %d incoming events", limit));
        // reuse the consumer object instead of creating a new one every time
        if (reader == null) {
        	LOG.info("Getting Consumer...");
        	reader = getConsumer();
        }
        for (String item : reader.fetch(READ_TIMEOUT * 1000, limit)) {
            out.add(item);
        }
        LOG.info(String.format("Read %d messages from %s as %s/%s.", out.size(), readTopic, clientName, clientId));
        return out;
    }

    @Override
    public <T> List<T> getIncomingEvents(Class<T> cls) {
        return getIncomingEvents(cls, 1000);
    }

    @Override
    public <T> List<T> getIncomingEvents(Class<T> cls, int limit) {
        List<String> incomingStrings = getIncomingEvents(limit);
        return Mapper.mapList(incomingStrings, cls);
    }

    @Override
    public void postStatus(String event) {
        postStatus(null, event);
    }

    @Override
    public void postStatus(String partition, String event) {
        LOG.debug(String.format("Posting Message [%s]", event));
        if (producer == null) {
        	LOG.info("Getting Producer...");
        	producer = getProducer();
        }
        producer.post(partition, event);
    }

    /**
     * Returns a consumer object for direct access to our Cambria consumer interface
     * 
     * @return An instance of the consumer interface
     */
    protected Consumer getConsumer() {
        LOG.debug(String.format("Getting Consumer: %s  %s/%s/%s", pool, readTopic, clientName, clientId));
        if (filter_json == null && writeTopics.contains(readTopic)) {
            LOG.error(
                "*****We will be writing and reading to the same topic without a filter. This will cause an infinite loop.*****");
        }
        Consumer out;
        out = new DmaapConsumer(pool, readTopic, clientName, clientId, filter_json, apiKey, apiSecret);
        for (String url : pool) {
            if (url.contains("3905") || url.contains("https")) {
                out.useHttps(true);
                break;
            }
        }
        return out;
    }

    /**
     * Returns a consumer object for direct access to our Cambria producer interface
     * 
     * @return An instance of the producer interface
     */
    protected Producer getProducer() {
        LOG.debug(String.format("Getting Producer: %s  %s", pool, readTopic));

        Producer out;
        out = new DmaapProducer(pool,writeTopics);

        if (apiKey != null && apiSecret != null) {
            out.updateCredentials(apiKey, apiSecret);
        }

        for (String url : pool) {
            if (url.contains("3905") || url.contains("https")) {
                out.useHttps(true);
                break;
            }
        }
        return out;
    }

    @Override
    public void closeClients() {
    	LOG.debug("Closing Consumer and Producer DMaaP clients");
        switch (messageService) {
            case DMaaP:
                if (reader != null) {
                	((DmaapConsumer) reader).close();
                }
                if (producer != null) {
                	((DmaapProducer) producer).close();
                }
                break;
            default:
            	// close DMaaP clients
                if (reader != null) {
                	((DmaapConsumer) reader).close();
                }
                if (producer != null) {
                	((DmaapProducer) producer).close();
                }
        }
    }
    
    @Override
    public String getClientId() {
        return clientId;
    }

    @Override
    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    @Override
    public String getClientName() {
        return clientName;
    }

    @Override
    public void setClientName(String clientName) {
        this.clientName = clientName;
        MDC.put(LoggingConstants.MDCKeys.PARTNER_NAME, clientName);
    }

    @Override
    public void addToPool(String hostname) {
        pool.add(hostname);
    }

    @Override
    public Collection<String> getPool() {
        return pool;
    }

    @Override
    public void removeFromPool(String hostname) {
        pool.remove(hostname);
    }

    @Override
    public String getReadTopic() {
        return readTopic;
    }

    @Override
    public void setReadTopic(String readTopic) {
        this.readTopic = readTopic;
    }

    @Override
    public Set<String> getWriteTopics() {
        return writeTopics;
    }

    @Override
    public void setWriteTopics(Set<String> writeTopics) {
        this.writeTopics = writeTopics;
    }

    @Override
    public void clearCredentials() {
        apiKey = null;
        apiSecret = null;
    }

    @Override
    public void setCredentials(String key, String secret) {
        apiKey = key;
        apiSecret = secret;
    }
}