VirtualDatabaseWorkerThread.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): Nicolas Modrzyk.
 */

package org.objectweb.cjdbc.controller.virtualdatabase;

import java.io.EOFException;
import java.io.IOException;
import java.io.OptionalDataException;
import java.net.SocketException;
import java.rmi.RemoteException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import org.objectweb.cjdbc.common.log.Trace;
import org.objectweb.cjdbc.common.sql.AbstractRequest;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.AlterRequest;
import org.objectweb.cjdbc.common.sql.CreateRequest;
import org.objectweb.cjdbc.common.sql.DeleteRequest;
import org.objectweb.cjdbc.common.sql.DropRequest;
import org.objectweb.cjdbc.common.sql.InsertRequest;
import org.objectweb.cjdbc.common.sql.ParsingGranularities;
import org.objectweb.cjdbc.common.sql.SelectRequest;
import org.objectweb.cjdbc.common.sql.StoredProcedure;
import org.objectweb.cjdbc.common.sql.UpdateRequest;
import org.objectweb.cjdbc.common.sql.metadata.MetadataContainer;
import org.objectweb.cjdbc.common.stream.CJDBCInputStream;
import org.objectweb.cjdbc.common.stream.CJDBCOutputStream;
import org.objectweb.cjdbc.common.users.VirtualDatabaseUser;
import org.objectweb.cjdbc.common.util.Constants;
import org.objectweb.cjdbc.controller.core.Controller;
import org.objectweb.cjdbc.driver.protocol.CommandCompleted;
import org.objectweb.cjdbc.driver.protocol.Commands;

/**
 * The <code>VirtualDatabaseWorkerThread</code> handles a connection with a
 * C-JDBC driver.
 * 
 * @author <a href="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
 * @author <a href="mailto:Nicolas.Modrzyk@inria.fr">Nicolas Modrzyk </a>
 * @version 1.0
 */
public class VirtualDatabaseWorkerThread extends Thread
{
  //
  // How the code is organized?
  //
  // 1. Member variables
  // 2. Constructor(s)
  // 3. Request management
  // 4. Getter/Setters

  /** <code>true</code> ifthis has been killed. */
  private boolean                isKilled     = false;

  /** Needed for parsing */
  boolean                        needSkeleton = false;

  /** Virtual database instantiating this thread. */
  private VirtualDatabase        vdb;

  private final CommandCompleted completed    = new CommandCompleted();

  /** Logger instance. */
  private Trace                  logger       = null;

  private CJDBCInputStream       in           = null;
  private CJDBCOutputStream      out          = null;

  private VirtualDatabaseUser    user;

  private Controller             controller;

  private boolean                waitForCommand;

  private HashMap                streamedResultSet;

  /**
   * The following variables represent the state of the connection with the
   * client
   */
  private long                   currentTid;
  private boolean                transactionStarted;
  private boolean                queryExecutedInThisTransaction;
  private String                 login;
  private boolean                closed;

  /*
   * Constructor
   */

  /**
   * Creates a new <code>VirtualDatabaseWorkerThread</code> instance.
   * 
   * @param controller the thread was originated from
   * @param vdb the virtual database instantiating this thread.
   */
  public VirtualDatabaseWorkerThread(Controller controller, VirtualDatabase vdb)
  {
    super("VirtualDatabaseWorkerThread-" + vdb.getVirtualDatabaseName());
    this.vdb = vdb;
    this.controller = controller;
    try
    {
      this.logger = Trace
          .getLogger("org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread."
              + vdb.getVirtualDatabaseName());
    }
    catch (Exception e)
    {
      this.logger = vdb.logger;
    }
  }

  //
  // Decoding commands from the stream
  //

  /**
   * Read a StoredProcedure from the stream.
   * 
   * @param isRead true if we expect a read stored procedure
   * @return the corresponding <code>StoredProcedure</code> object.
   * @throws OptionalDataException if an error occurs
   * @throws IOException if an error occurs
   */
  private StoredProcedure decodeProcedureFromStream(boolean isRead)
      throws OptionalDataException, IOException
  {
    String sql = in.readUTF();
    boolean escape = in.readBoolean();
    String lineSeparator = in.readUTF();
    int timeout = in.readInt();
    StoredProcedure proc = new StoredProcedure(sql, escape, timeout,
        lineSeparator);
    proc.setIsAutoCommit(in.readBoolean());
    proc.setDriverProcessed(in.readBoolean());
    if (isRead)
    {
      proc.setMaxRows(in.readInt());
      proc.setFetchSize(in.readInt());
    }
    // Does the query has a skeleton ?
    if (needSkeleton || !proc.isDriverProcessed())
      if (in.readBoolean()) // is there a non null skeleton
        proc.setSqlSkeleton(in.readUTF());
    return proc;
  }

  /**
   * Read a SelectRequest from the stream.
   * 
   * @return the corresponding <code>SelectRequest</code> object.
   * @throws OptionalDataException if an error occurs
   * @throws IOException if an error occurs
   */
  private SelectRequest decodeReadRequestFromStream()
      throws OptionalDataException, IOException
  {
    String sql = in.readUTF();
    boolean escape = in.readBoolean();
    String lineSeparator = in.readUTF();
    int timeout = in.readInt();
    SelectRequest select = new SelectRequest(sql, escape, timeout,
        lineSeparator);
    select.setIsAutoCommit(in.readBoolean());
    select.setDriverProcessed(in.readBoolean());
    select.setMaxRows(in.readInt());
    select.setFetchSize(in.readInt());
    if (in.readBoolean()) // do we have a cursor name ?
      select.setCursorName(in.readUTF());
    // Does the query has a skeleton ?
    if (needSkeleton || !select.isDriverProcessed())
      if (in.readBoolean()) // is there a non null skeleton
        select.setSqlSkeleton(in.readUTF());
    return select;
  }

  /**
   * Read a write request send by the <code>Connection</code> object. Only the
   * needed parameters are sent, so we can reconstruct the object on the
   * controller side, without serializing everything.
   * 
   * @param withKeys true if we expect a write to return keus (as a ResultSet)
   * @return an instace of the <code>AbstractWriteRequest</code>
   */
  private AbstractWriteRequest decodeWriteRequestFromStream(boolean withKeys)
      throws OptionalDataException, IOException
  {
    AbstractWriteRequest writeRequest;
    int requestType = in.readInt();
    String sql = in.readUTF();
    boolean escape = in.readBoolean();
    String lineSeparator = in.readUTF();
    int timeout = in.readInt();
    switch (requestType)
    {
      case Commands.CreateRequest :
        writeRequest = new CreateRequest(sql, escape, timeout, lineSeparator);
        break;
      case Commands.AlterRequest :
        writeRequest = new AlterRequest(sql, escape, timeout, lineSeparator);
        break;
      case Commands.DeleteRequest :
        writeRequest = new DeleteRequest(sql, escape, timeout, lineSeparator);
        break;
      case Commands.DropRequest :
        writeRequest = new DropRequest(sql, escape, timeout, lineSeparator);
        break;
      case Commands.InsertRequest :
        writeRequest = new InsertRequest(sql, escape, timeout, lineSeparator);
        break;
      case Commands.UpdateRequest :
        writeRequest = new UpdateRequest(sql, escape, timeout, lineSeparator);
        break;
      default :
        throw new IOException("Invalid Write Query Type");
    }

    writeRequest.setIsAutoCommit(in.readBoolean());
    writeRequest.setDriverProcessed(in.readBoolean());
    if (withKeys)
    {
      writeRequest.setMaxRows(in.readInt());
      writeRequest.setFetchSize(in.readInt());
    }

    // Does the query has a skeleton ?
    if (needSkeleton || !writeRequest.isDriverProcessed())
      if (in.readBoolean())
        writeRequest.setSqlSkeleton(in.readUTF());

    return writeRequest;
  }

  /**
   * Set the login and transaction id on the given request. If the request is
   * autocommit and a transaction was started, the transaction is first commited
   * to return in autocommit mode.
   * 
   * @param request The request to set
   * @param login user login to set
   * @param tid the transaction id to set
   * @return new value of transaction started
   */
  private boolean setRequestParameters(AbstractRequest request, String login,
      long tid, boolean transactionStarted) throws SQLException
  {
    request.setLogin(login);
    if (request.isAutoCommit() && transactionStarted)
    {
      vdb.commit(tid);
      return false;
    }
    else
      request.setTransactionId(tid);
    return transactionStarted;
  }

  /**
   * Gets a connection from the connection queue and process it.
   */
  public void run()
  {
    ArrayList vdbActiveThreads = vdb.getActiveThreads();
    ArrayList vdbPendingQueue = vdb.getPendingConnections();
    // List of open ResultSets for streaming. This is not synchronized since the
    // connection does only handle one request at a time
    streamedResultSet = new HashMap();
    boolean isActive = true;

    if (vdbActiveThreads == null)
    {
      logger
          .error("Got null active threads queue in VirtualDatabaseWorkerThread");
      isKilled = true;
    }
    if (vdbPendingQueue == null)
    {
      logger.error("Got null connection queue in VirtualDatabaseWorkerThread");
      isKilled = true;
    }

    // Main loop
    while (!isKilled)
    {
      // Get a connection from the pending queue
      synchronized (vdbPendingQueue)
      {
        while (vdbPendingQueue.isEmpty())
        {
          if (!vdb.poolConnectionThreads)
          { // User does not want thread pooling, kill this thread!
            isKilled = true;
            break;
          }
          boolean timeout = false;
          try
          {
            if (isActive)
            {
              isActive = false;
              // Remove ourselves from the active thread list
              synchronized (vdbActiveThreads)
              {
                vdbActiveThreads.remove(this);
                vdb.addIdleThread();
              }
            }
            long before = System.currentTimeMillis();
            vdbPendingQueue.wait(vdb.getMaxThreadIdleTime());
            long now = System.currentTimeMillis();
            // Check if timeout has expired
            timeout = now - before >= vdb.getMaxThreadIdleTime();
          }
          catch (InterruptedException e)
          {
            logger.warn("VirtualDatabaseWorkerThread wait() interrupted");
          }
          if (timeout && vdbPendingQueue.isEmpty())
          {
            if (vdb.currentNbOfThreads > vdb.minNbOfThreads)
            { // We have enough threads, kill this one
              isKilled = true;
              break;
            }
          }
        }

        if (isKilled)
        { // Cleaning up
          synchronized (vdbActiveThreads)
          { // Remove ourselves from the appropriate thread list
            if (isActive)
            {
              vdbActiveThreads.remove(this);
              vdb.removeCurrentNbOfThread();
            }
            else
              vdb.removeIdleThread();
          }
          // Get out of the while loop
          continue;
        }

        // Get a connection
        try
        {
          in = (CJDBCInputStream) vdbPendingQueue.remove(0);
          out = (CJDBCOutputStream) vdbPendingQueue.remove(0);
        }
        catch (Exception e)
        {
          logger.error("Error while getting streams from connection");
          continue;
        }

        synchronized (vdbActiveThreads)
        {
          if (!isActive)
          {
            vdb.removeIdleThread();
            isActive = true;
            // Add this thread to the active thread list
            vdbActiveThreads.add(this);
          }
        }
      }

      // Handle connection
      // Read the user information and check authentication
      boolean success = false;
      try
      {
        login = in.readUTF();
        String password = in.readUTF();
        user = new VirtualDatabaseUser(login, password);

        if (vdb.getAuthenticationManager().isValidVirtualUser(user))
        { // Authentication ok, send if SQL skeleton is needed
          needSkeleton = vdb.getRequestManager()
              .getRequiredParsingGranularity() != ParsingGranularities.NO_PARSING;
          out.writeObject(needSkeleton ? Boolean.TRUE : Boolean.FALSE);
          out.writeUTF(vdb.getBlobFilter().getXml());
          out.flush();
          success = true;

          if (logger.isDebugEnabled())
            logger.debug("Login accepted for " + login);
        }
        else
        { // Authentication failed, close the connection
          String msg = "Authentication failed for user '" + login + "'";
          out.writeObject(new SQLException(msg));
          if (logger.isDebugEnabled())
            logger.debug(msg);
          continue;
        }
      }
      catch (OptionalDataException e)
      {
        logger.error("Protocol error while expecting user authentication (" + e
            + ")");
        continue;
      }
      catch (IOException e)
      {
        logger.error("I/O error during user authentication (" + e + ")");
        continue;
      }
      finally
      {
        if (!success)
        {
          try
          {
            out.close();
            in.close();
          }
          catch (IOException ignore)
          {
          }
        }
      }

      currentTid = 0;
      transactionStarted = false;
      queryExecutedInThisTransaction = false;
      closed = false;
      int command;
      while (!closed && !isKilled)
      {
        try
        {
          // Get the query
          waitForCommand = true;
          command = in.readInt();
          waitForCommand = false;

          // Process it
          switch (command)
          {
            case Commands.ExecReadRequest :
              execReadRequest();
              break;
            case Commands.ExecWriteRequest :
              execWriteRequest();
              break;
            case Commands.ExecWriteRequestWithKeys :
              execWriteRequestWithKeys();
              break;
            case Commands.ExecReadStoredProcedure :
              execReadStoredProcedure();
              break;
            case Commands.ExecWriteStoredProcedure :
              execWriteStoredProcedure();
              break;
            case Commands.Begin :
              begin();
              break;
            case Commands.Commit :
              commit();
              break;
            case Commands.SetAutoCommit :
              setAutoCommit();
              break;
            case Commands.Rollback :
              rollback();
              break;
            case Commands.GetVirtualDatabaseName :
              getVirtualDatabaseName();
              break;
            case Commands.DatabaseMetaDataGetDatabaseProductName :
              databaseMetaDataGetDatabaseProductName();
              break;
            case Commands.GetControllerVersionNumber :
              getControllerVersionNumber();
              break;
            case Commands.DatabaseMetaDataGetTables :
              databaseMetaDataGetTables();
              break;
            case Commands.DatabaseMetaDataGetColumns :
              databaseMetaDataGetColumns();
              break;
            case Commands.DatabaseMetaDataGetPrimaryKeys :
              databaseMetaDataGetPrimaryKeys();
              break;
            case Commands.DatabaseMetaDataGetProcedures :
              databaseMetaDataGetProcedures();
              break;
            case Commands.DatabaseMetaDataGetProcedureColumns :
              databaseMetaDataGetProcedureColumns();
              break;
            case Commands.ConnectionGetCatalogs :
              connectionGetCatalogs();
              break;
            case Commands.ConnectionGetCatalog :
              connectionGetCatalog();
              break;
            case Commands.DatabaseMetaDataGetTableTypes :
              databaseMetaDataGetTableTypes();
              break;
            case Commands.DatabaseMetaDataGetSchemas :
              databaseMetaDataGetSchemas();
              break;
            case Commands.DatabaseMetaDataGetTablePrivileges :
              databaseMetaDataGetTablePrivileges();
              break;
            case Commands.ConnectionSetCatalog :
              connectionSetCatalog();
              break;
            case Commands.Close :
              close();
              break;
            case Commands.Reset :
              reset();
              break;
            case Commands.FetchNextResultSetRows :
              fetchNextResultSetRows();
              break;
            case Commands.CloseRemoteResultSet :
              closeRemoteResultSet();
              break;
            case Commands.DatabaseStaticMetadata :
              databaseStaticMetadata();
              break;
            case Commands.RestoreConnectionState :
              restoreConnectionState();
              break;
            default :
              String errorMsg = "Unsupported protocol command: " + command;
              logger.error(errorMsg);
              out.writeObject(new RuntimeException(errorMsg));
              out.flush();
              break;
          }
        }
        catch (OptionalDataException e)
        {
          logger.warn("Protocol error (" + e + ")");
          try
          {
            out.writeObject(e);
            out.flush();
          }
          catch (IOException ignore)
          {
          }
        }
        catch (RemoteException e)
        {
          logger.warn("Error during command execution (" + e + ")");
          try
          {
            out.writeObject(e);
            out.flush();
          }
          catch (IOException ignore)
          {
          }
        }
        catch (EOFException e)
        {
          logger.warn("Client (login:" + login + ",host:"
              + in.getSocket().getInetAddress().getHostName()
              + " closed connection with server");
          closed = true;
        }
        catch (SocketException e)
        {
          // shutting down
          closed = true;
        }
        catch (IOException e)
        {
          closed = true;
          logger.warn("Closing connection with client " + login
              + " because of IOException.(" + e + ")");
        }
        catch (ClassNotFoundException e)
        {
          logger.error("Protocol error (" + e + ")");
          try
          {
            out.writeObject(e);
            out.flush();
          }
          catch (IOException ignore)
          {
          }
        }
        catch (SQLException e)
        {
          logger
              .warn("Error during command execution (" + e.getMessage() + ")");
          try
          {
            out.writeObject(e);
            out.flush();
          }
          catch (IOException ignore)
          {
          }
        }
        catch (RuntimeException e)
        {
          logger.warn("Runtime error during command execution ("
              + e.getMessage() + ")", e);
          try
          {
            out.writeObject(new SQLException(e.getMessage()));
            out.flush();
          }
          catch (IOException ignore)
          {
          }
        }
      }

      // Do the cleanup
      if (transactionStarted)
      {
        if (logger.isDebugEnabled())
          logger.debug("Forcing transaction " + currentTid + " rollback");
        try
        {
          vdb.rollback(currentTid);
        }
        catch (Exception e)
        {
          if (logger.isDebugEnabled())
            logger.debug("Error during rollback of transaction " + currentTid
                + "(" + e + ")");
        }
      }
      if (!streamedResultSet.isEmpty())
      {
        for (Iterator iter = streamedResultSet.values().iterator(); iter
            .hasNext();)
        {
          ControllerResultSet crs = (ControllerResultSet) iter.next();
          crs.closeResultSet();
        }
        streamedResultSet.clear();
      }
      try
      {
        in.close();
      }
      catch (IOException ignore)
      {
      }
      try
      {
        out.close();
      }
      catch (IOException ignore)
      {
      }
    }

    if (logger.isDebugEnabled())
      logger.debug("VirtualDatabaseWorkerThread associated to login: "
          + this.getUser() + " terminating.");
  }

  //
  // Connection management
  //

  private void close() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("Close command");
    out.writeObject(completed);
    out.flush();
    closed = true;
  }

  private void closeRemoteResultSet() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("CloseRemoteResultSet command");

    String cursor = in.readUTF();
    ControllerResultSet crsToClose = (ControllerResultSet) streamedResultSet
        .remove(cursor);
    if (crsToClose == null)
    {
      out
          .writeObject(new SQLException(
              "No valid ControllerResultSet to close."));
      out.flush();
    }
    else
    {
      crsToClose.closeResultSet();
      out.writeObject(completed);
      out.flush();
    }
  }

  private void reset()
  {
    // The client application has closed the connection but it is kept
    // open in case the transparent connection pooling reuses it.
    if (logger.isDebugEnabled())
      logger.debug("Reset command");

    // Do the cleanup
    if (transactionStarted)
    {
      if (queryExecutedInThisTransaction)
      { // Force rollback of this transaction
        if (logger.isDebugEnabled())
          logger.debug("Forcing transaction " + currentTid + " rollback");
        try
        {
          vdb.rollback(currentTid);
        }
        catch (Exception e)
        {
          if (logger.isDebugEnabled())
            logger.debug("Error during rollback of transaction " + currentTid
                + "(" + e + ")");
        }
      }
      else
      { // We need to abort the begin to cleanup the metadata
        // associated with the started transaction.
        if (logger.isDebugEnabled())
          logger.debug("Aborting transaction " + currentTid);
        try
        {
          vdb.abort(currentTid);
        }
        catch (Exception e)
        {
          if (logger.isDebugEnabled())
            logger.debug("Error while aborting transaction " + currentTid + "("
                + e + ")", e);
        }
      }
      currentTid = 0;
      transactionStarted = false;
    }
  }

  private void restoreConnectionState() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("RestoreConnectionState command");
    // We receive autocommit from driver
    transactionStarted = !in.readBoolean();
    if (transactionStarted)
      currentTid = in.readLong();
  }

  //
  // Catalog
  //

  private void connectionSetCatalog() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("ConnectionSetCatalog command");
    String catalog = in.readUTF();
    boolean change = controller.hasVirtualDatabase(catalog);
    if (change)
    {
      VirtualDatabase tempvdb = controller.getVirtualDatabase(catalog);
      if (!tempvdb.getAuthenticationManager().isValidVirtualUser(user))
        out.writeObject(new SQLException(
            "User authentication has failed for asked catalog. No change"));
      else
      {
        this.vdb = tempvdb;
        out.writeObject(Boolean.TRUE);
      }
    }
    else
      out.writeObject(Boolean.FALSE);
    out.flush();
  }

  private void connectionGetCatalog() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("ConnectionGetCatalog command");
    out.writeObject(vdb.getVirtualDatabaseName());
    out.flush();
  }

  private void connectionGetCatalogs() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("ConnectionGetCatalogs command");
    ArrayList list = controller.getVirtualDatabaseNames();
    out.writeObject(vdb.getDynamicMetaData().getCatalogs(list));
    out.flush();
  }

  //
  // Database MetaData
  //

  private void databaseMetaDataGetColumns() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("DatabaseMetaDataGetColumns command");
    String ccatalog = in.readUTF();
    String cschemaPattern = in.readUTF();
    String ctableNamePattern = in.readUTF();
    String ccolumnNamePattern = in.readUTF();
    out.writeObject(vdb.getDynamicMetaData().getColumns(ccatalog,
        cschemaPattern, ctableNamePattern, ccolumnNamePattern));
    out.flush();
  }

  private void databaseMetaDataGetDatabaseProductName() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("GetDatabaseProductName command");
    out.writeObject(vdb.getDatabaseProductName());
    out.flush();
  }

  private void databaseMetaDataGetPrimaryKeys() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("DatabaseMetaDataGetPrimaryKeys command");
    String pcatalog = in.readUTF();
    String pschemaPattern = in.readUTF();
    String ptableNamePattern = in.readUTF();
    out.writeObject(vdb.getDynamicMetaData().getPrimaryKeys(pcatalog,
        pschemaPattern, ptableNamePattern));
    out.flush();
  }

  private void databaseMetaDataGetProcedureColumns() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("DatabaseMetaDataGetProcedureColumns command");
    String pccatalog = in.readUTF();
    String pcschemaPattern = in.readUTF();
    String pcprocedureNamePattern = in.readUTF();
    String pccolumnNamePattern = in.readUTF();
    out.writeObject(vdb.getDynamicMetaData().getProcedureColumns(pccatalog,
        pcschemaPattern, pcprocedureNamePattern, pccolumnNamePattern));
    out.flush();
  }

  private void databaseMetaDataGetProcedures() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("DatabaseMetaDataGetProcedures command");
    String rcatalog = in.readUTF();
    String rschemaPattern = in.readUTF();
    String procedureNamePattern = in.readUTF();
    out.writeObject(vdb.getDynamicMetaData().getProcedures(rcatalog,
        rschemaPattern, procedureNamePattern));
    out.flush();
  }

  private void databaseMetaDataGetSchemas() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("DatabaseMetaDataGetSchemas Types command");
    out.writeObject(vdb.getDynamicMetaData().getSchemas());
    out.flush();
  }

  private void databaseMetaDataGetTablePrivileges() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("DatabaseMetaDataGetTablePrivileges command");
    String tpcatalog = in.readUTF();
    String tpschemaPattern = in.readUTF();
    String tptablePattern = in.readUTF();
    out.writeObject(vdb.getDynamicMetaData().getTablePrivileges(tpcatalog,
        tpschemaPattern, tptablePattern));
    out.flush();
  }

  private void databaseMetaDataGetTables() throws IOException,
      ClassNotFoundException, OptionalDataException
  {
    if (logger.isDebugEnabled())
      logger.debug("DatabaseMetaDataGetTables command");
    String tcatalog = in.readUTF();
    String tschemaPattern = in.readUTF();
    String ttableNamePattern = in.readUTF();
    String[] ttypes = (String[]) in.readObject();
    out.writeObject(vdb.getDynamicMetaData().getTables(tcatalog,
        tschemaPattern, ttableNamePattern, ttypes));
    out.flush();
  }

  private void databaseMetaDataGetTableTypes() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("DatabaseMetaDataGetTable Types command");
    out.writeObject(vdb.getDynamicMetaData().getTableTypes());
    out.flush();
  }

  private void databaseStaticMetadata() throws IOException
  {
    String key = in.readUTF();
    if (logger.isDebugEnabled())
      logger.debug("DatabaseStaticMetadata command for " + key);
    MetadataContainer container = vdb.getStaticMetaData()
        .getMetadataContainer();
    if (container == null)
    {
      String msg = "No metadata is available probably because no backend is enabled on that controller.";
      logger.info(msg);
      out.writeObject(new SQLException(msg));
    }
    else
      out.writeObject(container.get(key));
    out.flush();
  }

  private void getControllerVersionNumber() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("GetControllerVersionNumber command");
    out.writeObject(Constants.VERSION);
    out.flush();
  }

  private void getVirtualDatabaseName() throws IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("GetVirtualDatabaseName command");
    out.writeObject(vdb.getDatabaseName());
    out.flush();
  }

  //
  // Transaction management
  //

  private void begin() throws SQLException, IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("Begin command");
    currentTid = vdb.begin(login);
    out.writeObject(new Long(currentTid));
    out.flush();
    transactionStarted = true;
    queryExecutedInThisTransaction = false;
  }

  private void commit() throws SQLException, IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("Commit command");
    vdb.commit(currentTid);
    currentTid = vdb.begin(login);
    out.writeObject(new Long(currentTid));
    out.flush();
    queryExecutedInThisTransaction = false;
  }

  private void rollback() throws SQLException, IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("Rollback command");
    vdb.rollback(currentTid);
    currentTid = vdb.begin(login);
    out.writeObject(new Long(currentTid));
    out.flush();
    queryExecutedInThisTransaction = false;
  }

  private void setAutoCommit() throws SQLException, IOException
  {
    if (logger.isDebugEnabled())
      logger.debug("Set Auto commit command");
    vdb.commit(currentTid);
    currentTid = 0;
    transactionStarted = false;
    out.writeObject(Boolean.TRUE);
    out.flush();
  }

  //
  // Request execution
  //

  private void execReadRequest() throws OptionalDataException, IOException,
      SQLException
  {
    if (logger.isDebugEnabled())
      logger.debug("ExecReadRequest command");
    SelectRequest select = decodeReadRequestFromStream();
    transactionStarted = setRequestParameters(select, login, currentTid,
        transactionStarted);
    if (!transactionStarted)
      currentTid = 0;
    else
      queryExecutedInThisTransaction = true;
    ControllerResultSet crs = vdb.execReadRequest(select);
    out.writeObject(crs.getFields());
    out.writeObject(crs.getData());
    out.writeBoolean(crs.hasMoreData());
    if (crs.hasMoreData())
    { // Send the cursor name for further references
      out.writeUTF(crs.getCursorName());
      System.out.println("adding streamedresultset");
      // Keep a reference to the ResultSet
      streamedResultSet.put(crs.getCursorName(), crs);
    }
    else
      crs = null; // release the reference right away
    out.flush();
  }

  private void execReadStoredProcedure() throws OptionalDataException,
      IOException, SQLException
  {
    if (logger.isDebugEnabled())
      logger.debug("ExecReadStoredProcedure command");
    StoredProcedure readProc = decodeProcedureFromStream(true);
    transactionStarted = setRequestParameters(readProc, login, currentTid,
        transactionStarted);
    if (!transactionStarted)
      currentTid = 0;
    else
      queryExecutedInThisTransaction = true;
    ControllerResultSet sprs = vdb.execReadStoredProcedure(readProc);
    out.writeObject(sprs.getFields());
    out.writeObject(sprs.getData());
    out.writeBoolean(sprs.hasMoreData());
    if (sprs.hasMoreData())
    { // Send the cursor name for further references
      out.writeUTF(sprs.getCursorName());
      System.out.println("adding streamedresultset");
      // Keep a reference to the ResultSet
      streamedResultSet.put(sprs.getCursorName(), sprs);
    }
    else
      sprs = null; // release references right away
    readProc = null;
    out.flush();
  }

  private void execWriteRequest() throws OptionalDataException, IOException,
      SQLException
  {
    if (logger.isDebugEnabled())
      logger.debug("ExecWriteRequest command");
    AbstractWriteRequest write = decodeWriteRequestFromStream(false);
    transactionStarted = setRequestParameters(write, login, currentTid,
        transactionStarted);
    if (!transactionStarted)
      currentTid = 0;
    else
      queryExecutedInThisTransaction = true;
    out.writeObject(new Integer(vdb.execWriteRequest(write)));
    write = null; // release the reference right away
    out.flush();
  }

  private void execWriteRequestWithKeys() throws OptionalDataException,
      IOException, SQLException
  {
    if (logger.isDebugEnabled())
      logger.debug("ExecWriteRequestWithKeys command");
    AbstractWriteRequest writeWithKeys = decodeWriteRequestFromStream(true);
    transactionStarted = setRequestParameters(writeWithKeys, login, currentTid,
        transactionStarted);
    if (!transactionStarted)
      currentTid = 0;
    else
      queryExecutedInThisTransaction = true;
    ControllerResultSet keys = vdb.execWriteRequestWithKeys(writeWithKeys);
    out.writeObject(keys.getFields());
    out.writeObject(keys.getData());
    out.writeBoolean(keys.hasMoreData());
    if (keys.hasMoreData())
    { // Send the cursor name for further references
      out.writeUTF(keys.getCursorName());
      System.out.println("adding streamedresultset");
      // Keep a reference to the ResultSet
      streamedResultSet.put(keys.getCursorName(), keys);
    }
    else
      keys = null; // release references right away
    writeWithKeys = null;
    out.flush();
  }

  private void execWriteStoredProcedure() throws OptionalDataException,
      IOException, SQLException
  {
    if (logger.isDebugEnabled())
      logger.debug("ExecWriteStoredProcedure command");
    StoredProcedure writeProc = decodeProcedureFromStream(false);
    transactionStarted = setRequestParameters(writeProc, login, currentTid,
        transactionStarted);
    if (!transactionStarted)
      currentTid = 0;
    else
      queryExecutedInThisTransaction = true;
    out.writeObject(new Integer(vdb.execWriteStoredProcedure(writeProc)));
    writeProc = null; // release reference right away
    out.flush();
  }

  private void fetchNextResultSetRows() throws IOException, SQLException
  {
    if (logger.isDebugEnabled())
      logger.debug("FetchNextResultSetRows command");

    String cursorName = in.readUTF();
    int fetchSize = in.readInt();
    ControllerResultSet fetchCrs = (ControllerResultSet) streamedResultSet
        .get(cursorName);
    if (fetchCrs == null)
    {
      out.writeObject(new SQLException(
          "No valid ControllerResultSet to fetch data from"));
      out.flush();
    }
    else
    {
      out.writeObject(fetchCrs.fetchData(fetchSize));
      out.writeBoolean(fetchCrs.hasMoreData());
      out.flush();
      if (!fetchCrs.hasMoreData())
        streamedResultSet.remove(cursorName);
    }
  }

  //
  // Public API
  //

  /**
   * Retrieve general information on this client
   * 
   * @return an array of string
   */
  public String[] retrieveClientData()
  {
    String[] data = new String[6];
    data[0] = in.getSocket().getInetAddress().getHostName();
    data[1] = in.getSocket().getInetAddress().getHostAddress();
    data[2] = String.valueOf(in.getBytesRead());
    data[3] = String.valueOf(out.getBytesWritten());
    data[4] = String.valueOf(in.getUseCompression());
    data[5] = String
        .valueOf(((System.currentTimeMillis() - in.getDateCreated()) / 1000));
    data[6] = String.valueOf(in.getSpeed());
    data[7] = String.valueOf(out.getSpeed());
    return data;
  }

  /**
   * get bytes read
   * 
   * @return bytes read
   */
  public long getBytesRead()
  {
    return in.getBytesRead();
  }

  /**
   * get bytes written
   * 
   * @return bytes written
   */
  public long getBytesWritten()
  {
    return out.getBytesWritten();
  }

  /**
   * get time active
   * 
   * @return time active since started
   */
  public long getTimeActive()
  {
    return ((System.currentTimeMillis() - in.getDateCreated()) / 1000);
  }

  /**
   * get reading speed, WARNING! This is approximate
   * 
   * @return reading speed
   */
  public long getReadingSpeed()
  {
    return in.getSpeed();
  }

  /**
   * get writing speed, WARNING! This is approximate
   * 
   * @return writing speed
   */
  public long getWritingSpeed()
  {
    return out.getSpeed();
  }

  /**
   * @return Returns the login of the current user.
   */
  public String getUser()
  {
    return user.getLogin();
  }

  /**
   * Shutdown this thread by setting <code>isKilled</code> value to true. This
   * gives time to check for needed rollback transactions
   */
  public void shutdown()
  {
    // Tell this thread to stop working gently.
    // This will cancel transaction if needed
    this.isKilled = true;
    try
    {
      if (waitForCommand)
      {
        // close only the streams if we're not in the middle of a request
        in.close();
        out.close();
      }
    }
    catch (IOException e)
    {
      // ignore, only the input stream should be close
      // for this thread to end
    }
  }

}

---