* Data is read and written to an internal buffer. It is highly recommended
* that the PipedInputStream and connected PipedOutputStream
* 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
{
/** 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.
*/
protected static final int PIPE_SIZE = 2048;
/**
* 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 = new byte[PIPE_SIZE];
/**
* The index into buffer where the next byte from the connected
* PipedOutputStream will be written. If this variable is
* equal to out, 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 PipedInputStream that is not connected to a
* PipedOutputStream. It must be connected before bytes can
* be read from this stream.
*/
public PipedInputStream()
{
}
/**
* This constructor creates a new PipedInputStream and connects
* it to the passed in PipedOutputStream. The stream is then
* ready for reading.
*
* @param source The PipedOutputStream to connect this stream to
*
* @exception IOException If source is already connected.
*/
public PipedInputStream(PipedOutputStream source) throws IOException
{
connect(source);
}
/**
* This method connects this stream to the passed in PipedOutputStream.
* This stream is then ready for reading. If this stream is already
* connected or has been previously closed, then an exception is thrown
*
* @param src The PipedOutputStream to connect this stream to
*
* @exception IOException If this PipedInputStream or source
* 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 byte_received 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 b) throws IOException
{
read_buf[0] = (byte) (b & 0xff);
receive (read_buf, 0, 1);
}
/**
* This method is used by the connected PipedOutputStream 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;
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;
// 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 bytes from the stream into a caller supplied buffer.
* It starts storing bytes at position offset into the buffer and
* reads a maximum of len bytes. Note that this method can actually
* read fewer than len 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. If the stream is already
* closed, a -1 will again be returned to indicate the end of the stream.
*
* 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.
*/
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);
if (r == -1)
return -1;
else
return read_buf[0];
}
/**
* This method reads bytes from the stream into a caller supplied buffer.
* It starts storing bytes at position offset into the buffer and
* reads a maximum of len bytes. Note that this method can actually
* read fewer than len 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.
*
* 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 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");
// 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;
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);
System.arraycopy (buffer, out, buf, offset, copylen);
offset += copylen;
len -= copylen;
out += copylen;
total += copylen;
if (out == buffer.length)
out = 0;
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.
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();
}
}