/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */

package org.apache.hcatalog.listener;

import java.util.ArrayList;
import java.util.HashMap;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStore.HMSHandler;
import org.apache.hadoop.hive.metastore.MetaStoreEventListener;
import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.events.AddPartitionEvent;
import org.apache.hadoop.hive.metastore.events.AlterPartitionEvent;
import org.apache.hadoop.hive.metastore.events.AlterTableEvent;
import org.apache.hadoop.hive.metastore.events.CreateDatabaseEvent;
import org.apache.hadoop.hive.metastore.events.CreateTableEvent;
import org.apache.hadoop.hive.metastore.events.DropDatabaseEvent;
import org.apache.hadoop.hive.metastore.events.DropPartitionEvent;
import org.apache.hadoop.hive.metastore.events.DropTableEvent;
import org.apache.hadoop.hive.metastore.events.ListenerEvent;
import org.apache.hadoop.hive.metastore.events.LoadPartitionDoneEvent;
import org.apache.hcatalog.common.HCatConstants;
import org.apache.hcatalog.messaging.HCatEventMessage;
import org.apache.hcatalog.messaging.MessageFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Implementation of
 * {@link org.apache.hadoop.hive.metastore.MetaStoreEventListener} It sends
 * message on two type of topics. One has name of form dbName.tblName On this
 * topic, two kind of messages are sent: add/drop partition and
 * finalize_partition message. Second topic has name "HCAT" and messages sent on
 * it are: add/drop database and add/drop table. All messages also has a
 * property named "HCAT_EVENT" set on them whose value can be used to configure
 * message selector on subscriber side.
 */
public class NotificationListener extends MetaStoreEventListener {

    private static final Logger LOG = LoggerFactory.getLogger(NotificationListener.class);
    protected Session session;
    protected Connection conn;
    private static MessageFactory messageFactory = MessageFactory.getInstance();

    /**
     * Create message bus connection and session in constructor.
     */
    public NotificationListener(final Configuration conf) {

        super(conf);
        createConnection();
    }

    private static String getTopicName(Partition partition,
                                       ListenerEvent partitionEvent) throws MetaException {
        try {
            return partitionEvent.getHandler()
                .get_table(partition.getDbName(), partition.getTableName())
                .getParameters().get(HCatConstants.HCAT_MSGBUS_TOPIC_NAME);
        } catch (NoSuchObjectException e) {
            throw new MetaException(e.toString());
        }
    }

    @Override
    public void onAddPartition(AddPartitionEvent partitionEvent)
        throws MetaException {
        // Subscriber can get notification of newly add partition in a
        // particular table by listening on a topic named "dbName.tableName"
        // and message selector string as "HCAT_EVENT = HCAT_ADD_PARTITION"
        if (partitionEvent.getStatus()) {

            Partition partition = partitionEvent.getPartition();
            String topicName = getTopicName(partition, partitionEvent);
            if (topicName != null && !topicName.equals("")) {
                send(messageFactory.buildAddPartitionMessage(partitionEvent.getTable(), partition), topicName);
            } else {
                LOG.info("Topic name not found in metastore. Suppressing HCatalog notification for "
                    + partition.getDbName()
                    + "."
                    + partition.getTableName()
                    + " To enable notifications for this table, please do alter table set properties ("
                    + HCatConstants.HCAT_MSGBUS_TOPIC_NAME
                    + "=<dbname>.<tablename>) or whatever you want topic name to be.");
            }
        }

    }

    /**
     * Send dropped partition notifications. Subscribers can receive these notifications for a
     * particular table by listening on a topic named "dbName.tableName" with message selector
     * string {@value org.apache.hcatalog.common.HCatConstants#HCAT_EVENT} =
     * {@value org.apache.hcatalog.common.HCatConstants#HCAT_DROP_PARTITION_EVENT}.
     * </br>
     * TODO: DataNucleus 2.0.3, currently used by the HiveMetaStore for persistence, has been
     * found to throw NPE when serializing objects that contain null. For this reason we override
     * some fields in the StorageDescriptor of this notification. This should be fixed after
     * HIVE-2084 "Upgrade datanucleus from 2.0.3 to 3.0.1" is resolved.
     */
    @Override
    public void onDropPartition(DropPartitionEvent partitionEvent) throws MetaException {
        if (partitionEvent.getStatus()) {
            Partition partition = partitionEvent.getPartition();
            StorageDescriptor sd = partition.getSd();
            sd.setBucketCols(new ArrayList<String>());
            sd.setSortCols(new ArrayList<Order>());
            sd.setParameters(new HashMap<String, String>());
            sd.getSerdeInfo().setParameters(new HashMap<String, String>());
            sd.getSkewedInfo().setSkewedColNames(new ArrayList<String>());
            String topicName = getTopicName(partition, partitionEvent);
            if (topicName != null && !topicName.equals("")) {
                send(messageFactory.buildDropPartitionMessage(partitionEvent.getTable(), partition), topicName);
            } else {
                LOG.info("Topic name not found in metastore. Suppressing HCatalog notification for "
                    + partition.getDbName()
                    + "."
                    + partition.getTableName()
                    + " To enable notifications for this table, please do alter table set properties ("
                    + HCatConstants.HCAT_MSGBUS_TOPIC_NAME
                    + "=<dbname>.<tablename>) or whatever you want topic name to be.");
            }
        }
    }

    @Override
    public void onCreateDatabase(CreateDatabaseEvent dbEvent)
        throws MetaException {
        // Subscriber can get notification about addition of a database in HCAT
        // by listening on a topic named "HCAT" and message selector string
        // as "HCAT_EVENT = HCAT_ADD_DATABASE"
        if (dbEvent.getStatus()) {
            String topicName = getTopicPrefix(dbEvent.getHandler().getHiveConf());
            send(messageFactory.buildCreateDatabaseMessage(dbEvent.getDatabase()), topicName);
        }
    }

    @Override
    public void onDropDatabase(DropDatabaseEvent dbEvent) throws MetaException {
        // Subscriber can get notification about drop of a database in HCAT
        // by listening on a topic named "HCAT" and message selector string
        // as "HCAT_EVENT = HCAT_DROP_DATABASE"
        if (dbEvent.getStatus())  {
            String topicName = getTopicPrefix(dbEvent.getHandler().getHiveConf());
            send(messageFactory.buildDropDatabaseMessage(dbEvent.getDatabase()), topicName);
        }
    }

    @Override
    public void onCreateTable(CreateTableEvent tableEvent) throws MetaException {
        // Subscriber can get notification about addition of a table in HCAT
        // by listening on a topic named "HCAT" and message selector string
        // as "HCAT_EVENT = HCAT_ADD_TABLE"
        if (tableEvent.getStatus()) {
            Table tbl = tableEvent.getTable();
            HMSHandler handler = tableEvent.getHandler();
            HiveConf conf = handler.getHiveConf();
            Table newTbl;
            try {
                newTbl = handler.get_table(tbl.getDbName(), tbl.getTableName())
                    .deepCopy();
                newTbl.getParameters().put(
                    HCatConstants.HCAT_MSGBUS_TOPIC_NAME,
                    getTopicPrefix(conf) + "." + newTbl.getDbName().toLowerCase() + "."
                        + newTbl.getTableName().toLowerCase());
                handler.alter_table(newTbl.getDbName(), newTbl.getTableName(), newTbl);
            } catch (InvalidOperationException e) {
                MetaException me = new MetaException(e.toString());
                me.initCause(e);
                throw me;
            } catch (NoSuchObjectException e) {
                MetaException me = new MetaException(e.toString());
                me.initCause(e);
                throw me;
            }
            String topicName = getTopicPrefix(conf) + "." + newTbl.getDbName().toLowerCase();
            send(messageFactory.buildCreateTableMessage(newTbl), topicName);
        }
    }

    private String getTopicPrefix(HiveConf conf) {
        return conf.get(HCatConstants.HCAT_MSGBUS_TOPIC_PREFIX,
            HCatConstants.HCAT_DEFAULT_TOPIC_PREFIX);
    }

    /**
     * Send dropped table notifications. Subscribers can receive these notifications for
     * dropped tables by listening on topic "HCAT" with message selector string
     * {@value org.apache.hcatalog.common.HCatConstants#HCAT_EVENT} =
     * {@value org.apache.hcatalog.common.HCatConstants#HCAT_DROP_TABLE_EVENT}
     * </br>
     * TODO: DataNucleus 2.0.3, currently used by the HiveMetaStore for persistence, has been
     * found to throw NPE when serializing objects that contain null. For this reason we override
     * some fields in the StorageDescriptor of this notification. This should be fixed after
     * HIVE-2084 "Upgrade datanucleus from 2.0.3 to 3.0.1" is resolved.
     */
    @Override
    public void onDropTable(DropTableEvent tableEvent) throws MetaException {
        // Subscriber can get notification about drop of a table in HCAT
        // by listening on a topic named "HCAT" and message selector string
        // as "HCAT_EVENT = HCAT_DROP_TABLE"

        // Datanucleus throws NPE when we try to serialize a table object
        // retrieved from metastore. To workaround that we reset following objects

        if (tableEvent.getStatus()) {
            Table table = tableEvent.getTable();
            String topicName = getTopicPrefix(tableEvent.getHandler().getHiveConf()) + "." + table.getDbName().toLowerCase();
            send(messageFactory.buildDropTableMessage(table), topicName);
        }
    }

    /**
     * @param hCatEventMessage The HCatEventMessage being sent over JMS.
     * @param topicName is the name on message broker on which message is sent.
     */
    protected void send(HCatEventMessage hCatEventMessage, String topicName) {
        try {
            if(null == session){
                // this will happen, if we never able to establish a connection.
                createConnection();
                if (null == session){
                    // Still not successful, return from here.
                    LOG.error("Invalid session. Failed to send message on topic: " +
                            topicName + " event: " + hCatEventMessage.getEventType());
                    return;
                }
            }

            Destination topic = getTopic(topicName);

            if (null == topic){
                // Still not successful, return from here.
                LOG.error("Invalid session. Failed to send message on topic: " +
                        topicName + " event: " + hCatEventMessage.getEventType());
                return;
            }

            MessageProducer producer = session.createProducer(topic);
            Message msg = session.createTextMessage(hCatEventMessage.toString());

            msg.setStringProperty(HCatConstants.HCAT_EVENT, hCatEventMessage.getEventType().toString());
            msg.setStringProperty(HCatConstants.HCAT_MESSAGE_VERSION, messageFactory.getVersion());
            msg.setStringProperty(HCatConstants.HCAT_MESSAGE_FORMAT, messageFactory.getMessageFormat());
            producer.send(msg);
            // Message must be transacted before we return.
            session.commit();
        }
        catch(Exception e){
            // Gobble up the exception. Message delivery is best effort.
            LOG.error("Failed to send message on topic: " + topicName +
                    " event: " + hCatEventMessage.getEventType(), e);
        }
    }

    /**
     * Get the topic object for the topicName, it also tries to reconnect
     * if the connection appears to be broken.
     *
     * @param topicName The String identifying the message-topic.
     * @return A {@link Topic} object corresponding to the specified topicName.
     * @throws JMSException
     */
    protected Topic getTopic(final String topicName) throws JMSException {
        Topic topic;
        try {
            // Topics are created on demand. If it doesn't exist on broker it will
            // be created when broker receives this message.
            topic = session.createTopic(topicName);
        } catch (IllegalStateException ise) {
            // this will happen if we were able to establish connection once, but its no longer valid,
            // ise is thrown, catch it and retry.
            LOG.error("Seems like connection is lost. Retrying", ise);
            createConnection();
            topic = session.createTopic(topicName);
        }
        return topic;
    }

    protected void createConnection() {

        Context jndiCntxt;
        try {
            jndiCntxt = new InitialContext();
            ConnectionFactory connFac = (ConnectionFactory) jndiCntxt
                .lookup("ConnectionFactory");
            Connection conn = connFac.createConnection();
            conn.start();
            conn.setExceptionListener(new ExceptionListener() {
                @Override
                public void onException(JMSException jmse) {
                    LOG.error(jmse.toString());
                }
            });
            // We want message to be sent when session commits, thus we run in
            // transacted mode.
            session = conn.createSession(true, Session.SESSION_TRANSACTED);
        } catch (NamingException e) {
            LOG.error("JNDI error while setting up Message Bus connection. "
                + "Please make sure file named 'jndi.properties' is in "
                + "classpath and contains appropriate key-value pairs.", e);
        } catch (JMSException e) {
            LOG.error("Failed to initialize connection to message bus", e);
        } catch (Throwable t) {
            LOG.error("Unable to connect to JMS provider", t);
        }
    }

    @Override
    protected void finalize() throws Throwable {
        // Close the connection before dying.
        try {
            if (null != session)
                session.close();
            if (conn != null) {
                conn.close();
            }

        } catch (Exception ignore) {
            LOG.info("Failed to close message bus connection.", ignore);
        }
    }

    @Override
    public void onLoadPartitionDone(LoadPartitionDoneEvent lpde)
        throws MetaException {
//  TODO: Fix LoadPartitionDoneEvent. Currently, LPDE can only carry a single partition-spec. And that defeats the purpose.
//		if(lpde.getStatus())
//			send(lpde.getPartitionName(),lpde.getTable().getParameters().get(HCatConstants.HCAT_MSGBUS_TOPIC_NAME),HCatConstants.HCAT_PARTITION_DONE_EVENT);
    }

    @Override
    public void onAlterPartition(AlterPartitionEvent ape) throws MetaException {
        // no-op
    }

    @Override
    public void onAlterTable(AlterTableEvent ate) throws MetaException {
        // no-op
    }
}