/*
 * 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.log4j.db;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Hashtable;
import java.util.Properties;
import java.util.StringTokenizer;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.plugins.Pauseable;
import org.apache.log4j.plugins.Receiver;
import org.apache.log4j.scheduler.Job;
import org.apache.log4j.scheduler.Scheduler;
import org.apache.log4j.spi.LocationInfo;
import org.apache.log4j.spi.LoggerRepositoryEx;
import org.apache.log4j.spi.LoggingEvent;
import org.apache.log4j.spi.ThrowableInformation;
import org.apache.log4j.xml.DOMConfigurator;
import org.apache.log4j.xml.UnrecognizedElementHandler;
import org.w3c.dom.Element;

/**
 * Converts log data stored in a database into LoggingEvents.
 * <p>
 * <b>NOTE:</b> This receiver cannot yet be created through Chainsaw's receiver panel.  
 * It must be created through an XML configuration file.
 * <p>
 * This receiver supports database configuration via ConnectionSource, in the
 * org.apache.log4j.db package: DriverManagerConnectionSource,
 * DataSourceConnectionSource, JNDIConnectionSource
 * <p>
 * This database receiver differs from DBReceiver in that this receiver relies
 * on custom SQL to retrieve logging event data, where DBReceiver requires the
 * use of a log4j-defined schema.
 * <p>
 * A 'refreshMillis' int parameter controls SQL execution. If 'refreshMillis' is
 * zero (the default), the receiver will run only one time. If it is set to any
 * other numeric value, the SQL will be executed on a recurring basis every
 * 'refreshMillis' milliseconds.
 * <p>
 * The receiver closes the connection and acquires a new connection on each 
 * execution of the SQL (use pooled connections if possible).
 * <p>
 * If the SQL will be executing on a recurring basis, specify the IDField param -
 * the column name holding the unique identifier (int) representing the logging
 * event.
 * <p>
 * As events are retrieved, the column represented by IDField is examined and
 * the largest value is held and used by the next execution of the SQL statement
 * to avoid retrieving previously processed events.
 * <p>
 * As an example, the IDField references a 'COUNTER' (int, auto-increment,
 * unique) column. The first execution of the SQL statement returns 500 rows,
 * with a final value in the COUNTER field of 500.
 * <p>
 * The SQL statement is manipulated prior to the next execution, adding ' WHERE
 * COUNTER > 500' to the statement to avoid retrieval of previously processed
 * events.
 * <p>
 * The select statement must provide ALL fields which define a LoggingEvent.
 * <p>
 * The SQL statement MUST include the columns: LOGGER, TIMESTAMP, LEVEL, THREAD,
 * MESSAGE, NDC, MDC, CLASS, METHOD, FILE, LINE, PROPERTIES, THROWABLE
 * <p>
 * Use ' AS ' in the SQL statement to alias the SQL's column names to match your
 * database schema. (see example below).
 * <p>
 * Include all fields in the SQL statement, even if you don't have data for the
 * field (specify an empty string as the value for columns which you don't have
 * data).
 * <p>
 * The TIMESTAMP column must be a datetime.
 * <p>
 * Both a PROPERTIES column and an MDC column are supported. These fields
 * represent Maps on the logging event, but require the use of string
 * concatenation database functions to hold the (possibly multiple) name/value
 * pairs in the column.
 * <p>
 * For example, to include both 'userid' and 'lastname' properties in the
 * logging event (from either the PROPERTIES or MDC columns), the name/value
 * pairs must be concatenated together by your database.
 * <p>
 * The resulting PROPERTIES or MDC column must have data in this format: {{name,
 * value, name2, value2}}
 * <p>
 * The resulting PROPERTIES column would contain this text: {{userid, someone,
 * lastname, mylastname}}
 * <p>
 * Here is an example of concatenating a PROPERTIES or MDC column using MySQL's
 * concat function, where the 'application' and 'hostname' parameters were fixed
 * text, but the 'log4jid' key's value is the value of the COUNTER column:
 * <p>
 * concat("{{application,databaselogs,hostname,mymachine,log4jid,", COUNTER,
 * "}}") as PROPERTIES
 * <p>
 * log4jid is a special property that is used by Chainsaw to represent an 'ID'
 * field. Specify this property to ensure you can map events in Chainsaw to
 * events in the database if you need to go back and view events at a later time
 * or save the events to XML for later analysis.
 * <p>
 * Here is a complete MySQL SQL statement which can be used to provide events to
 * Chainsaw (note how in the example below, there is no column in logtable representing the throwable, so an
 * empty string is passed in and an ALIAS is still defined):
 * <p>
 * select myloggercolumn as LOGGER, mytimestampcolumn as TIMESTAMP, mylevelcolumn as LEVEL, mythreadcolumn as
 * THREAD, mymessagecolumn as MESSAGE, myndccolumn as NDC, mymdccolumn as MDC, myclasscolumn as CLASS, mymethodcolumn as
 * METHOD, myfilecolumn as FILE, mylinecolumn as LINE,
 * concat("{{application,databaselogs,hostname,mymachine, log4jid,",
 * COUNTER,"}}") as PROPERTIES, "" as THROWABLE from logtable
 * <p>
 * @author Scott Deboy <sdeboy@apache.org>
 * <p>
 */
public class CustomSQLDBReceiver extends Receiver implements Pauseable, UnrecognizedElementHandler {

    protected volatile Connection connection = null;

    protected String sqlStatement = "";

    /**
     * By default we refresh data every 1000 milliseconds.
     * 
     * @see #setRefreshMillis
     */
    static int DEFAULT_REFRESH_MILLIS = 1000;

    int refreshMillis = DEFAULT_REFRESH_MILLIS;

    protected String idField = null;

    int lastID = -1;

    private static final String WHERE_CLAUSE = " WHERE ";

    private static final String AND_CLAUSE = " AND ";

    private boolean whereExists = false;

    private boolean paused = false;

    private ConnectionSource connectionSource;

    public static final String LOG4J_ID_KEY = "log4jid";

    private Job customReceiverJob;

    public void activateOptions() {
      
      if(connectionSource == null)  {
        throw new IllegalStateException(
          "CustomSQLDBReceiver cannot function without a connection source");
      }
      whereExists = (sqlStatement.toUpperCase().indexOf(WHERE_CLAUSE) > -1);
    
      customReceiverJob = new CustomReceiverJob();
        
      if(this.repository == null) {
        throw new IllegalStateException(
        "CustomSQLDBReceiver cannot function without a reference to its owning repository");
      }
     
    
      if (repository instanceof LoggerRepositoryEx) {
        Scheduler scheduler = ((LoggerRepositoryEx) repository).getScheduler();
      
        scheduler.schedule(
          customReceiverJob, System.currentTimeMillis() + 500, refreshMillis);
      }

    }

    void closeConnection() {
        if (connection != null) {
            try {
                // LogLog.warn("closing the connection. ", new Exception("x"));
                connection.close();
            } catch (SQLException sqle) {
                // nothing we can do here
            }
        }
    }

    public void setRefreshMillis(int refreshMillis) {
        this.refreshMillis = refreshMillis;
    }

    public int getRefreshMillis() {
        return refreshMillis;
    }

    /**
     * @return Returns the connectionSource.
     */
    public ConnectionSource getConnectionSource() {
        return connectionSource;
    }

    /**
     * @param connectionSource
     *            The connectionSource to set.
     */
    public void setConnectionSource(ConnectionSource connectionSource) {
        this.connectionSource = connectionSource;
    }

    public void close() {
        try {
            if ((connection != null) && !connection.isClosed()) {
                connection.close();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            connection = null;
        }
    }

    public void finalize() throws Throwable {
        super.finalize();
        close();
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.apache.log4j.plugins.Plugin#shutdown()
     */
    public void shutdown() {
        getLogger().info("removing receiverJob from the Scheduler.");

        if(this.repository instanceof LoggerRepositoryEx) {
          Scheduler scheduler = ((LoggerRepositoryEx) repository).getScheduler();
          scheduler.delete(customReceiverJob);
        }

        lastID = -1;
    }

    public void setSql(String s) {
        sqlStatement = s;
    }

    public String getSql() {
        return sqlStatement;
    }

    public void setIDField(String id) {
        idField = id;
    }

    public String getIDField() {
        return idField;
    }

    public synchronized void setPaused(boolean p) {
        paused = p;
    }

    public synchronized boolean isPaused() {
        return paused;
    }

    class CustomReceiverJob implements Job {
        public void execute() {
            int oldLastID = lastID;
            try {
                connection = connectionSource.getConnection();
                Statement statement = connection.createStatement();

                Logger eventLogger = null;
                long timeStamp = 0L;
                String level = null;
                String threadName = null;
                Object message = null;
                String ndc = null;
                Hashtable mdc = null;
                String[] throwable = null;
                String className = null;
                String methodName = null;
                String fileName = null;
                String lineNumber = null;
                Hashtable properties = null;

                String currentSQLStatement = sqlStatement;
                if (whereExists) {
                    currentSQLStatement = sqlStatement + AND_CLAUSE + idField
                            + " > " + lastID;
                } else {
                    currentSQLStatement = sqlStatement + WHERE_CLAUSE + idField
                            + " > " + lastID;
                }

                ResultSet rs = statement.executeQuery(currentSQLStatement);

                int i = 0;
                while (rs.next()) {
                    // add a small break every 1000 received events
                    if (++i == 1000) {
                        synchronized (this) {
                            try {
                                // add a delay
                                wait(300);
                            } catch (InterruptedException ie) {
                            }
                            i = 0;
                        }
                    }
                    eventLogger = Logger.getLogger(rs.getString("LOGGER"));
                    timeStamp = rs.getTimestamp("TIMESTAMP").getTime();

                    level = rs.getString("LEVEL");
                    threadName = rs.getString("THREAD");
                    message = rs.getString("MESSAGE");
                    ndc = rs.getString("NDC");

                    String mdcString = rs.getString("MDC");
                    mdc = new Hashtable();

                    if (mdcString != null) {
                        // support MDC being wrapped in {{name, value}}
                        // or
                        // just name, value
                        if ((mdcString.indexOf("{{") > -1)
                                && (mdcString.indexOf("}}") > -1)) {
                            mdcString = mdcString
                                    .substring(mdcString.indexOf("{{") + 2,
                                            mdcString.indexOf("}}"));
                        }

                        StringTokenizer tok = new StringTokenizer(mdcString,
                                ",");

                        while (tok.countTokens() > 1) {
                            mdc.put(tok.nextToken(), tok.nextToken());
                        }
                    }

                    throwable = new String[] { rs.getString("THROWABLE") };
                    className = rs.getString("CLASS");
                    methodName = rs.getString("METHOD");
                    fileName = rs.getString("FILE");
                    lineNumber = rs.getString("LINE");

                    // if properties are provided in the
                    // SQL they can be used here (for example, to route
                    // events to a unique tab in
                    // Chainsaw if the machinename and/or appname
                    // property
                    // are set)
                    String propertiesString = rs.getString("PROPERTIES");
                    properties = new Hashtable();

                    if (propertiesString != null) {
                        // support properties being wrapped in {{name,
                        // value}} or just name, value
                        if ((propertiesString.indexOf("{{") > -1)
                                && (propertiesString.indexOf("}}") > -1)) {
                            propertiesString = propertiesString.substring(
                                    propertiesString.indexOf("{{") + 2,
                                    propertiesString.indexOf("}}"));
                        }

                        StringTokenizer tok2 = new StringTokenizer(
                                propertiesString, ",");
                        while (tok2.countTokens() > 1) {
                            String tokenName = tok2.nextToken();
                            String value = tok2.nextToken();
                            if (tokenName.equals(LOG4J_ID_KEY)) {
                                try {
                                    int thisInt = Integer.parseInt(value);
                                    value = String.valueOf(thisInt);
                                    if (thisInt > lastID) {
                                        lastID = thisInt;
                                    }
                                } catch (Exception e) {
                                }
                            }
                            properties.put(tokenName, value);
                        }
                    }

                    Level levelImpl = Level.toLevel(level);


					LocationInfo locationInfo = new LocationInfo(fileName,
	                            className, methodName, lineNumber);
	
					ThrowableInformation throwableInfo =  new ThrowableInformation(
		                            throwable);
	
					properties.putAll(mdc);
		
				    LoggingEvent event = new LoggingEvent(eventLogger.getName(),
				            eventLogger, timeStamp, levelImpl, message,
				            threadName,
				            throwableInfo,
				            ndc,
				            locationInfo,
				            properties);

                    doPost(event);
                }
                //log when rows are retrieved
                if (lastID != oldLastID) {
                    getLogger().debug("lastID: " + lastID);
                    oldLastID = lastID;
                }

                statement.close();
                statement = null;
            } catch (SQLException sqle) {
                getLogger()
                        .error("*************Problem receiving events", sqle);
            } finally {
                closeConnection();
            }

            // if paused, loop prior to executing sql query
            synchronized (this) {
                while (isPaused()) {
                    try {
                        wait(1000);
                    } catch (InterruptedException ie) {
                    }
                }
            }
        }
    }

    /**
     * @{inheritDoc}
     */
  public boolean parseUnrecognizedElement(Element element, Properties props) throws Exception {
        if ("connectionSource".equals(element.getNodeName())) {
            Object instance =
                    DOMConfigurator.parseElement(element, props, ConnectionSource.class);
            if (instance instanceof ConnectionSource) {
               ConnectionSource source = (ConnectionSource) instance;
               source.activateOptions();
               setConnectionSource(source);
            }
            return true;
        }
        return false;
  }
    
}