Rewritten to support I/O via channels.

git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@1957 542714f4-19e9-0310-aa3c-eee0fc999fb1

src/java/com/threerings/io/FramedInputStream.java

@@ -1,5 +1,5 @@
 //
-// $Id: FramedInputStream.java,v 1.1 2002/07/23 05:42:34 mdb Exp $
+// $Id: FramedInputStream.java,v 1.2 2002/11/18 18:51:33 mdb Exp $
 
 package com.threerings.io;
 
@@ -7,8 +7,8 @@ import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 
-import com.samskivert.util.StringUtil;
+import java.nio.ByteBuffer;
-import com.threerings.presents.Log;
+import java.nio.channels.SocketChannel;
 
 /**
  * The framed input stream reads input that was framed by a framing output
@@ -37,81 +37,125 @@ import com.threerings.presents.Log;
  */
 public class FramedInputStream extends InputStream
 {
+    /**
+     * Creates a new framed input stream.
+     */
     public FramedInputStream ()
     {
-        _header = new byte[HEADER_SIZE];
+        _buffer = ByteBuffer.allocate(INITIAL_BUFFER_CAPACITY);
-        _buffer = new byte[INITIAL_BUFFER_SIZE];
     }
 
     /**
-     * Reads a frame from the provided input stream, or appends to a
+     * Reads a frame from the provided channel, appending to any partially
-     * partially read frame. Appends the read data to the existing data
+     * read frame. If the entire frame data is not yet available,
-     * available via the framed input stream's read methods. If the entire
+     * <code>readFrame</code> will return false, otherwise true.
-     * frame data is not yet available, <code>readFrame</code> will return
-     * false, otherwise true.
      *
-     * <p> The code assumes that it will be able to read the entire frame
+     * <p> <em>Note:</em> when this method returns true, it is required
-     * header in a single read. The header is only four bytes and should
+     * that the caller read <em>all</em> of the frame data from the stream
-     * always arrive at the beginning of a packet, so unless something is
+     * before again calling {@link #readFrame} as the previous frame's
-     * very funky with the networking layer, this should be a safe
+     * data will be elimitated upon the subsequent call.
-     * assumption.
      *
      * @return true if the entire frame has been read, false if the buffer
      * contains only a partial frame.
      */
-    public boolean readFrame (InputStream source)
+    public boolean readFrame (SocketChannel source)
         throws IOException
     {
-        // if the buffer currently contains a complete frame, that means
+        // flush data from any previous frame from the buffer
-        // we're not halfway through reading a frame and that we can start
+        if (_buffer.limit() == _length) {
-        // anew.
+            // this will remove the old frame's bytes from the buffer,
-        if (_count == _length) {
+            // shift our old data to the start of the buffer, position the
-            // read in the frame length
+            // buffer appropriately for appending new data onto the end of
-            int got = source.read(_header, 0, HEADER_SIZE);
+            // our existing data, and set the limit to the capacity
-            if (got < 0) {
+            _buffer.limit(_have);
+            _buffer.position(_length);
+            _buffer.compact();
+            _have -= _length;
+
+            // we may have picked up the next frame in a previous read, so
+            // try decoding the length straight away
+            _length = decodeLength();
+        }
+
+        // we may already have the next frame entirely in the buffer from
+        // a previous read
+        if (checkForCompleteFrame()) {
+            return true;
+        }
+
+        // read whatever data we can from the source
+        do {
+            int got = source.read(_buffer);
+            if (got == -1) {
                 throw new EOFException();
+            }
+            _have += got;
 
-            } else if (got == 0) {
+            // if there's room remaining in the buffer, that means we've
-                // TBD: don't log this for now, but look into it later
+            // read all there is to read, so we can move on to inspecting
-                // Log.info("Woke up to read data, but there ain't none. Sigh.");
+            // what we've got
-                return false;
+            if (_buffer.remaining() > 0) {
-
+                break;
-            } else if (got < HEADER_SIZE) {
-                String errmsg = "FramedInputStream does not support " +
-                    "partially reading the header. Needed " + HEADER_SIZE +
-                    " bytes, got " + got + " bytes.";
-                throw new RuntimeException(errmsg);
             }
 
-            // now that we've read our new frame length, we can clear out
+            // otherwise, we've filled up our buffer as a result of this
-            // any prior data
+            // read, expand it and try reading some more
-            _pos = 0;
+            ByteBuffer newbuf = ByteBuffer.allocate(_buffer.capacity() << 1);
-            _count = 0;
+            newbuf.put((ByteBuffer)_buffer.flip());
+            _buffer = newbuf;
 
-            // decode the frame length
+            // don't let things grow without bounds
-            _length = (_header[0] & 0xFF) << 24;
+        } while (_buffer.capacity() < MAX_BUFFER_CAPACITY);
-            _length += (_header[1] & 0xFF) << 16;
-            _length += (_header[2] & 0xFF) << 8;
-            _length += (_header[3] & 0xFF);
 
-            // if necessary, expand our buffer to accomodate the frame
+        // if we didn't already have our length, see if we now have enough
-            if (_length > _buffer.length) {
+        // data to obtain it
-                // increase the buffer size in large increments
+        if (_length == -1) {
-                _buffer = new byte[Math.max(_buffer.length << 1, _length)];
+            _length = decodeLength();
-            }
         }
 
-        // read the data into the buffer
+        // finally check to see if there's a complete frame in the buffer
-        int got = source.read(_buffer, _count, _length-_count);
+        // and prepare to serve it up if there is
-        if (got < 0) {
+        return checkForCompleteFrame();
-            throw new EOFException();
+    }
+
+    /**
+     * Decodes and returns the length of the current frame from the buffer
+     * if possible. Returns -1 otherwise.
+     */
+    protected final int decodeLength ()
+    {
+        // if we don't have enough bytes to determine our frame size, stop
+        // here and let the caller know that we're not ready
+        if (_have < HEADER_SIZE) {
+            return -1;
         }
-        _count += got;
 
-//         System.err.println("Read frame " + _count +
+        // decode the frame length
-//                            " (want " + _length + " pos " + _pos + ")");
+        _buffer.rewind();
+        int length = (_buffer.get() & 0xFF) << 24;
+        length += (_buffer.get() & 0xFF) << 16;
+        length += (_buffer.get() & 0xFF) << 8;
+        length += (_buffer.get() & 0xFF);
+        _buffer.position(_have);
 
-        return (_count == _length);
+        return length;
+    }
+
+    /**
+     * Returns true if a complete frame is in the buffer, false otherwise.
+     * If a complete frame is in the buffer, the buffer will be prepared
+     * to deliver that frame via our {@link InputStream} interface.
+     */
+    protected final boolean checkForCompleteFrame ()
+    {
+        if (_length == -1 || _have < _length) {
+            return false;
+        }
+
+        // prepare the buffer such that this frame can be read
+        _buffer.position(HEADER_SIZE);
+        _buffer.limit(_length);
+        return true;
     }
 
     /**
@@ -127,7 +171,7 @@ public class FramedInputStream extends InputStream
      */
     public int read ()
     {
-        return (_pos < _count) ? (_buffer[_pos++] & 0xFF) : -1;
+        return (_buffer.remaining() > 0) ? (_buffer.get() & 0xFF) : -1;
     }
 
     /**
@@ -155,32 +199,21 @@ public class FramedInputStream extends InputStream
      */
     public int read (byte[] b, int off, int len)
     {
-        // sanity check the arguments
+        // if they want no bytes, we give them no bytes; this is
-	if (b == null) {
+        // purportedly the right thing to do regardless of whether we're
-	    throw new NullPointerException();
+        // at EOF or not
-	} else if ((off < 0) || (off > b.length) || (len < 0) ||
+        if (len == 0) {
-		   ((off + len) > b.length) || ((off + len) < 0)) {
+            return 0;
-	    throw new IndexOutOfBoundsException();
+        }
-	}
 
-        // figure out how much data we'll return
+        // trim the amount to be read to what is available; if they wanted
-	if (_pos >= _count) {
+        // bytes and we have none, return -1 to indicate EOF
-            // if they asked to read zero bytes and we have no bytes
+        if ((len = Math.min(len, _buffer.remaining())) == 0) {
-            // remaining; we're supposed to return 0 rather than EOF
+            return -1;
-	    return (len == 0) ? 0 : -1;
+        }
-	}
-	if (_pos + len > _count) {
-	    len = _count - _pos;
-	}
-	if (len <= 0) {
-	    return 0;
-	}
 
-        // copy and advance
+        _buffer.get(b, off, len);
-	System.arraycopy(_buffer, _pos, b, off, len);
+        return len;
-	_pos += len;
-
-	return len;
     }
 
     /**
@@ -197,28 +230,19 @@ public class FramedInputStream extends InputStream
      */
     public long skip (long n)
     {
-	if (_pos + n > _count) {
+        throw new UnsupportedOperationException();
-	    n = _count - _pos;
-	}
-	if (n <= 0) {
-	    return 0;
-	}
-	_pos += n;
-	return n;
     }
 
     /**
      * Returns the number of bytes that can be read from this input stream
-     * without blocking. The value returned is <code>count - pos</code>,
+     * without blocking.
-     * which is the number of bytes remaining to be read from the input
-     * buffer.
      *
      * @return the number of bytes remaining to be read from the buffered
-     * frames.
+     * frame.
      */
     public int available ()
     {
-	return _count - _pos;
+	return _buffer.remaining();
     }
 
     /**
@@ -243,19 +267,26 @@ public class FramedInputStream extends InputStream
      */
     public void reset ()
     {
-	_pos = 0;
+        // position our buffer at the beginning of the frame data
+        _buffer.position(HEADER_SIZE);
     }
 
-    protected byte[] _header;
+    /** The buffer in which we maintain our frame data. */
-    protected int _length;
+    protected ByteBuffer _buffer;
 
-    protected byte[] _buffer;
+    /** The length of the current frame being read. */
-    protected int _pos;
+    protected int _length = -1;
-    protected int _count;
+
+    /** The number of bytes total that we have in our buffer (these bytes
+     * may comprise more than one frame. */
+    protected int _have = 0;
 
     /** The size of the frame header (a 32-bit integer). */
     protected static final int HEADER_SIZE = 4;
 
     /** The default initial size of the internal buffer. */
-    protected static final int INITIAL_BUFFER_SIZE = 32;
+    protected static final int INITIAL_BUFFER_CAPACITY = 32;
+
+    /** No need to get out of hand. */
+    protected static final int MAX_BUFFER_CAPACITY = 512 * 1024;
 }

src/java/com/threerings/io/FramingOutputStream.java

@@ -1,5 +1,5 @@
 //
-// $Id: FramingOutputStream.java,v 1.2 2002/11/05 02:16:46 mdb Exp $
+// $Id: FramingOutputStream.java,v 1.3 2002/11/18 18:51:33 mdb Exp $
 
 package com.threerings.io;
 
@@ -7,29 +7,29 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+
 /**
  * The framing output stream accumulates output into a byte array just
  * like the byte array output stream, but can then be instructed to send
- * its contents down another output stream, prefixed by the length
+ * its contents over a channel, prefixed by the length (written as an
- * (written as an integer) of those contents. It does this efficiently so
+ * integer) of the entire frame (contents plus length prefix). It does
- * that data is copied as little as possible and so that the output stream
+ * this efficiently so that data is copied as little as possible and so
- * to which the data is written need not be buffered because the framed
+ * that the output stream to which the data is written need not be
- * output is written in a single call to <code>write()</code>.
+ * buffered because the framed output is written in a single call to
+ * <code>write()</code>.
  *
  * <p><em>Note:</em> The framing output stream does not synchronize writes
  * to its internal buffer. It is intended to only be accessed from a
  * single thread.
- *
- * <p>Implementation note: maybe this should derive from
- * <code>FilterOutputStream</code> and be tied to a single
- * <code>OutputStream</code> for its lifetime.
  */
 public class FramingOutputStream extends OutputStream
 {
     public FramingOutputStream ()
     {
-        _buffer = new byte[INITIAL_BUFFER_SIZE];
+        _buffer = ByteBuffer.allocate(INITIAL_BUFFER_SIZE);
-        _count = 4; // leave room for the frame size at the beginning
+        _buffer.put(HEADER_PAD);
     }
 
     /**
@@ -39,18 +39,12 @@ public class FramingOutputStream extends OutputStream
      */
     public void write (int b)
     {
-        // expand our buffer if necessary
+        try {
-	int newcount = _count + 1;
+            _buffer.put((byte)b);
-	if (newcount > _buffer.length) {
+        } catch (BufferOverflowException boe) {
-            // increase the buffer size in large increments
+            expand(1);
-	    byte[] newbuf = new byte[Math.max(_buffer.length << 1, newcount)];
+            _buffer.put((byte)b);
-	    System.arraycopy(_buffer, 0, newbuf, 0, _count);
+        }
-	    _buffer = newbuf;
-	}
-
-        // copy and advance
-	_buffer[_count] = (byte)b;
-	_count = newcount;
     }
 
     /**
@@ -71,59 +65,68 @@ public class FramingOutputStream extends OutputStream
 	    return;
 	}
 
-        // expand the buffer if necessary
+        try {
-        int newcount = _count + len;
+            _buffer.put(b, off, len);
-        if (newcount > _buffer.length) {
+        } catch (BufferOverflowException boe) {
-            // increase the buffer size in large increments
+            expand(len);
-            byte[] newbuf = new byte[Math.max(_buffer.length << 1, newcount)];
+            _buffer.put(b, off, len);
-            System.arraycopy(_buffer, 0, newbuf, 0, _count);
-            _buffer = newbuf;
         }
-
-        // copy and advance
-        System.arraycopy(b, off, _buffer, _count, len);
-        _count = newcount;
     }
 
     /**
-     * Writes the contents of this framing output stream to the target
+     * Expands our buffer to accomodate the specified capacity.
-     * output stream, prefixed by an integer with value equal to the
-     * number of bytes written following that integer. It then resets the
-     * framing output stream to prepare for another framed message.
-     *
-     * @return the total number of bytes written.
      */
-    public int writeFrameAndReset (OutputStream target)
+    protected final void expand (int needed)
-        throws IOException
     {
-        // prefix the frame with the byte count in network byte order (the
+        int ocapacity = _buffer.capacity();
-        // format used by DataOutputStream)
+        int ncapacity = _buffer.position() + needed;
-        int count = _count - 4;
+	if (ncapacity > ocapacity) {
-        _buffer[0] = (byte)((count >>> 24) & 0xFF);
+            // increase the buffer size in large increments
-	_buffer[1] = (byte)((count >>> 16) & 0xFF);
+            ncapacity = Math.max(ocapacity << 1, ncapacity);
-	_buffer[2] = (byte)((count >>>  8) & 0xFF);
+            ByteBuffer newbuf = ByteBuffer.allocate(ncapacity);
-	_buffer[3] = (byte)((count >>>  0) & 0xFF);
+            newbuf.put((ByteBuffer)_buffer.flip());
-
+	    _buffer = newbuf;
-        // write the data
+	}
-        target.write(_buffer, 0, _count);
-
-//         System.err.println("Wrote frame " + (_count-4));
-
-        // reset our internal buffer
-        reset();
-
-        return count + 4;
     }
 
-    public void reset ()
+    /**
+     * Writes the frame length to the beginning of our buffer and returns
+     * it for writing to the appropriate channel. This should be followed
+     * by a call to {@link #reset} when the frame has been written.
+     */
+    public ByteBuffer frameAndReturnBuffer ()
     {
-        // leave room for the frame size at the beginning
+        // flip the buffer which will limit it to it's current position
-        _count = 4;
+        _buffer.flip();
+
+        // then write the frame length and rewind back to the start of the
+        // buffer so that all the data is available
+        int count = _buffer.limit();
+        _buffer.put((byte)((count >>> 24) & 0xFF));
+	_buffer.put((byte)((count >>> 16) & 0xFF));
+	_buffer.put((byte)((count >>>  8) & 0xFF));
+	_buffer.put((byte)((count >>>  0) & 0xFF));
+        _buffer.rewind();
+
+        return _buffer;
     }
 
-    protected byte[] _buffer;
+    /**
-    protected int _count;
+     * Resets our internal buffer and prepares to write a new frame.
+     */
+    public void resetFrame ()
+    {
+        _buffer.clear();
+        _buffer.put(HEADER_PAD);
+    }
+
+    /** The buffer in which we store our frame data. */
+    protected ByteBuffer _buffer;
 
     /** The default initial size of the internal buffer. */
     protected static final int INITIAL_BUFFER_SIZE = 32;
+
+    /** We pad the beginning of our buffer so that we can write the frame
+     * length when the time comes. */
+    protected static final byte[] HEADER_PAD = new byte[4];
 }