/*
    * Copyright (c) 1998-2002 The Jgroup Team.
    *
    * This program is free software; you can redistribute it and/or modify
    * it under the terms of the GNU Lesser General Public License version 2 as
    * published by the Free Software Foundation.
    *
    * This program 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 program; if not, write to the Free Software
   * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
   *
   */
  
  package jgroup.util;
  
  import java.io.EOFException;
  import java.io.Externalizable;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.ObjectInput;
  import java.io.ObjectInputStream;
  import java.io.ObjectOutput;
  import java.io.PushbackInputStream;
  import java.io.UTFDataFormatException;
  
  import jgroup.core.ConfigManager;
  
  /**
   *  Input stream message optimized to avoid message copying.
   *
   *  @author Alberto Montresor
   *  @author Hein Meling
   *  @since Jgroup 0.8
   */
  public final class InMessage
    extends InputStream
    implements ObjectInput, Externalizable
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Constants
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    private static final long serialVersionUID = 9133407439996893111L;
  
    /** Initial number of fragments composing a message; can grow */
    private static final int NUM_BUFFER = 16;     
    
    
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Static section
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Temporary buffer used in writeTemp */
    private static byte[] temp = new byte[8];
  
    
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    private int       payload;            // Payload size in bytes
    private int       header;             // Header size in bytes
    private int       trailer;            // Trailer size in bytes
    private int       totlen;             // Total length of a fragment
    private int       endpayload;         // Last byte used for payload
  
    private int       position;           // Current position in the stream [0..byteCount]
    private int       fragment;           // Current fragment number
    private int       offset;             // Current offset in the fragment
  
    private int       fragmentCount;      // Total number of fragments used
    private int       byteCount;          // Total number of bytes written
    
    private byte[]    buffer;             // Current buffer
    private byte[][]  fragments;          // Byte array fragments
  
    private int       marked;             // Mark position
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Constructors
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /**
     *  This constructor builds an empty message subdivided in fragments.
     *  Each fragment has a length of <code> header + payload + trailer</code>.
     *  <p>
     *  A <tt>IndexOutOfBoundsException</tt> is thrown if: <tt>payload</tt> is 
     *  negative or equal to zero; <tt>header</tt> is negative; <tt>trailer</tt> 
     *  is negative; <tt>size</tt> is negative or equal to zero.
     */
    public InMessage(int payload, int header, int trailer)
    {
     if (payload <= 0 || header < 0 || trailer < 0)
       throw new IndexOutOfBoundsException();
 
     // Store and compute fixed limits
     this.payload = payload;
     this.header  = header;
     this.trailer = trailer;
     endpayload = payload + header;
     totlen = endpayload + trailer;
 
     // Initialize variables regarding message size
     byteCount = 0;
     fragmentCount = 0;
 
     // Initialize variables regarding message position in the stream
     position = 0;
     fragment = 0;
     offset = header;
 
     // Create fragments
     fragments = new byte[NUM_BUFFER][];
   }
 
 
   /**
    *  This constructor builds a message input stream subdivided in 
    *  fragments with length <code> header + payload + trailer</code>.
    *
    *  @exception IndexOutOfBoundsException
    *    Raised if <code>payload</code> is negative or equal to zero;
    *    <code>header</code> is negative; <code>trailer</code> is
    *    negative; <code>size</code> is negative or equal to zero.
    */
   public InMessage(int payload, int header, int trailer, int size)
   {
     if (payload <= 0 || header < 0 || trailer < 0 || size <= 0)
       throw new IndexOutOfBoundsException();
 
     // Store and compute fixed limits
     this.payload = payload;
     this.header  = header;
     this.trailer = trailer;
     endpayload = payload + header;
     totlen = endpayload + trailer;
 
     // Initialize variables regarding message size
     byteCount = 0;
     fragmentCount = 0;
 
     // Initialize variables regarding message position in the stream
     position = 0;
     fragment = 0;
     offset = header;
 
     // Create fragments
     fragments = new byte[size % payload + 1][];
   }
 
 
   /**
    * Construct a InMessage starting from an OutMessage. In this way, it possible 
    * to local deliver messages without copying them.
    */
   public InMessage(OutMessage msg)
   {
     // Store and compute fixed limits
     this.payload = msg.getPayload();
     this.header  = msg.getHeader();
     this.trailer = msg.getTrailer();
     endpayload = payload + header;
     totlen = endpayload + trailer;
     
     // Initialize variables regarding message size
     byteCount = msg.getByteCount();
     fragmentCount = msg.getFragmentCount();
 
     // Initialize variables regarding message position in the stream
     position = 0;
     fragment = 0;
     offset = header;
     
     // Create fragments
     fragments = msg.getFragments();
     buffer = fragments[0];
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods from Externalizable
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Default constructor for externalization.
    */
   public InMessage() { }
 
 
   /**
    *  Restores the content of this object from the marshalled data contained
    *  in the specified input stream.
    *
    *  @param in the stream to be read
    */
   public void readExternal(ObjectInput in)
     throws IOException
   {
     payload       = in.readInt();
     header        = in.readInt();
     trailer       = in.readInt();
     endpayload    = payload + header;
     totlen        = endpayload + trailer;
     fragmentCount = in.readInt();
     byteCount     = in.readInt();
     fragment      = in.readInt();
     position      = in.readInt();
     marked        = in.readInt();
 
     // Initialize the offset variable based on the parameters from the stream
     offset        = header + position;
 
     // Reconstruct fragments from stream
     fragments = new byte[fragmentCount][];
     int fcount = fragmentCount - 1;
     for (int i = 0; i < fcount; i++) {
       fragments[i] = new byte[totlen];
       in.readFully(fragments[i], 0, totlen);
     }
     fragments[fcount] = new byte[totlen];
     int lastFragLen = getLastPayload() + header + trailer;
     in.readFully(fragments[fcount], 0, lastFragLen);
     buffer = fragments[0];
 
 //    System.out.println("readExternal: " + this.toString(true));
   }
 
   /**
    *  Marshals the content of this object to the specified output stream.
    * 
    *  @param out the stream to be written
    */
   public void writeExternal(ObjectOutput out)
     throws IOException
   {
     out.writeInt(payload);
     out.writeInt(header);
     out.writeInt(trailer);
     out.writeInt(fragmentCount);
     out.writeInt(byteCount);
     out.writeInt(fragment);
     out.writeInt(position);
     out.writeInt(marked);
 
     int fcount = fragmentCount - 1;
     for (int i = 0; i < fcount; i++) {
       out.write(fragments[i]);
     }
     int lastFragLen = getLastPayload() + header + trailer;
     out.write(fragments[fcount], 0, lastFragLen);
 
 //    System.out.println("writeExternal: " + this.toString(true));
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods inherited from InputStream
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * Reads the next byte of data from the input stream. The value byte is
    * returned as an <code>int</code> in the range <code>0</code> to
    * <code>255</code>. If no byte is available because the end of the stream
    * has been reached, the value <code>-1</code> is returned. This method
    * blocks until input data is available, the end of the stream is detected,
    * or an exception is thrown.
    *
    * @return     the next byte of data, or <code>-1</code> if the end of the
    *             stream is reached.
    * @exception  IOException  if an I/O error occurs.
    */
   public int read()
   {
     if (position == byteCount) {
       /*
        *  The end of the stream has been reached
        */
       return -1;
     }
     
     if (offset == endpayload) {        
       /*
        *  The end of the current fragment has been reached; we move to the next
        *  fragment and we reset the offset.
        */
       fragment++;
       buffer = fragments[fragment];
       offset = header;
     }
     position++;
     return buffer[offset++] & 0xFF;
   }
 
 
   /**
    * Reads some number of bytes from the input stream and stores them into
    * the buffer array <code>b</code>. The number of bytes actually read is
    * returned as an integer.  This method blocks until input data is
    * available, end of file is detected, or an exception is thrown.
    *
    * <p> If <code>b</code> is <code>null</code>, a
    * <code>NullPointerException</code> is thrown.  If the length of
    * <code>b</code> is zero, then no bytes are read and <code>0</code> is
    * returned; otherwise, there is an attempt to read at least one byte. If
    * no byte is available because the stream is at end of file, the value
    * <code>-1</code> is returned; otherwise, at least one byte is read and
    * stored into <code>b</code>.
    *
    * <p> The first byte read is stored into element <code>b[0]</code>, the
    * next one into <code>b[1]</code>, and so on. The number of bytes read is,
    * at most, equal to the length of <code>b</code>. Let <i>k</i> be the
    * number of bytes actually read; these bytes will be stored in elements
    * <code>b[0]</code> through <code>b[</code><i>k</i><code>-1]</code>,
    * leaving elements <code>b[</code><i>k</i><code>]</code> through
    * <code>b[b.length-1]</code> unaffected.
    *
    * <p> The <code>read(b)</code> method for class <code>InputStream</code>
    * has the same effect as: <pre><code> read(b, 0, b.length) </code></pre>
    *
    * @param      b   the buffer into which the data is read.
    * @return     the total number of bytes read into the buffer, or
    *             <code>-1</code> is there is no more data because the end of
    *             the stream has been reached.
    */
   public int read(byte b[]) 
   {
     return read(b, 0, b.length);
   }
 
   /**
    * Reads up to <code>len</code> bytes of data from the input stream into
    * an array of bytes.  An attempt is made to read as many as
    * <code>len</code> bytes, but a smaller number may be read, possibly
    * zero. The number of bytes actually read is returned as an integer.
    *
    * <p> If <code>b</code> is <code>null</code>, a
    * <code>NullPointerException</code> is thrown.
    *
    * <p> If <code>off</code> is negative, or <code>len</code> is negative, or
    * <code>off+len</code> is greater than the length of the array
    * <code>b</code>, then an <code>IndexOutOfBoundsException</code> is
    * thrown.
    *
    * <p> If <code>len</code> is zero, then no bytes are read and
    * <code>0</code> is returned; otherwise, there is an attempt to read at
    * least one byte. If no byte is available because the stream is at end of
    * file, the value <code>-1</code> is returned; otherwise, at least one
    * byte is read and stored into <code>b</code>.
    *
    * <p> The first byte read is stored into element <code>b[off]</code>, the
    * next one into <code>b[off+1]</code>, and so on. The number of bytes read
    * is, at most, equal to <code>len</code>. Let <i>k</i> be the number of
    * bytes actually read; these bytes will be stored in elements
    * <code>b[off]</code> through <code>b[off+</code><i>k</i><code>-1]</code>,
    * leaving elements <code>b[off+</code><i>k</i><code>]</code> through
    * <code>b[off+len-1]</code> unaffected.
    *
    * <p> In every case, elements <code>b[0]</code> through
    * <code>b[off]</code> and elements <code>b[off+len]</code> through
    * <code>b[b.length-1]</code> are unaffected.
    *
    * @param      b     the buffer into which the data is read.
    * @param      off   the start offset in array <code>b</code>
    *                   at which the data is written.
    * @param      len   the maximum number of bytes to read.
    * @return     the total number of bytes read into the buffer, or
    *             <code>-1</code> if there is no more data because the end of
    *             the stream has been reached.
    */
   public int read(byte b[], int off, int len)
   {
     int   tobecopied;
     int   remained;
 
     if (b == null)
       throw new NullPointerException();
     if (off < 0 || off > b.length || len < 0 || (off + len > b.length) || (off + len < 0))
       throw new IndexOutOfBoundsException();
     if (len == 0)
       return 0;
     if (position == byteCount)
       return -1;
 
     if (position + len > byteCount)
       len = byteCount - position;
     if (b.length < len)
       len = b.length;
     position += len;
     remained = len;
     tobecopied = endpayload - offset;
     while (remained > tobecopied) {
       System.arraycopy(fragments[fragment], offset, b, off, tobecopied);
       off += tobecopied;
       remained -= tobecopied;
       fragment++;
       offset = header;
       tobecopied = payload;
     }
     buffer = fragments[fragment];
     System.arraycopy(buffer, offset, b, off, remained);
     offset += remained;
 
     return len;
   }
 
   /**
    * Skips over and discards <code>n</code> bytes of data from this input
    * stream. The <code>skip</code> method may, for a variety of reasons, end
    * up skipping over some smaller number of bytes, possibly <code>0</code>.
    * This may result from any of a number of conditions; reaching end of file
    * before <code>n</code> bytes have been skipped is only one possibility.
    * The actual number of bytes skipped is returned.  If <code>n</code> is
    * negative, no bytes are skipped.
    *
    * <p> The <code>skip</code> method of <code>InputStream</code> creates a
    * byte array and then repeatedly reads into it until <code>n</code> bytes
    * have been read or the end of the stream has been reached. Subclasses are
    * encouraged to provide a more efficient implementation of this method.
    *
    * @param      n   the number of bytes to be skipped.
    * @return     the actual number of bytes skipped.
    * @exception  IOException  if an I/O error occurs.
    */
   public long skip(long n)
   {
     long value = n;
     position += n;
     if (position > byteCount) {
       value -= (position - byteCount);
       position = byteCount;
     }
     return value;
   }
 
   /**
    * Marks the current position in this input stream. A subsequent call to
    * the <code>reset</code> method repositions this stream at the last marked
    * position so that subsequent reads re-read the same bytes.
    *
    * <p> The <code>readlimit</code> arguments tells this input stream to
    * allow that many bytes to be read before the mark position gets
    * invalidated.
    *
    * <p> The general contract of <code>mark</code> is that, if the method
    * <code>markSupported</code> returns <code>true</code>, the stream somehow
    * remembers all the bytes read after the call to <code>mark</code> and
    * stands ready to supply those same bytes again if and whenever the method
    * <code>reset</code> is called.  However, the stream is not required to
    * remember any data at all if more than <code>readlimit</code> bytes are
    * read from the stream before <code>reset</code> is called.
    *
    *
    * @param   readlimit   not used
    */
   public synchronized void mark(int readlimit) 
   {
     marked = position;
   }
 
 
   /**
    * Repositions this stream to the position at the time the
    * <code>mark</code> method was last called on this input stream.
    *
    * <p> The general contract of <code>reset</code> is:
    *
    * <p><ul>
    *
    * <li> If the method <code>markSupported</code> returns
    * <code>true</code>, then:
    *
    *     <ul><li> If the method <code>mark</code> has not been called since
    *     the stream was created, or the number of bytes read from the stream
    *     since <code>mark</code> was last called is larger than the argument
    *     to <code>mark</code> at that last call, then an
    *     <code>IOException</code> might be thrown.
    *
    *     <li> If such an <code>IOException</code> is not thrown, then the
    *     stream is reset to a state such that all the bytes read since the
    *     most recent call to <code>mark</code> (or since the start of the
    *     file, if <code>mark</code> has not been called) will be resupplied
    *     to subsequent callers of the <code>read</code> method, followed by
    *     any bytes that otherwise would have been the next input data as of
    *     the time of the call to <code>reset</code>. </ul>
    *
    * <li> If the method <code>markSupported</code> returns
    * <code>false</code>, then:
    *
    *     <ul><li> The call to <code>reset</code> may throw an
    *     <code>IOException</code>.
    *
    *     <li> If an <code>IOException</code> is not thrown, then the stream
    *     is reset to a fixed state that depends on the particular type of the
    *     input stream and how it was created. The bytes that will be supplied
    *     to subsequent callers of the <code>read</code> method depend on the
    *     particular type of the input stream. </ul></ul>
    *
    * <p> The method <code>reset</code> for class <code>InputStream</code>
    * does nothing and always throws an <code>IOException</code>.
    *
    * @exception  IOException  if this stream has not been marked or if the
    *               mark has been invalidated.
    */
   public synchronized void reset() 
     throws IOException 
   {
     fragment = marked / payload;
     offset = header + marked % payload;
     position = marked;
     buffer = fragments[fragment];
   }
   
   /**
    * Tests if this input stream supports the <code>mark</code> and
    * <code>reset</code> methods. The <code>markSupported</code> method of
    * <code>InputStream</code> returns <code>false</code>.
    *
    * @return  <code>true</code> if this true type supports the mark and reset
    *          method; <code>false</code> otherwise.
    */
   public boolean markSupported() 
   {
     return true;
   }
   
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods from DataInput
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Reads some bytes from an input stream and stores them into the
    *  buffer array <code>b</code>. The number of bytes read is equal to
    *  the length of <code>b</code>.  <p>
    *
    *  This method blocks until one of the following conditions occurs:
    *  <p>
    *
    *  <ul>
    *
    *  <li><code>b.length</code> bytes of input data are available, in
    *  which case a normal return is made.
    *
    *  <li>End of file is detected, in which case an
    *  <code>EOFException</code> is thrown.
    *
    *  <li>An I/O error occurs, in which case an <code>IOException</code>
    *  other than <code>EOFException</code> is thrown.
    *
    *  </ul>
    *  <p>
    *
    *  If <code>b.length</code> is zero, then no bytes are read.
    *  Otherwise, the first byte read is stored into element
    *  <code>b[0]</code>, the next one into <code>b[1]</code>, and so on.
    *  If an exception is thrown from this method, then it may be that
    *  some but not all bytes of <code>b</code> have been updated with
    *  data from the input stream.
    *
    *  @param b
    *    The buffer into which the data is read.
    *  @exception EOFException
    *    If this stream reaches the end before reading all the bytes.
    *  @exception IOException
    *    If an I/O error occurs.
    *  @exception NullPointerException
    *    Thrown if <code>b</code> is <code>null</code>.
    */
   public void readFully(byte b[]) throws IOException
   {
     read(b);
   }
   
   /**
    *
    * Reads <code>len</code>
    * bytes from
    * an input stream.
    * <p>
    * This method
    * blocks until one of the following conditions
    * occurs:<p>
    * <ul>
    * <li><code>len</code> bytes
    * of input data are available, in which case
    * a normal return is made.
    *
    * <li>End of file
    * is detected, in which case an <code>EOFException</code>
    * is thrown.
    *
    * <li>An I/O error occurs, in
    * which case an <code>IOException</code> other
    * than <code>EOFException</code> is thrown.
    * </ul>
    * <p>
    * If <code>b</code> is <code>null</code>,
    * a <code>NullPointerException</code> is thrown.
    * If <code>off</code> is negative, or <code>len</code>
    * is negative, or <code>off+len</code> is
    * greater than the length of the array <code>b</code>,
    * then an <code>IndexOutOfBoundsException</code>
    * is thrown.
    * If <code>len</code> is zero,
    * then no bytes are read. Otherwise, the first
    * byte read is stored into element <code>b[off]</code>,
    * the next one into <code>b[off+1]</code>,
    * and so on. The number of bytes read is,
    * at most, equal to <code>len</code>.
    *
    * @param     b   the buffer into which the data is read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public void readFully(byte b[], int off, int len) throws IOException
   {
     read(b, off, len);
   }
   
   /**
    * Makes an attempt to skip over
    * <code>n</code> bytes
    * of data from the input
    * stream, discarding the skipped bytes. However,
    * it may skip
    * over some smaller number of
    * bytes, possibly zero. This may result from
    * any of a
    * number of conditions; reaching
    * end of file before <code>n</code> bytes
    * have been skipped is
    * only one possibility.
    * This method never throws an <code>EOFException</code>.
    * The actual
    * number of bytes skipped is returned.
    *
    * @param      n   the number of bytes to be skipped.
    * @return     the number of bytes skipped, which is always <code>n</code>.
    * @exception  EOFException  if this stream reaches the end before skipping
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public int skipBytes(int n) throws IOException
   {
     return (int) skip(n);
   }
   
   /**
    * Reads one input byte and returns
    * <code>true</code> if that byte is nonzero,
    * <code>false</code> if that byte is zero.
    * This method is suitable for reading
    * the byte written by the <code>writeBoolean</code>
    * method of interface <code>DataOutput</code>.
    *
    * @return     the <code>boolean</code> value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public boolean readBoolean() throws IOException
   {
     int ch = read();
     if (ch < 0)
       throw new EOFException();
     return (ch != 0);
   }
   
   /**
    * Reads and returns one input byte.
    * The byte is treated as a signed value in
    * the range <code>-128</code> through <code>127</code>,
    * inclusive.
    * This method is suitable for
    * reading the byte written by the <code>writeByte</code>
    * method of interface <code>DataOutput</code>.
    *
    * @return     the 8-bit value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public byte readByte() throws IOException
   {
     int ch = read();
     if (ch < 0)
       throw new EOFException();
     return (byte)(ch);
   }
   
   /**
    * Reads one input byte, zero-extends
    * it to type <code>int</code>, and returns
    * the result, which is therefore in the range
    * <code>0</code>
    * through <code>255</code>.
    * This method is suitable for reading
    * the byte written by the <code>writeByte</code>
    * method of interface <code>DataOutput</code>
    * if the argument to <code>writeByte</code>
    * was intended to be a value in the range
    * <code>0</code> through <code>255</code>.
    *
    * @return     the unsigned 8-bit value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public int readUnsignedByte() throws IOException
   {
     int ch = read();
     if (ch < 0)
       throw new EOFException();
     return ch;
   }
   
   /**
    * Reads two input bytes and returns
    * a <code>short</code> value. Let <code>a</code>
    * be the first byte read and <code>b</code>
    * be the second byte. The value
    * returned
    * is:
    * <p><pre><code>(short)((a &lt;&lt; 8) * | (b &amp; 0xff))
    * </code></pre>
    * This method
    * is suitable for reading the bytes written
    * by the <code>writeShort</code> method of
    * interface <code>DataOutput</code>.
    *
    * @return     the 16-bit value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public short readShort() throws IOException
   {
     return (short) read2();
   }
   
   /**
    * Reads two input bytes and returns
    * an <code>int</code> value in the range <code>0</code>
    * through <code>65535</code>. Let <code>a</code>
    * be the first byte read and
    * <code>b</code>
    * be the second byte. The value returned is:
    * <p><pre><code>(((a &amp; 0xff) &lt;&lt; 8) | (b &amp; 0xff))
    * </code></pre>
    * This method is suitable for reading the bytes
    * written by the <code>writeShort</code> method
    * of interface <code>DataOutput</code>  if
    * the argument to <code>writeShort</code>
    * was intended to be a value in the range
    * <code>0</code> through <code>65535</code>.
    *
    * @return     the unsigned 16-bit value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public int readUnsignedShort() throws IOException
   {
     return read2();
   }
   
   /**
    * Reads an input <code>char</code> and returns the <code>char</code> value.
    * A Unicode <code>char</code> is made up of two bytes.
    * Let <code>a</code>
    * be the first byte read and <code>b</code>
    * be the second byte. The value
    * returned is:
    * <p><pre><code>(char)((a &lt;&lt; 8) | (b &amp; 0xff))
    * </code></pre>
    * This method
    * is suitable for reading bytes written by
    * the <code>writeChar</code> method of interface
    * <code>DataOutput</code>.
    *
    * @return     the Unicode <code>char</code> read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public char readChar() throws IOException
   {
     return (char) read2();
   }
   
   /**
    * Reads four input bytes and returns an
    * <code>int</code> value. Let <code>a</code>
    * be the first byte read, <code>b</code> be
    * the second byte, <code>c</code> be the third
    * byte,
    * and <code>d</code> be the fourth
    * byte. The value returned is:
    * <p><pre>
    * <code>
    * (((a &amp; 0xff) &lt;&lt; 24) | ((b &amp; 0xff) &lt;&lt; 16) |
    * &#32;((c &amp; 0xff) &lt;&lt; 8) | (d &amp; 0xff))
    * </code></pre>
    * This method is suitable
    * for reading bytes written by the <code>writeInt</code>
    * method of interface <code>DataOutput</code>.
    *
    * @return     the <code>int</code> value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public int readInt() throws IOException
   {
     return read4();
   }
   
   /**
    * Reads eight input bytes and returns
    * a <code>long</code> value. Let <code>a</code>
    * be the first byte read, <code>b</code> be
    * the second byte, <code>c</code> be the third
    * byte, <code>d</code>
    * be the fourth byte,
    * <code>e</code> be the fifth byte, <code>f</code>
    * be the sixth byte, <code>g</code> be the
    * seventh byte,
    * and <code>h</code> be the
    * eighth byte. The value returned is:
    * <p><pre> <code>
    * (((long)(a &amp; 0xff) &lt;&lt; 56) |
    *  ((long)(b &amp; 0xff) &lt;&lt; 48) |
    *  ((long)(c &amp; 0xff) &lt;&lt; 40) |
    *  ((long)(d &amp; 0xff) &lt;&lt; 32) |
    *  ((long)(e &amp; 0xff) &lt;&lt; 24) |
    *  ((long)(f &amp; 0xff) &lt;&lt; 16) |
    *  ((long)(g &amp; 0xff) &lt;&lt;  8) |
    *  ((long)(h &amp; 0xff)))
    * </code></pre>
    * <p>
    * This method is suitable
    * for reading bytes written by the <code>writeLong</code>
    * method of interface <code>DataOutput</code>.
    *
    * @return     the <code>long</code> value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public long readLong() throws IOException
   {
     return read8();
   }
   
   /**
    * Reads four input bytes and returns
    * a <code>float</code> value. It does this
    * by first constructing an <code>int</code>
    * value in exactly the manner
    * of the <code>readInt</code>
    * method, then converting this <code>int</code>
    * value to a <code>float</code> in
    * exactly the manner of the method <code>Float.intBitsToFloat</code>.
    * This method is suitable for reading
    * bytes written by the <code>writeFloat</code>
    * method of interface <code>DataOutput</code>.
    *
    * @return     the <code>float</code> value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public float readFloat() throws IOException
   {
     return Float.intBitsToFloat(read4());
   }
   
   /**
    * Reads eight input bytes and returns
    * a <code>double</code> value. It does this
    * by first constructing a <code>long</code>
    * value in exactly the manner
    * of the <code>readlong</code>
    * method, then converting this <code>long</code>
    * value to a <code>double</code> in exactly
    * the manner of the method <code>Double.longBitsToDouble</code>.
    * This method is suitable for reading
    * bytes written by the <code>writeDouble</code>
    * method of interface <code>DataOutput</code>.
    *
    * @return     the <code>double</code> value read.
    * @exception  EOFException  if this stream reaches the end before reading
    *         all the bytes.
    * @exception  IOException   if an I/O error occurs.
    */
   public double readDouble() throws IOException
   {
     return Double.longBitsToDouble(read8());
   }
   
   private char lineBuffer[];
 
   /**
    * Reads the next line of text from the input stream.
    * It reads successive bytes, converting
    * each byte separately into a character,
    * until it encounters a line terminator or
    * end of
    * file; the characters read are then
    * returned as a <code>String</code>. Note
    * that because this
    * method processes bytes,
    * it does not support input of the full Unicode
    * character set.
    * <p>
    * If end of file is encountered
    * before even one byte can be read, then <code>null</code>
    * is returned. Otherwise, each byte that is
    * read is converted to type <code>char</code>
    * by zero-extension. If the character <code>'\n'</code>
    * is encountered, it is discarded and reading
    * ceases. If the character <code>'\r'</code>
    * is encountered, it is discarded and, if
    * the following byte converts &#32;to the
    * character <code>'\n'</code>, then that is
    * discarded also; reading then ceases. If
    * end of file is encountered before either
    * of the characters <code>'\n'</code> and
    * <code>'\r'</code> is encountered, reading
    * ceases. Once reading has ceased, a <code>String</code>
    * is returned that contains all the characters
    * read and not discarded, taken in order.
    * Note that every character in this string
    * will have a value less than <code>&#92;u0100</code>,
    * that is, <code>(char)256</code>.
    *
    * @return     if this stream reaches the end before reading all the bytes.
    * @exception  IOException  if an I/O error occurs.
    */
   public String readLine() throws IOException
   {
     PushbackInputStream in = new PushbackInputStream(this);
     
     char buf[] = lineBuffer;
     
     if (buf == null) {
       buf = lineBuffer = new char[128];
     }
     
     int room = buf.length;
     int offset = 0;
     int c;
     
     loop: while (true) {
         switch (c = in.read()) {
         case -1:
         case '\n':
           break loop;
           
         case '\r':
           int c2 = in.read();
           if ((c2 != '\n') && (c2 != -1)) {
             in.unread(c2);
           }
           break loop;
           
         default:
           if (--room < 0) {
             buf = new char[offset + 128];
             room = buf.length - offset - 1;
             System.arraycopy(lineBuffer, 0, buf, 0, offset);
             lineBuffer = buf;
           }
           buf[offset++] = (char) c;
           break;
         }
       }
     if ((c == -1) && (offset == 0)) {
       return null;
     }
     return String.copyValueOf(buf, 0, offset);
   }
   
   /**
    * Reads in a string that has been encoded using a modified UTF-8 format.
    * The general contract of <code>readUTF</code>
    * is that it reads a representation of a Unicode
    * character string encoded in Java modified
    * UTF-8 format; this string of characters
    * is then returned as a <code>String</code>.
    * <p>
    * First, two bytes are read and used to
    * construct an unsigned 16-bit integer in
    * exactly the manner of the <code>readUnsignedShort</code>
    * method . This integer value is called the
    * <i>UTF length</i> and specifies the number
    * of additional bytes to be read. These bytes
    * are then converted to characters by considering
    * them in groups. The length of each group
    * is computed from the value of the first
    * byte of the group. The byte following a
    * group, if any, is the first byte of the
    * next group.
    * <p>
    * If the first byte of a group
    * matches the bit pattern <code>0xxxxxxx</code>
    * (where <code>x</code> means "may be <code>0</code>
    * or <code>1</code>"), then the group consists
    * of just that byte. The byte is zero-extended
    * to form a character.
    * <p>
    * If the first byte
    * of a group matches the bit pattern <code>110xxxxx</code>,
    * then the group consists of that byte <code>a</code>
    * and a second byte <code>b</code>. If there
    * is no byte <code>b</code> (because byte
    * <code>a</code> was the last of the bytes
    * to be read), or if byte <code>b</code> does
    * not match the bit pattern <code>10xxxxxx</code>,
    * then a <code>UTFDataFormatException</code>
    * is thrown. Otherwise, the group is converted
    * to the character:<p>
    * <pre><code>(char)(((a&amp; 0x1F) &lt;&lt; 6) | (b &amp; 0x3F))
    * </code></pre>
    * If the first byte of a group
    * matches the bit pattern <code>1110xxxx</code>,
    * then the group consists of that byte <code>a</code>
    * and two more bytes <code>b</code> and <code>c</code>.
    * If there is no byte <code>c</code> (because
    * byte <code>a</code> was one of the last
    * two of the bytes to be read), or either
    * byte <code>b</code> or byte <code>c</code>
    * does not match the bit pattern <code>10xxxxxx</code>,
    * then a <code>UTFDataFormatException</code>
    * is thrown. Otherwise, the group is converted
    * to the character:<p>
    * <pre><code>
    * (char)(((a &amp; 0x0F) &lt;&lt; 12) | ((b &amp; 0x3F) &lt;&lt; 6) | (c &amp; 0x3F))
    * </code></pre>
    * If the first byte of a group matches the
    * pattern <code>1111xxxx</code> or the pattern
    * <code>10xxxxxx</code>, then a <code>UTFDataFormatException</code>
    * is thrown.
    * <p>
    * If end of file is encountered
    * at any time during this entire process,
    * then an <code>EOFException</code> is thrown.
    * <p>
    * After every group has been converted to
    * a character by this process, the characters
    * are gathered, in the same order in which
    * their corresponding groups were read from
    * the input stream, to form a <code>String</code>,
    * which is returned.
    * <p>
    * The <code>writeUTF</code>
    * method of interface <code>DataOutput</code>
    * may be used to write data that is suitable
    * for reading by this method.
    * @return     a Unicode string.
    * @exception  EOFException      if this stream reaches the end
    *         before reading all the bytes.
    * @exception  IOException       if an I/O error occurs.
    * @exception  UTFDataFormatException  if the bytes do not represent a
    *         valid UTF-8 encoding of a string.
    */
   public String readUTF() throws IOException
   {
     int utflen = readUnsignedShort();
     StringBuilder str = new StringBuilder(utflen);
     byte bytearr [] = new byte[utflen];
     int c, char2, char3;
     int count = 0;
     
     readFully(bytearr, 0, utflen);
     
     while (count < utflen) {
       c = bytearr[count] & 0xff;
       switch (c >> 4) {
       case 0: case 1: case 2: case 3: case 4: case 5: case 6: case 7:
         /* 0xxxxxxx*/
         count++;
         str.append((char)c);
         break;
       case 12: case 13:
         /* 110x xxxx   10xx xxxx*/
         count += 2;
         if (count > utflen)
           throw new UTFDataFormatException();
         char2 = bytearr[count-1];
         if ((char2 & 0xC0) != 0x80)
           throw new UTFDataFormatException(); 
         str.append((char)(((c & 0x1F) << 6) | (char2 & 0x3F)));
         break;
       case 14:
         /* 1110 xxxx  10xx xxxx  10xx xxxx */
         count += 3;
         if (count > utflen)
           throw new UTFDataFormatException();
         char2 = bytearr[count-2];
         char3 = bytearr[count-1];
         if (((char2 & 0xC0) != 0x80) || ((char3 & 0xC0) != 0x80))
           throw new UTFDataFormatException();   
         str.append((char)(((c     & 0x0F) << 12) |
           ((char2 & 0x3F) << 6)  |
           ((char3 & 0x3F) << 0)));
         break;
       default:
         /* 10xx xxxx,  1111 xxxx */
         throw new UTFDataFormatException();     
       }
     }
     // The number of chars produced may be less than utflen
     return new String(str);
   }
   
   
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods from ObjectInput
   ////////////////////////////////////////////////////////////////////////////////////////////
   
 
   /**
    * Read and return an object. The class that implements this interface
    * defines where the object is "read" from.
    *
    * @exception java.lang.ClassNotFoundException If the class of a serialized
    *      object cannot be found.
    * @exception IOException If any of the usual Input/Output
    * related exceptions occur.
    */
   public Object readObject()
     throws ClassNotFoundException, IOException
   {
     // workaround since InMessage don't provide native support for readObject()
     ObjectInputStream ois = new ObjectInputStream(this);
     return ois.readObject();
   }
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Public methods
   ////////////////////////////////////////////////////////////////////////////////////////////
   
   /**
    *  Move the current position in the stream to the specified
    *  <code>position</code>. <p>
    *
    *  @exception IndexOutOfBoundsException
    *    Raised if the new position in the message exceeds the maximum
    *    length.
    */
   public void seek(int newpos)
   {
     if (newpos > byteCount) {
       throw new IndexOutOfBoundsException("New position (" + newpos
         + ") exceeds maximum length:" + byteCount);
     }
     fragment = newpos / payload;
     offset = header + newpos % payload;
     position = newpos;
     buffer = fragments[fragment];
   }
 
   /**
    * Read a short integer from the stream at the current offset.
    * <p>
    * If the message ends before having the possibility to read the
    * value, a <tt>ArrayIndexOutOfBoundsException</tt> is thrown.
    */
   private int read2()
     throws IOException
   {
     position += 2;
     if (position > byteCount) {
       position = byteCount;
       throw new EOFException("pos="+position+" > byteCount="+byteCount);
     }
     if (offset + 2 > endpayload) {
       readTemp(2);
       return ((temp[0] << 8) & 0xFF00) | (temp[1] & 0xFF);
     } else {
       return ((buffer[offset++] << 8) & 0xFF00) | (buffer[offset++] & 0xFF);
     }
   }
 
 
   /**
    *  Read an integer from the stream at the current offset.
    *  <p>
    *
    *  If the message ends before all four bytes of the <code>int</code>
    *  value can be read, a <code>EOFException</code> is thrown.
    */
   private int read4()
     throws IOException
   {
     position += 4;
     if (position > byteCount) {
       position = byteCount;
       throw new EOFException("pos: "+ position + ", byteCount: " + byteCount + ", offset: " + offset);
     }
     if (offset + 4 > endpayload) {
       readTemp(4);
       return ((temp[0] << 24) & 0xFF000000) | ((temp[1] << 16)
                 & 0xFF0000) | ((temp[2] << 8) & 0xFF00) | (temp[3] & 0xFF);
     } else {
       return ((buffer[offset++] << 24) & 0xFF000000) | ((buffer[offset++] << 16)
                 & 0xFF0000) | ((buffer[offset++] << 8) & 0xFF00) | (buffer[offset++] & 0xFF);
     }
   }
 
 
   /**
    * Read a long integer from the stream at the current offset.
    * <p>
    * If the message ends before having the possibility to read the
    * value, a <tt>ArrayIndexOutOfBoundsException</tt> is thrown.
    */
   private long read8()
     throws IOException
   {
     int v1, v2;
     
     position += 8;
     if (position > byteCount) {
       position = byteCount;
       throw new EOFException();
     }
     if (offset + 8 > endpayload) {
       readTemp(8);
       v1 = ((temp[0] << 24) & 0xFF000000) | ((temp[1] << 16) & 0xFF0000) | 
            ((temp[2] << 8) & 0xFF00) | (temp[3] & 0xFF);
       v2 = ((temp[4] << 24) & 0xFF000000) | ((temp[5] << 16) & 0xFF0000) | 
            ((temp[6] << 8) & 0xFF00) | (temp[7] & 0xFF);
     } else {
       v1 = ((buffer[offset++] << 24) & 0xFF000000) | ((buffer[offset++] << 16)
            & 0xFF0000) | ((buffer[offset++] << 8) & 0xFF00) | (buffer[offset++] & 0xFF);
       v2 = ((buffer[offset++] << 24) & 0xFF000000) | ((buffer[offset++] << 16)
            & 0xFF0000) | ((buffer[offset++] << 8) & 0xFF00) | (buffer[offset++] & 0xFF);
     }
     return ((long)(v1) << 32) | (v2 & 0xFFFFFFFFL);
   }
 
 
   /**
    *  Compares the bytes contained in this InMessage with those
    *  contained in another InMessage. Starting from position
    *  <t>pos1</t> in this message and from position <t>pos2</t>
    *  from the other message, compares <t>len</t> bytes and
    *  returns false if they differ.
    *
    *  @param pos1 starting position in this message
    *  @param msg the meessage to be compared
    *  @param pos2 starting position in <t>msg</t>
    *  @param len the number of bytes to be compared
    */
   public boolean compare(int pos1, InMessage msg, int pos2, int len)
   {
     /* 
      *  Verify that comparison arguments are regular
      *  FIX 16/11/2000: Fixed the check by removing unnecessary (and wrong)
      *  comparison between pos2+len and byteCount
      */
     if (pos1 >= byteCount || pos2 >= msg.byteCount)
       throw new IllegalArgumentException("Pos1: " + pos1 + "Pos2: " + pos2 + " byteCount: " + byteCount + " len " + len);
 
     /*
      *  Compute start positions for comparison
      */
     int frag1 = pos1 / payload;
     int frag2 = pos2 / msg.payload;
     int index1 = header + pos1 % payload;
     int index2 = msg.header + pos2 % msg.payload;
 
     for (int i = 0; i < len; i++) {
       /*
        *  Stop comparison if bytes in byte arrays differ
        */
       if (fragments[frag1][index1] != msg.fragments[frag2][index2]) 
         return false;
 
       /*
        *  Go to the next byte, moving to the next fragment if necessary
        */
       index1++;
       index2++;
       if (index1 == endpayload) {
         index1 = header;
         frag1++;
       }
       if (index2 == msg.endpayload) {
         index2 = header;
         frag2++;
       }
     }
 
     /*
      *  The byte arrays are equal.
      */
     return true;
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Package methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * Insert a new fragment in the message whose payload field is
    * <tt>blen</tt> bytes long.
    */
   public void insert(byte[] data, int blen)
   {
     byteCount += blen;
     if (fragmentCount == fragments.length) {
       // Double the space
       byte[][] btemp = new byte[fragments.length*2][];
       System.arraycopy(fragments, 0, btemp, 0, fragments.length);
       fragments = btemp;
     }
     fragments[fragmentCount] = data;
     if (fragmentCount == 0)
       buffer = fragments[0];
     fragmentCount++;
   }
 
 
   /**
    *  Returns the total number of bytes stored in this message output stream.
    */
   public int getByteCount()
   {
     return byteCount;
   }
 
 
   /**
    *  Returns the current position in the message.
    */
   public int getPosition()
   {
     return position;
   }
 
 
   /**
    *  Returns the number of fragments composing the message.
    */
   public int getFragmentCount()
   {
     return fragmentCount;
   }
 
  
   public byte[][] getFragments()
   {
     return fragments;
   }
 
 
   /**
    *  Returns the length of the payload field for this message.
    */
   public int getPayload()
   {
     return payload;
   }
 
   /**
    *  Returns the length of the payload of the last fragment composing
    *  the message.
    */
   public int getLastPayload()
   {
     if (byteCount % payload == 0)
       return payload;
     else
       return (byteCount % payload);
   }
 
   /**
    *
    */
   public int getHeader()
   {
     return header;
   }
 
   /**
    *
    */
   public int getTrailer()
   {
     return trailer;
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Private methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Copy <tt>size</tt> bytes in the temp array; this function is invoked when
    *  the bytes read are part in fragment and part in the following one.
    */
   private void readTemp(int size)
   {
     for (int i = 0; i < size; i++) {
       if (offset == endpayload) {
         fragment++;
         buffer = fragments[fragment];
         offset = header;
       }
       temp[i] = buffer[offset++];
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Object methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Returns a string representation of this object
    */
   public String toString()
   {
     StringBuilder buf = new StringBuilder();
     buf.append("[InMessage: byteCnt=");
     buf.append(byteCount);
     buf.append(", payload=");
     buf.append(payload);
     buf.append(", header=");
     buf.append(header);
     buf.append(", trailer=");
     buf.append(trailer);
     buf.append(", totlen=");
     buf.append(totlen);
     buf.append(", endpayload=");
     buf.append(endpayload);
     buf.append(", position=");
     buf.append(position);
     buf.append(", fragment=");
     buf.append(fragment);
     buf.append(", offset=");
     buf.append(offset);
     buf.append(", fragmentCount=");
     buf.append(fragmentCount);
     buf.append(", marked=");
     buf.append(marked);
     buf.append("]");
 
     if (ConfigManager.logMsgContent) {
       buf.append(": ");
     } else {
       // Return without logging message content
       return buf.toString();
     }
 
     for (int i = 0; i < fragmentCount-1; i++) {
       for (int j = header; j < endpayload; j++) {
         buf.append(" ");
         if (j-header+1 == position) {
           buf.append(">> ");
         }
         buf.append(Util.byte2str(fragments[i][j]));
         if (j-header+1 == position) {
           buf.append(" << ");
         }
       }
     }
 
     if (fragmentCount > 0) {
       /*
        * Print the content of the fragments, assuming that the InMessage
        * object has been initialized; that is fragmentCount > 0.
        */
       int stop = byteCount - (fragmentCount-1)*payload + header;
       int lastFragStop = getLastPayload() + header;
 
       for (int j = header; j < stop; j++) {
         buf.append(" ");
         if (j-header+1 == position) {
           buf.append(">> ");
         }
         buf.append(Util.byte2str(fragments[fragmentCount-1][j]));
         if (j-header+1 == position) {
           buf.append(" << ");
         }
       }
       buf.append(" --- stop=");
       buf.append(stop);
       buf.append(", lastFragStop=");
       buf.append(lastFragStop);
     }
 
     return buf.toString();
   }
 
 } // END InMessage