Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

piped i/o streams from gnu classpath

  • Loading branch information...
commit 78b904f774f4b441a101b10c6614a8c29423524b 1 parent 35c8c1e
@maandree authored
Showing with 576 additions and 179 deletions.
  1. +399 −135 src/cnt/util/PipedInputStream.java
  2. +177 −44 src/cnt/util/PipedOutputStream.java
View
534 src/cnt/util/PipedInputStream.java
@@ -1,152 +1,416 @@
-package cnt.util;
+/* PipedInputStream.java -- Read portion of piped streams.
+ Copyright (C) 1998, 1999, 2000, 2001, 2003, 2005 Free Software Foundation, Inc.
+
+This file is part of GNU Classpath.
+
+GNU Classpath is free software; you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation; either version 2, or (at your option)
+any later version.
+
+GNU Classpath 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
+General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with GNU Classpath; see the file COPYING. If not, write to the
+Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+02110-1301 USA.
+
+Linking this library statically or dynamically with other modules is
+making a combined work based on this library. Thus, the terms and
+conditions of the GNU General Public License cover the whole
+combination.
-import java.io.IOException;
-import java.io.InputStream;
+As a special exception, the copyright holders of this library give you
+permission to link this library with independent modules to produce an
+executable, regardless of the license terms of these independent
+modules, and to copy and distribute the resulting executable under
+terms of your choice, provided that you also meet, for each linked
+independent module, the terms and conditions of the license of that
+module. An independent module is a module which is not derived from
+or based on this library. If you modify this library, you may extend
+this exception to your version of the library, but you are not
+obligated to do so. If you do not wish to do so, delete this
+exception statement from your version. */
+package cnt.util;
+import cnt.util.PipedOutputStream;
+import java.io.*;
+
+// NOTE: This implementation is very similar to that of PipedReader. If you
+// fix a bug in here, chances are you should make a similar change to the
+// PipedReader code.
+/**
+ * An input stream that reads its bytes from an output stream
+ * to which it is connected.
+ * <p>
+ * Data is read and written to an internal buffer. It is highly recommended
+ * that the <code>PipedInputStream</code> and connected
+ * <code>PipedOutputStream</code>
+ * be part of different threads. If they are not, the read and write
+ * operations could deadlock their thread.
+ *
+ * @specnote The JDK implementation appears to have some undocumented
+ * functionality where it keeps track of what thread is writing
+ * to pipe and throws an IOException if that thread susequently
+ * dies. This behaviour seems dubious and unreliable - we don't
+ * implement it.
+ *
+ * @author Aaron M. Renn (arenn@urbanophile.com)
+ */
public class PipedInputStream extends InputStream
{
- public PipedInputStream(final PipedOutputStream src) throws IOException
- {
- src.connect(this);
- }
-
- public PipedInputStream()
- {
- }
-
-
-
- protected byte buffer[] = new byte[1024];
+ /** PipedOutputStream to which this is connected. Null only if this
+ * InputStream hasn't been connected yet. */
+ PipedOutputStream source;
+
+ /** Set to true if close() has been called on this InputStream. */
+ boolean closed;
+
+
+ /**
+ * The size of the internal buffer used for input/output.
+ */
+ /* The "Constant Field Values" Javadoc of the Sun J2SE 1.4
+ * specifies 1024.
+ */
+ protected static final int PIPE_SIZE = 1024;
+
+
+ /**
+ * This is the internal circular buffer used for storing bytes written
+ * to the pipe and from which bytes are read by this stream
+ */
+ protected byte[] buffer = null;
+
+ /**
+ * The index into buffer where the next byte from the connected
+ * <code>PipedOutputStream</code> will be written. If this variable is
+ * equal to <code>out</code>, then the buffer is full. If set to < 0,
+ * the buffer is empty.
+ */
+ protected int in = -1;
+
+ /**
+ * This index into the buffer where bytes will be read from.
+ */
+ protected int out = 0;
+
+ /** Buffer used to implement single-argument read/receive */
+ private byte[] read_buf = new byte[1];
+
+ /**
+ * Creates a new <code>PipedInputStream</code> that is not connected to a
+ * <code>PipedOutputStream</code>. It must be connected before bytes can
+ * be read from this stream.
+ */
+ public PipedInputStream()
+ {
+ this(PIPE_SIZE);
+ }
+
+ /**
+ * Creates a new <code>PipedInputStream</code> of the given size that is not
+ * connected to a <code>PipedOutputStream</code>.
+ * It must be connected before bytes can be read from this stream.
+ *
+ * @since 1.6
+ * @since IllegalArgumentException If pipeSize <= 0.
+ */
+ public PipedInputStream(int pipeSize) throws IllegalArgumentException
+ {
+ if (pipeSize <= 0)
+ throw new IllegalArgumentException("pipeSize must be > 0");
- protected int in = -1;
- protected int out = 0;
+ this.buffer = new byte[pipeSize];
+ }
+
+ /**
+ * This constructor creates a new <code>PipedInputStream</code> and connects
+ * it to the passed in <code>PipedOutputStream</code>. The stream is then
+ * ready for reading.
+ *
+ * @param source The <code>PipedOutputStream</code> to connect this
+ * stream to
+ *
+ * @exception IOException If <code>source</code> is already connected.
+ */
+ public PipedInputStream(PipedOutputStream source) throws IOException
+ {
+ this();
+ connect(source);
+ }
+
+ /**
+ * This constructor creates a new <code>PipedInputStream</code> of the given
+ * size and connects it to the passed in <code>PipedOutputStream</code>.
+ * The stream is then ready for reading.
+ *
+ * @param source The <code>PipedOutputStream</code> to connect this
+ * stream to
+ *
+ * @since 1.6
+ * @exception IOException If <code>source</code> is already connected.
+ */
+ public PipedInputStream(PipedOutputStream source, int pipeSize)
+ throws IOException
+ {
+ this(pipeSize);
+ connect(source);
+ }
+
+ /**
+ * This method connects this stream to the passed in
+ * <code>PipedOutputStream</code>.
+ * This stream is then ready for reading. If this stream is already
+ * connected or has been previously closed, then an exception is thrown
+ *
+ * @param source The <code>PipedOutputStream</code> to connect this stream to
+ *
+ * @exception IOException If this PipedInputStream or <code>source</code>
+ * has been connected already.
+ */
+ public void connect(PipedOutputStream source) throws IOException
+ {
+ // The JDK (1.3) does not appear to check for a previously closed
+ // connection here.
+ if (this.source != null || source.sink != null)
+ throw new IOException ("Already connected");
+ source.sink = this;
+ this.source = source;
+ }
+
+ /**
+ * This method receives a byte of input from the source PipedOutputStream.
+ * If the internal circular buffer is full, this method blocks.
+ *
+ * @param val The byte to write to this stream
+ *
+ * @exception IOException if error occurs
+ * @specnote Weird. This method must be some sort of accident.
+ */
+ protected synchronized void receive(int val) throws IOException
+ {
+ read_buf[0] = (byte) (val & 0xff);
+ receive (read_buf, 0, 1);
+ }
+
+ /**
+ * This method is used by the connected <code>PipedOutputStream</code> to
+ * write bytes into the buffer.
+ *
+ * @param buf The array containing bytes to write to this stream
+ * @param offset The offset into the array to start writing from
+ * @param len The number of bytes to write.
+ *
+ * @exception IOException If an error occurs
+ * @specnote This code should be in PipedOutputStream.write, but we
+ * put it here in order to support that bizarre recieve(int)
+ * method.
+ */
+ synchronized void receive(byte[] buf, int offset, int len)
+ throws IOException
+ {
+ if (closed)
+ throw new IOException ("Pipe closed");
+
+ int bufpos = offset;
+ int copylen;
- protected synchronized void receive(final int b) throws IOException
- {
- if (this.in == this.out)
- awaitSpace();
-
- if (this.in < 0)
- this.in = this.out = 0;
-
- this.buffer[this.in++] = (byte)(b & 0xFF);
+ while (len > 0)
+ {
+ try
+ {
+ while (in == out)
+ {
+ // The pipe is full. Wake up any readers and wait for them.
+ notifyAll();
+ wait();
+ // The pipe could have been closed while we were waiting.
+ if (closed)
+ throw new IOException ("Pipe closed");
+ }
+ }
+ catch (InterruptedException ix)
+ {
+ throw new InterruptedIOException ();
+ }
+
+ if (in < 0) // The pipe is empty.
+ in = 0;
- if (this.in >= this.buffer.length)
- this.in = 0;
- }
-
- synchronized void receive(final byte b[], final int off, final int len) throws IOException
- {
- int coff = off;
- int bytesToTransfer = len;
- while (bytesToTransfer > 0)
- {
- if (this.in == this.out)
- awaitSpace();
- int nextTransferAmount = 0;
- if (this.out < this.in)
- nextTransferAmount = this.buffer.length - this.in;
- else if (this.in < this.out)
- if (this.in == -1)
- {
- this.in = this.out = 0;
- nextTransferAmount = this.buffer.length - this.in;
- }
- else
- nextTransferAmount = this.out - this.in;
- if (nextTransferAmount > bytesToTransfer)
- nextTransferAmount = bytesToTransfer;
-
- System.arraycopy(b, coff, this.buffer, this.in, nextTransferAmount);
-
- bytesToTransfer -= nextTransferAmount;
- coff += nextTransferAmount;
- this.in += nextTransferAmount;
-
- if (this.in >= this.buffer.length)
- this.in = 0;
- }
- }
-
-
- private void awaitSpace() throws IOException
- {
- while (this.in == this.out)
- { notifyAll();
- try
- { wait(1000);
- } catch (InterruptedException ex)
- { throw new java.io.InterruptedIOException();
- } }
- }
+ // Figure out how many bytes from buf can be copied without
+ // overrunning out or going past the length of the buffer.
+ if (in < out)
+ copylen = Math.min (len, out - in);
+ else
+ copylen = Math.min (len, buffer.length - in);
+
+ // Copy bytes until the pipe is filled, wrapping if necessary.
+ System.arraycopy(buf, bufpos, buffer, in, copylen);
+ len -= copylen;
+ bufpos += copylen;
+ in += copylen;
+ if (in == buffer.length)
+ in = 0;
+ }
+ // Notify readers that new data is in the pipe.
+ notifyAll();
+ }
+
+ /**
+ * This method reads one byte from the stream.
+ * -1 is returned to indicated that no bytes can be read
+ * because the end of the stream was reached. If the stream is already
+ * closed, a -1 will again be returned to indicate the end of the stream.
+ *
+ * <p>This method will block if no byte is available to be read.</p>
+ *
+ * @return the value of the read byte value, or -1 of the end of the stream
+ * was reached
+ *
+ * @throws IOException if an error occured
+ */
+ public int read() throws IOException
+ {
+ // Method operates by calling the multibyte overloaded read method
+ // Note that read_buf is an internal instance variable. I allocate it
+ // there to avoid constant reallocation overhead for applications that
+ // call this method in a loop at the cost of some unneeded overhead
+ // if this method is never called.
+
+ int r = read(read_buf, 0, 1);
+ return r != -1 ? (read_buf[0] & 0xff) : -1;
+ }
+
+ /**
+ * This method reads bytes from the stream into a caller supplied buffer.
+ * It starts storing bytes at position <code>offset</code> into the
+ * buffer and
+ * reads a maximum of <code>len</code> bytes. Note that this method
+ * can actually
+ * read fewer than <code>len</code> bytes. The actual number of bytes
+ * read is
+ * returned. A -1 is returned to indicated that no bytes can be read
+ * because the end of the stream was reached - ie close() was called on the
+ * connected PipedOutputStream.
+ * <p>
+ * This method will block if no bytes are available to be read.
+ *
+ * @param buf The buffer into which bytes will be stored
+ * @param offset The index into the buffer at which to start writing.
+ * @param len The maximum number of bytes to read.
+ *
+ * @exception IOException If <code>close()</code> was called on this Piped
+ * InputStream.
+ */
+ public synchronized int read(byte[] buf, int offset, int len)
+ throws IOException
+ {
+ if (source == null)
+ throw new IOException ("Not connected");
+ if (closed)
+ throw new IOException ("Pipe closed");
+
+ // Don't block if nothing was requested.
+ if (len == 0)
+ return 0;
+
+ // If the buffer is empty, wait until there is something in the pipe
+ // to read.
+ try
+ {
+ while (in < 0)
+ {
+ if (source.closed)
+ return -1;
+ wait();
+ }
+ }
+ catch (InterruptedException ix)
+ {
+ throw new InterruptedIOException();
+ }
+ int total = 0;
+ int copylen;
- public synchronized int read() throws IOException
- {
- while (this.in < 0)
- { notifyAll();
- try
- { wait(1000);
- }
- catch (InterruptedException ex)
- { throw new java.io.InterruptedIOException();
- } }
-
- int ret = this.buffer[this.out++] & 0xFF;
-
- if (this.out >= this.buffer.length)
- this.out = 0;
- if (this.in == this.out)
- this.in = -1;
+ while (true)
+ {
+ // Figure out how many bytes from the pipe can be copied without
+ // overrunning in or going past the length of buf.
+ if (out < in)
+ copylen = Math.min (len, in - out);
+ else
+ copylen = Math.min (len, buffer.length - out);
- return ret;
- }
-
- public synchronized int read(final byte b[], final int off, final int len) throws IOException
- {
- if (len == 0)
- return 0;
+ System.arraycopy (buffer, out, buf, offset, copylen);
+ offset += copylen;
+ len -= copylen;
+ out += copylen;
+ total += copylen;
- int c = read();
- if (c < 0)
- return -1;
+ if (out == buffer.length)
+ out = 0;
- b[off] = (byte)c;
- int rlen = 1;
- int clen = len;
- while ((this.in >= 0) && (clen > 1))
- {
- int available;
-
- if (this.in > this.out) available = Math.min(this.buffer.length - this.out, this.in - this.out);
- else available = this.buffer.length - this.out;
-
- if (available > clen - 1)
- available = clen - 1;
-
- System.arraycopy(this.buffer, this.out, b, off + rlen, available);
- this.out += available;
- rlen += available;
- clen -= available;
-
- if (this.out >= this.buffer.length)
- this.out = 0;
-
- if (this.in == this.out)
- this.in = -1;
- }
- return rlen;
- }
+ if (out == in)
+ {
+ // Pipe is now empty.
+ in = -1;
+ out = 0;
+ }
+
+ // If output buffer is filled or the pipe is empty, we're done.
+ if (len == 0 || in == -1)
+ {
+ // Notify any waiting outputstream that there is now space
+ // to write.
+ notifyAll();
+ return total;
+ }
+ }
+ }
+
+ /**
+ * This method returns the number of bytes that can be read from this stream
+ * before blocking could occur. This is the number of bytes that are
+ * currently unread in the internal circular buffer. Note that once this
+ * many additional bytes are read, the stream may block on a subsequent
+ * read, but it not guaranteed to block.
+ *
+ * @return The number of bytes that can be read before blocking might occur
+ *
+ * @exception IOException If an error occurs
+ */
+ public synchronized int available() throws IOException
+ {
+ // The JDK 1.3 implementation does not appear to check for the closed or
+ // unconnected stream conditions here.
- public synchronized int available() throws IOException
- {
- if (this.in < 0) return 0;
- if (this.in == this.out) return this.buffer.length;
- if (this.in > this.out) return this.in - this.out;
-
- return this.in + this.buffer.length - this.out;
- }
+ if (in < 0)
+ return 0;
+ else if (out < in)
+ return in - out;
+ else
+ return (buffer.length - out) + in;
+ }
+
+ /**
+ * This methods closes the stream so that no more data can be read
+ * from it.
+ *
+ * @exception IOException If an error occurs
+ */
+ public synchronized void close() throws IOException
+ {
+ closed = true;
+ // Wake any thread which may be in receive() waiting to write data.
+ notifyAll();
+ }
}
+
View
221 src/cnt/util/PipedOutputStream.java
@@ -1,50 +1,183 @@
-package cnt.util;
+/* PipedOutputStream.java -- Write portion of piped streams.
+ Copyright (C) 1998, 2000, 2001, 2003 Free Software Foundation, Inc.
+
+This file is part of GNU Classpath.
+
+GNU Classpath is free software; you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation; either version 2, or (at your option)
+any later version.
+
+GNU Classpath 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
+General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with GNU Classpath; see the file COPYING. If not, write to the
+Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+02110-1301 USA.
+
+Linking this library statically or dynamically with other modules is
+making a combined work based on this library. Thus, the terms and
+conditions of the GNU General Public License cover the whole
+combination.
+
+As a special exception, the copyright holders of this library give you
+permission to link this library with independent modules to produce an
+executable, regardless of the license terms of these independent
+modules, and to copy and distribute the resulting executable under
+terms of your choice, provided that you also meet, for each linked
+independent module, the terms and conditions of the license of that
+module. An independent module is a module which is not derived from
+or based on this library. If you modify this library, you may extend
+this exception to your version of the library, but you are not
+obligated to do so. If you do not wish to do so, delete this
+exception statement from your version. */
-import java.io.IOException;
-import java.io.OutputStream;
+package cnt.util;
+import cnt.util.PipedInputStream;
+import java.io.*;
+
+// NOTE: This implementation is very similar to that of PipedWriter. If you
+// fix a bug in here, chances are you should make a similar change to the
+// PipedWriter code.
+/**
+ * This class writes its bytes to a <code>PipedInputStream</code> to
+ * which it is connected.
+ * <p>
+ * It is highly recommended that a <code>PipedOutputStream</code> and its
+ * connected <code>PipedInputStream</code> be in different threads. If
+ * they are in the same thread, read and write operations could deadlock
+ * the thread.
+ *
+ * @author Aaron M. Renn (arenn@urbanophile.com)
+ */
public class PipedOutputStream extends OutputStream
{
- public PipedOutputStream(final PipedInputStream sink) throws IOException
- {
- connect(sink);
- }
-
- public PipedOutputStream()
- {
- }
-
-
-
- private PipedInputStream sink;
-
-
-
- public void connect(final PipedInputStream sink)
- {
- this.sink = sink;
- sink.in = -1;
- sink.out = 0;
- }
-
- public void write(final int b) throws IOException
- {
- this.sink.receive(b);
- }
-
- public void write(final byte b[], final int off, final int len) throws IOException
- {
- this.sink.receive(b, off, len);
- }
-
- public synchronized void flush() throws IOException
- {
- if (this.sink != null)
- synchronized (this.sink)
- {
- this.sink.notifyAll();
- }
- }
-
+ /** Target PipedInputStream to which this is connected. Null only if this
+ * OutputStream hasn't been connected yet. */
+ PipedInputStream sink;
+
+ /** Set to true if close() has been called on this OutputStream. */
+ boolean closed;
+
+ /**
+ * Create an unconnected PipedOutputStream. It must be connected
+ * to a <code>PipedInputStream</code> using the <code>connect</code>
+ * method prior to writing any data or an exception will be thrown.
+ */
+ public PipedOutputStream()
+ {
+ }
+
+ /**
+ * Create a new <code>PipedOutputStream</code> instance
+ * to write to the specified <code>PipedInputStream</code>. This stream
+ * is then ready for writing.
+ *
+ * @param sink The <code>PipedInputStream</code> to connect this stream to.
+ *
+ * @exception IOException If <code>sink</code> has already been connected
+ * to a different PipedOutputStream.
+ */
+ public PipedOutputStream(PipedInputStream sink) throws IOException
+ {
+ sink.connect(this);
+ }
+
+ /**
+ * Connects this object to the specified <code>PipedInputStream</code>
+ * object. This stream will then be ready for writing.
+ *
+ * @param sink The <code>PipedInputStream</code> to connect this stream to
+ *
+ * @exception IOException If the stream has not been connected or has
+ * been closed.
+ */
+ public void connect(PipedInputStream sink) throws IOException
+ {
+ if (this.sink != null || sink.source != null)
+ throw new IOException ("Already connected");
+ sink.connect(this);
+ }
+
+ /**
+ * Write a single byte of date to the stream. Note that this method will
+ * block if the <code>PipedInputStream</code> to which this object is
+ * connected has a full buffer.
+ *
+ * @param b The byte of data to be written, passed as an <code>int</code>.
+ *
+ * @exception IOException If the stream has not been connected or has
+ * been closed.
+ */
+ public void write(int b) throws IOException
+ {
+ if (sink == null)
+ throw new IOException ("Not connected");
+ if (closed)
+ throw new IOException ("Pipe closed");
+
+ sink.receive (b);
+ }
+
+ /**
+ * This method writes <code>len</code> bytes of data from the byte array
+ * <code>buf</code> starting at index <code>offset</code> in the array
+ * to the stream. Note that this method will block if the
+ * <code>PipedInputStream</code> to which this object is connected has
+ * a buffer that cannot hold all of the bytes to be written.
+ *
+ * @param buffer The array containing bytes to write to the stream.
+ * @param offset The index into the array to start writing bytes from.
+ * @param len The number of bytes to write.
+ *
+ * @exception IOException If the stream has not been connected or has
+ * been closed.
+ */
+ public void write(byte[] buffer, int offset, int len) throws IOException
+ {
+ if (sink == null)
+ throw new IOException ("Not connected");
+ if (closed)
+ throw new IOException ("Pipe closed");
+
+ sink.receive(buffer, offset, len);
+ }
+
+ /**
+ * This method does nothing.
+ *
+ * @exception IOException If the stream is closed.
+ * @specnote You'd think that this method would block until the sink
+ * had read all available data. Thats not the case - this method
+ * appears to be a no-op?
+ */
+ public void flush() throws IOException
+ {
+ }
+
+ /**
+ * This method closes this stream so that no more data can be written
+ * to it. Any further attempts to write to this stream may throw an
+ * exception
+ *
+ * @exception IOException If an error occurs
+ */
+ public void close() throws IOException
+ {
+ // A close call on an unconnected PipedOutputStream has no effect.
+ if (sink != null)
+ {
+ closed = true;
+ // Notify any waiting readers that the stream is now closed.
+ synchronized (sink)
+ {
+ sink.notifyAll();
+ }
+ }
+ }
}

0 comments on commit 78b904f

Please sign in to comment.
Something went wrong with that request. Please try again.