/*
    * 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 org.apache.log4j.Level;
  import org.apache.log4j.Logger;
  import org.apache.log4j.scheduler.Job;
  import org.apache.log4j.spi.ComponentBase;
  import org.apache.log4j.spi.LocationInfo;
  import org.apache.log4j.spi.LoggingEvent;
  import org.apache.log4j.spi.ThrowableInformation;
  
  import java.sql.Connection;
  import java.sql.PreparedStatement;
  import java.sql.ResultSet;
  import java.sql.SQLException;
  import java.util.Hashtable;
  import java.util.Vector;
  
  /**
   * Actual retrieval of data is made by the instance of DBReceiverJob associated
   * with DBReceiver.
   * 
   * @author Ceki Gülcü
   */
  class DBReceiverJob extends ComponentBase implements Job {
  
    String sqlException = "SELECT trace_line FROM logging_event_exception where event_id=? ORDER by i ASC";
    String sqlProperties = "SELECT mapped_key, mapped_value FROM logging_event_property WHERE event_id=?";
    String sqlSelect = 
      "SELECT " +
      "sequence_number, timestamp, rendered_message, logger_name, " +
      "level_string, ndc, thread_name, reference_flag, " +
      "caller_filename, caller_class, caller_method, caller_line, " +
      "event_id " +
      "FROM logging_event " +
      "WHERE event_id > ?  ORDER BY event_id ASC";
  
  
    long lastId = Short.MIN_VALUE;
  
    DBReceiver parentDBReceiver;
  
    DBReceiverJob(DBReceiver parent) {
      parentDBReceiver = parent;
    }
  
    public void execute() {
      getLogger().debug("DBReceiverJob.execute() called");
  
      Connection connection = null;
  
      try {
        connection = parentDBReceiver.connectionSource.getConnection();
        PreparedStatement statement = connection.prepareStatement(sqlSelect);
        statement.setLong(1, lastId);
        ResultSet rs = statement.executeQuery();
        //rs.beforeFirst();
  
        while (rs.next()) {
  	    Logger logger = null;
  	    long timeStamp = 0L;
  	    String level = null;
  	    String threadName = null;
  	    Object message = null;
  	    String ndc = null;
  	    String className = null;
  	    String methodName = null;
  	    String fileName = null;
  	    String lineNumber = null;
  	    Hashtable properties = new Hashtable();
  	
  
          //event.setSequenceNumber(rs.getLong(1));
          timeStamp = rs.getLong(2);
          message = rs.getString(3);
  		logger = Logger.getLogger(rs.getString(4));
          level = rs.getString(5);
  		Level levelImpl = Level.toLevel(level.trim());
  
          ndc = rs.getString(6);
          threadName = rs.getString(7);
  
          short mask = rs.getShort(8);
 
         fileName = rs.getString(9);
         className = rs.getString(10);
         methodName = rs.getString(11);
         lineNumber = rs.getString(12).trim();
 
 		LocationInfo locationInfo = null;
         if (fileName.equals(LocationInfo.NA)) {
           locationInfo = LocationInfo.NA_LOCATION_INFO;
         } else {
           locationInfo = new LocationInfo(fileName, className,
               methodName, lineNumber);
         }
 
         long id = rs.getLong(13);
         //LogLog.info("Received event with id=" + id);
         lastId = id;
 
 		ThrowableInformation throwableInfo = null;
         if ((mask & DBHelper.EXCEPTION_EXISTS) != 0) {
           throwableInfo = getException(connection, id);
         }
 
 	    LoggingEvent event = new LoggingEvent(logger.getName(),
 	            logger, timeStamp, levelImpl, message,
 	            threadName,
 	            throwableInfo,
 	            ndc,
 	            locationInfo,
 	            properties);
 
 
         // Scott asked for this info to be
         event.setProperty("log4jid", Long.toString(id));
 
         if ((mask & DBHelper.PROPERTIES_EXIST) != 0) {
           getProperties(connection, id, event);
         }
 
 
 
 
         if (!parentDBReceiver.isPaused()) {
           parentDBReceiver.doPost(event);
         }
       } // while
       statement.close();
       statement = null;
     } catch (SQLException sqle) {
       getLogger().error("Problem receiving events", sqle);
     } finally {
       closeConnection(connection);
     }
   }
 
   void closeConnection(Connection connection) {
     if (connection != null) {
       try {
         //LogLog.warn("closing the connection. ", new Exception("x"));
         connection.close();
       } catch (SQLException sqle) {
         // nothing we can do here
       }
     }
   }
 
   /**
    * Retrieve the event properties from the logging_event_property table.
    * 
    * @param connection
    * @param id
    * @param event
    * @throws SQLException
    */
   void getProperties(Connection connection, long id, LoggingEvent event)
       throws SQLException {
 
     PreparedStatement statement = connection.prepareStatement(sqlProperties);
     try {
       statement.setLong(1, id);
       ResultSet rs = statement.executeQuery();
 
       while (rs.next()) {
         String key = rs.getString(1);
         String value = rs.getString(2);
         event.setProperty(key, value);
       }
     } finally {
       statement.close();
     }
   }
 
   /**
    * Retrieve the exception string representation from the
    * logging_event_exception table.
    * 
    * @param connection
    * @param id
    * @throws SQLException
    */
   ThrowableInformation getException(Connection connection, long id)
       throws SQLException {
 
     PreparedStatement statement = null;
 
     try {
       statement = connection.prepareStatement(sqlException);
       statement.setLong(1, id);
       ResultSet rs = statement.executeQuery();
 
       Vector v = new Vector();
 
       while (rs.next()) {
         //int i = rs.getShort(1);
         v.add(rs.getString(1));
       }
 
       int len = v.size();
       String[] strRep = new String[len];
       for (int i = 0; i < len; i++) {
         strRep[i] = (String) v.get(i);
       }
       // we've filled strRep, we now attach it to the event
       return new ThrowableInformation(strRep);
     } finally {
       if (statement != null) {
         statement.close();
       }
     }
   }
 }