CJDBCInputStream.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): Nicolas Modrzyk
 * Contributor(s): Emmanuel Cecchet.
 */

package org.objectweb.cjdbc.common.stream;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.OptionalDataException;
import java.io.StreamCorruptedException;
import java.net.Socket;

/**
 * This class is to provide an ObjectInputStream implementation, but we cannot
 * override the readObject method, so we made as a wrapper. The CJDBCInputStream
 * in pair with the CJDBCOutputStream offers a five times compression using the
 * java Inflater and Deflater class. <code>useCompression</code> can be set to
 * on or off anytime.
 * 
 * @author <a href="mailto:Nicolas.Modrzyk@inrialpes.fr">Nicolas Modrzyk </a>
 * @author <a href="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
 */
public class CJDBCInputStream
{
  private ObjectInputStream input;
  private long              bytesRead;
  private Socket            socket;
  private boolean           useCompression = false;
  private long              speed          = -1;
  private long              dateCreated;

  /**
   * Creates a new CJDBCInputStream from the given input stream.
   * 
   * @param in inputstream to wrap
   * @throws IOException if <code>new ObjectInputStream()</code> throws
   *           anException
   * @throws StreamCorruptedException same as above
   */
  public CJDBCInputStream(InputStream in) throws IOException,
      StreamCorruptedException
  {
    input = new ObjectInputStream(new BufferedInputStream(in));
    bytesRead = 0;
  }

  /**
   * Useful constructor for statistics on sockets ..
   * 
   * @param socket socket for this stream
   * @throws IOException if an error occurs
   * @throws StreamCorruptedException if an error occurs
   */
  public CJDBCInputStream(Socket socket) throws IOException,
      StreamCorruptedException
  {
    this(socket.getInputStream());
    this.socket = socket;
    dateCreated = System.currentTimeMillis();
  }

  /**
   * @return the Object read from the stream
   * @throws ClassNotFoundException if an error occurs
   * @throws IOException if an error occurs
   * @throws OptionalDataException if an error occurs
   * @see java.io.ObjectInputStream#readObject()
   */
  public Object readObject() throws ClassNotFoundException, IOException,
      OptionalDataException
  {
    if (useCompression)
    {
      int len = input.readInt();
      byte[] read = new byte[len];
      input.readFully(read);

      bytesRead += len + 4;
      // we read an integer in the pipe for the size of the array.
      return CJDBCStream.decompressObject(read);
    }
    else
    {
      int objectType = input.readInt();
      switch (objectType)
      {
        case CJDBCStream.OBJECT :
          // Use the unshared version to prevent the stream to keep references
          // on objects. This prevents objects sent through the channel from
          // being garbage collected and results in memory leaks.
          return input.readUnshared();
        case CJDBCStream.NULL :
          return null;
        case CJDBCStream.STRING_OBJECT :
          return input.readUTF();
        case CJDBCStream.INTEGER_OBJECT :
          return new Integer(input.readInt());
        case CJDBCStream.LONG_OBJECT :
          return new Long(input.readLong());
        case CJDBCStream.BOOLEAN_OBJECT :
          return input.readBoolean() ? Boolean.TRUE : Boolean.FALSE;
        default :
          throw new RuntimeException("Unhandled type on stream: " + objectType);
      }
    }
  }

  /**
   * @see java.io.ObjectInputStream#close()
   * @throws IOException if an error occurs
   */
  public void close() throws IOException
  {
    input.close();
  }

  /**
   * @see java.io.ObjectInputStream#readBoolean()
   * @return a boolean value
   * @throws IOException if an error occurs
   */
  public boolean readBoolean() throws IOException
  {
    bytesRead++;
    return input.readBoolean();
  }

  /**
   * @see java.io.ObjectInputStream#readInt()
   * @return an int value
   * @throws IOException if an error occurs
   */
  public int readInt() throws IOException
  {
    bytesRead += 4;
    return input.readInt();
  }

  /**
   * @see java.io.ObjectInputStream#readLong()
   * @return a long value
   * @throws IOException if an error occurs
   */
  public long readLong() throws IOException
  {
    bytesRead += 8;
    return input.readLong();
  }

  /**
   * @see java.io.ObjectInputStream#readUTF()
   * @return a String in UTF format
   * @throws IOException if an error occurs
   */
  public String readUTF() throws IOException
  {
    Object read;
    try
    {
      read = this.readObject();
    }
    catch (Exception e)
    {
      throw new IOException(e.getMessage());
    }
    if (read == null)
      return null;
    bytesRead += ((String) read).length();
    return (String) read;
  }

  /**
   * @see java.io.ObjectInputStream#available()
   * @return the number of available bytes.
   * @throws IOException if an error occurs
   */
  public int available() throws IOException
  {
    return input.available();
  }

  /**
   * @return Returns the bytesRead.
   */
  public long getBytesRead()
  {
    return bytesRead;
  }

  /**
   * @return Returns the socket.
   */
  public Socket getSocket()
  {
    return socket;
  }

  /**
   * @return Returns the useCompression.
   */
  public boolean getUseCompression()
  {
    return useCompression;
  }

  /**
   * @return Returns the dateCreated.
   */
  public long getDateCreated()
  {
    return dateCreated;
  }

  /**
   * @return Returns the speed.
   */
  public long getSpeed()
  {
    return speed;
  }
}

---