src/org/objectweb/cjdbc/controller/virtualdatabase/VirtualDatabaseWorkerThread.java

説明を見る。

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.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;

public class VirtualDatabaseWorkerThread extends Thread
{
  //
  // How the code is organized?
  //
  // 1. Member variables
  // 2. Constructor(s)
  // 3. Request management
  // 4. Getter/Setters

  private static int             threadID     = 0;

  private boolean                isKilled     = false;

  boolean                        needSkeleton = false;

  private VirtualDatabase        vdb;

  private final CommandCompleted completed    = new CommandCompleted();

  private Trace                  logger       = null;

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

  private VirtualDatabaseUser    user;

  private Controller             controller;

  private boolean                waitForCommand;

  private HashMap                streamedResultSet;

  /*
   * Constructor
   */

  public VirtualDatabaseWorkerThread(Controller controller, VirtualDatabase vdb)
  {
    super("VirtualDatabaseWorkerThread-" + vdb.getName());
    this.vdb = vdb;
    this.controller = controller;
    try
    {
      this.logger = Trace
          .getLogger("org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread."
              + vdb.getName() + "." + (threadID++));
    }
    catch (Exception e)
    {
      this.logger = vdb.logger;
    }
  }

  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
    }

  }

  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;
  }

  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;
      String login;
      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(new Boolean(needSkeleton));
          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)
          {
          }
        }
      }

      // Ok, let's process the queries on this connection
      long currentTid = 0;
      boolean transactionStarted = false;
      boolean queryExecutedInThisTransaction = false;
      boolean 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 :
              if (logger.isDebugEnabled())
                logger.debug("ExecReadRequest command");
              /* Recreate select request begin */
              SelectRequest select = readRequestFromStream();
              transactionStarted = setRequestParameters(select, login,
                  currentTid, transactionStarted);
              /* Recreate select request end */

              if (!transactionStarted)
                currentTid = 0;
              else
                queryExecutedInThisTransaction = true;
              ControllerResultSet crs = vdb.execReadRequest(select);
              if (crs.hasMoreData())
                streamedResultSet.put(crs.getCursorName(), crs);
              out.writeObject(crs.getFields());
              out.writeObject(crs.getData());
              out.writeBoolean(crs.hasMoreData());
              if (crs.hasMoreData())
                out.writeUTF(crs.getCursorName());
              out.flush();
              break;
            case Commands.ExecWriteRequest :
              if (logger.isDebugEnabled())
                logger.debug("ExecWriteRequest command");
              AbstractWriteRequest write = writeRequestFromStream(false);
              transactionStarted = setRequestParameters(write, login,
                  currentTid, transactionStarted);
              if (!transactionStarted)
                currentTid = 0;
              else
                queryExecutedInThisTransaction = true;
              out.writeObject(new Integer(vdb.execWriteRequest(write)));
              out.flush();
              break;
            case Commands.ExecWriteRequestWithKeys :
              if (logger.isDebugEnabled())
                logger.debug("ExecWriteRequestWithKeys command");
              AbstractWriteRequest writeWithKeys = writeRequestFromStream(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())
                out.writeUTF(keys.getCursorName());
              out.flush();
              break;
            case Commands.ExecReadStoredProcedure :
              if (logger.isDebugEnabled())
                logger.debug("ExecReadStoredProcedure command");
              StoredProcedure readProc = procedureFromStream(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())
                out.writeUTF(sprs.getCursorName());
              out.flush();
              break;
            case Commands.ExecWriteStoredProcedure :
              if (logger.isDebugEnabled())
                logger.debug("ExecWriteStoredProcedure command");
              StoredProcedure writeProc = procedureFromStream(false);
              transactionStarted = setRequestParameters(writeProc, login,
                  currentTid, transactionStarted);
              if (!transactionStarted)
                currentTid = 0;
              else
                queryExecutedInThisTransaction = true;
              out.writeObject(new Integer(vdb
                  .execWriteStoredProcedure(writeProc)));
              out.flush();
              break;
            case Commands.Begin :
              if (logger.isDebugEnabled())
                logger.debug("Begin command");
              currentTid = vdb.begin(login);
              out.writeObject(new Long(currentTid));
              out.flush();
              transactionStarted = true;
              break;
            case Commands.Commit :
              if (logger.isDebugEnabled())
                logger.debug("Commit command");
              vdb.commit(currentTid);
              currentTid = vdb.begin(login);
              out.writeObject(new Long(currentTid));
              out.flush();
              break;
            case Commands.SetAutoCommit :
              if (logger.isDebugEnabled())
                logger.debug("Set Auto commit command");
              vdb.commit(currentTid);
              currentTid = 0;
              transactionStarted = false;
              out.writeObject(Boolean.TRUE);
              out.flush();
              break;
            case Commands.Rollback :
              if (logger.isDebugEnabled())
                logger.debug("Rollback command");
              vdb.rollback(currentTid);
              currentTid = vdb.begin(login);
              out.writeObject(new Long(currentTid));
              out.flush();
              break;
            case Commands.GetVirtualDatabaseName :
              if (logger.isDebugEnabled())
                logger.debug("GetVirtualDatabaseName command");
              out.writeObject(vdb.getDatabaseName());
              out.flush();
              break;
            case Commands.GetDatabaseProductName :
              if (logger.isDebugEnabled())
                logger.debug("GetDatabaseProductName command");
              out.writeObject(vdb.getDatabaseProductName());
              out.flush();
              break;
            case Commands.GetControllerVersionNumber :
              if (logger.isDebugEnabled())
                logger.debug("GetControllerVersionNumber command");
              out.writeObject(Constants.VERSION);
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetTables :
              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.getMetaData().getTables(tcatalog,
                  tschemaPattern, ttableNamePattern, ttypes));
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetColumns :
              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.getMetaData().getColumns(ccatalog,
                  cschemaPattern, ctableNamePattern, ccolumnNamePattern));
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetPrimaryKeys :
              if (logger.isDebugEnabled())
                logger.debug("DatabaseMetaDataGetPrimaryKeys command");
              String pcatalog = in.readUTF();
              String pschemaPattern = in.readUTF();
              String ptableNamePattern = in.readUTF();
              out.writeObject(vdb.getMetaData().getPrimaryKeys(pcatalog,
                  pschemaPattern, ptableNamePattern));
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetProcedures :
              if (logger.isDebugEnabled())
                logger.debug("DatabaseMetaDataGetProcedures command");
              String rcatalog = in.readUTF();
              String rschemaPattern = in.readUTF();
              String procedureNamePattern = in.readUTF();
              out.writeObject(vdb.getMetaData().getProcedures(rcatalog,
                  rschemaPattern, procedureNamePattern));
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetProcedureColumns :
              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.getMetaData().getProcedureColumns(pccatalog,
                      pcschemaPattern, pcprocedureNamePattern,
                      pccolumnNamePattern));
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetCatalogs :
              if (logger.isDebugEnabled())
                logger.debug("DatabaseMetaDataGetCatalogs command");
              ArrayList list = controller.listVirtualDatabases();
              out.writeObject(vdb.getMetaData().getCatalogs(list));
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetCatalog :
              if (logger.isDebugEnabled())
                logger.debug("DatabaseMetaDataGetCatalog command");
              out.writeObject(vdb.getName());
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetTableTypes :
              if (logger.isDebugEnabled())
                logger.debug("DatabaseMetaDataGetTable Types command");
              out.writeObject(vdb.getMetaData().getTableTypes());
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetSchemas :
              if (logger.isDebugEnabled())
                logger.debug("DatabaseMetaDataGetSchemas Types command");
              out.writeObject(vdb.getMetaData().getSchemas());
              out.flush();
              break;
            case Commands.DatabaseMetaDataGetTablePrivileges :
              if (logger.isDebugEnabled())
                logger.debug("DatabaseMetaDataGetTablePrivileges command");
              String tpcatalog = in.readUTF();
              String tpschemaPattern = in.readUTF();
              String tptablePattern = in.readUTF();
              out.writeObject(vdb.getMetaData().getTablePrivileges(tpcatalog,
                  tpschemaPattern, tptablePattern));
              out.flush();
              break;
            case Commands.ConnectionSetCatalog :
              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(new Boolean(true));
                }
              }
              else
                out.writeObject(new Boolean(false));
              out.flush();
              break;
            case Commands.Close :
              if (logger.isDebugEnabled())
                logger.debug("Close command");
              out.writeObject(completed);
              out.flush();
              closed = true;
              break;
            case Commands.Reset :
              if (logger.isDebugEnabled())
                logger.debug("Reset command");

              // Do the cleanup
              if (transactionStarted && queryExecutedInThisTransaction)
              {
                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 + ")");
                }
              }
              currentTid = 0;
              transactionStarted = false;
              break;
            case Commands.FetchNextResultSetRows :
              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);
              }
              break;
            case Commands.CloseRemoteResultSet :
              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();
              }
              break;
            case Commands.RestoreConnectionState :
              if (logger.isDebugEnabled())
                logger.debug("RestoreConnectionState command");
              // We receive autocommit from driver
              transactionStarted = !in.readBoolean();
              if (transactionStarted)
                currentTid = in.readLong();
              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 + " closed connection with server", e);
          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() + ")");
          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.");
  }

  private StoredProcedure procedureFromStream(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;
  }

  private AbstractWriteRequest writeRequestFromStream(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;
  }

  private SelectRequest readRequestFromStream() 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;
  }

  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;
  }

  public long getBytesRead()
  {
    return in.getBytesRead();
  }

  public long getBytesWritten()
  {
    return out.getBytesWritten();
  }

  public long getTimeActive()
  {
    return ((System.currentTimeMillis() - in.getDateCreated()) / 1000);
  }

  public long getReadingSpeed()
  {
    return in.getSpeed();
  }

  public long getWritingSpeed()
  {
    return out.getSpeed();
  }

  public String getUser()
  {
    return user.getLogin();
  }
}

---