JDBCLoggerThread.java

/**
 * C-JDBC: Clustered JDBC.
 * Copyright (C) 2002-2005 French National Institute For Research In Computer
 * Science And Control (INRIA).
 * Contact: c-jdbc@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published by the
 * Free Software Foundation; either version 2.1 of the License, or any later
 * version.
 * 
 * This library is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation,
 * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
 *
 * Initial developer(s): Emmanuel Cecchet.
 * Contributor(s): ______________________.
 */

package org.objectweb.cjdbc.controller.recoverylog;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;

import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.log.Trace;

/**
 * Logger thread for the JDBCRecoveryLog.
 * 
 * @author <a href="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
 * @version 1.0
 */
public class JDBCLoggerThread extends Thread
{
  private boolean           killed = false; // Control thread death
  private ArrayList         logQueue;
  private PreparedStatement pstmt;
  private Trace             logger;

  private class LogObject
  {
    private long    tid;
    private String  query;
    private String  login;
    private long    id;
    private boolean escapeProcessing;

    /**
     * Create a log object.
     * 
     * @param id unique request id
     * @param login login used for this request
     * @param query query to log
     * @param tid transaction id of this request
     * @param escapeProcessing true if escape processing must be done
     */
    public LogObject(long id, String login, String query, long tid,
        boolean escapeProcessing)
    {
      this.id = id;
      this.login = login;
      this.query = query;
      this.tid = tid;
      this.escapeProcessing = escapeProcessing;
    }

    /**
     * @return the request id
     */
    public long getId()
    {
      return id;
    }

    /**
     * @return the login used for this request
     */
    public String getLogin()
    {
      return login;
    }

    /**
     * @return the request itself
     */
    public String getQuery()
    {
      return query;
    }

    /**
     * @return the transaction id
     */
    public long getTid()
    {
      return tid;
    }

    /**
     * @return true if escape processing is needed
     */
    public boolean getEscapeProcessing()
    {
      return escapeProcessing;
    }
  }

  /**
   * Create a new JDBCLoggerThread
   * 
   * @param pstmt the PreparedStatement to use to log the requests
   * @param logger where to log error messages
   */
  public JDBCLoggerThread(PreparedStatement pstmt, Trace logger)
  {
    super("JDBCLoggerThread");
    this.pstmt = pstmt;
    this.logger = logger;
    logQueue = new ArrayList();
  }

  /**
   * Log a query.
   * 
   * @param id unique request id
   * @param login login used for this request
   * @param query query to log
   * @param tid transaction id of this request
   * @param escapeProcessing true if escape processing must be done
   */
  public synchronized void log(long id, String login, String query, long tid,
      boolean escapeProcessing)
  {
    logQueue.add(new LogObject(id, login, query, tid, escapeProcessing));
    notify();
  }

  /**
   * Rollback all unfinished transactions.
   */
  public synchronized void rollbackTransactions()
  {
    if (logQueue.isEmpty())
    {
      if (logger.isDebugEnabled())
        logger
            .debug(Translate.get("recovery.jdbc.loggerthread.no.transaction"));
      return;
    }
    // Find all begun transactions
    LogObject log;
    for (int i = 0; i < logQueue.size(); i++)
    {
      log = (LogObject) logQueue.get(i);
      if (log.getQuery().equalsIgnoreCase("begin"))
      {
        // Found beginning of transaction, let's roll it back
        removeQueriesOfTransactionFromQueue(log.getTid());
      }
    }
  }

  /**
   * Remove all queries that have not been logged yet and belonging to the
   * specified transaction.
   * 
   * @param tid transaction id to rollback
   */
  public synchronized void removeQueriesOfTransactionFromQueue(long tid)
  {
    if (logger.isDebugEnabled())
      logger.debug(Translate.get("recovery.jdbc.loggerthread.removing", tid));
    LogObject logEntry;
    for (int i = 0; i < logQueue.size(); i++)
    {
      logEntry = (LogObject) logQueue.get(i);
      if (logEntry.getTid() == tid)
        logQueue.remove(i);
    }
  }

  /**
   * Tells whether there are pending logs
   * 
   * @return true if no more jobs in the log queue
   */
  public synchronized boolean getLogQueueIsEmpty()
  {
    if (logQueue.isEmpty())
    {
      // Notifies the Recovery log that the queue is empty.
      notify();
      return true;
    }
    else
    {
      return false;
    }
  }

  /**
   * Log the requests from queue until the thread is explicetly killed. The
   * logger used is the one of the JDBCRecoveryLog.
   */
  public void run()
  {
    LogObject params;

    while (!killed)
    {
      synchronized (this)
      {
        while (getLogQueueIsEmpty())
        {
          try
          {
            wait();
          }
          catch (InterruptedException e)
          {
            logger.warn(Translate.get("recovery.jdbc.loggerthread.awaken"), e);
          }
        }
        // Pump first log entry from the queue
        params = (LogObject) logQueue.remove(0);
        try
        {
          pstmt.setLong(1, params.getId());
          pstmt.setString(2, params.getLogin());
          pstmt.setString(3, params.getQuery());
          pstmt.setLong(4, params.getTid());
          try
          {
            pstmt.setEscapeProcessing(params.getEscapeProcessing());
          }
          catch (Exception ignore)
          {
          }
          pstmt.executeUpdate();
          if (logger.isDebugEnabled())
            logger.debug(Translate.get("recovery.jdbc.loggerthread.log.debug",
                new String[]{String.valueOf(params.getId()), params.getLogin(),
                    params.getQuery(), String.valueOf(params.getTid()),
                    String.valueOf(params.getEscapeProcessing())}));
          else if (logger.isInfoEnabled())
            logger.info(Translate.get("recovery.jdbc.loggerthread.log.info",
                params.getId()));
        }
        catch (SQLException e)
        {
          logger.error(
              Translate.get("recovery.jdbc.loggerthread.log.failed",
                  new String[]{params.getQuery(),
                      String.valueOf(params.getTid())}), e);
          // Push object back in the queue, it needs to be logged again
          logQueue.add(0, params);
        }
      }
    }
  }

}

---