libjava/java/io/PipedInputStream.java

   1 /* PipedInputStream.java -- Read portion of piped streams.
   2    Copyright (C) 1998, 1999, 2000, 2001 Free Software Foundation, Inc.
   3
   4 This file is part of GNU Classpath.
   5
   6 GNU Classpath is free software; you can redistribute it and/or modify
   7 it under the terms of the GNU General Public License as published by
   8 the Free Software Foundation; either version 2, or (at your option)
   9 any later version.
  10
  11 GNU Classpath is distributed in the hope that it will be useful, but
  12 WITHOUT ANY WARRANTY; without even the implied warranty of
  13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 General Public License for more details.
  15
  16 You should have received a copy of the GNU General Public License
  17 along with GNU Classpath; see the file COPYING.  If not, write to the
  18 Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
  19 02111-1307 USA.
  20
  21 As a special exception, if you link this library with other files to
  22 produce an executable, this library does not by itself cause the
  23 resulting executable to be covered by the GNU General Public License.
  24 This exception does not however invalidate any other reasons why the
  25 executable file might be covered by the GNU General Public License. */
  26
  27 // NOTE: This implementation is very similar to that of PipedReader.  If you
  28 // fix a bug in here, chances are you should make a similar change to the
  29 // PipedReader code.
  30
  31 package java.io;
  32
  33 /**
  34   * An input stream that reads its bytes from an output stream
  35   * to which it is connected.
  36   * <p>
  37   * Data is read and written to an internal buffer.  It is highly recommended
  38   * that the <code>PipedInputStream</code> and connected <code>PipedOutputStream</code>
  39   * be part of different threads.  If they are not, the read and write
  40   * operations could deadlock their thread.
  41   *
  42   * @specnote The JDK implementation appears to have some undocumented
  43   *           functionality where it keeps track of what thread is writing
  44   *           to pipe and throws an IOException if that thread susequently
  45   *           dies. This behaviour seems dubious and unreliable - we don't
  46   *           implement it.
  47   *
  48   * @author Aaron M. Renn (arenn@urbanophile.com)
  49   */
  50 public class PipedInputStream extends InputStream
  51 {
  52   /** PipedOutputStream to which this is connected. Null only if this
  53     * InputStream hasn't been connected yet. */
  54   PipedOutputStream source;
  55
  56   /** Set to true if close() has been called on this InputStream. */
  57   boolean closed;
  58
  59   /**
  60     * The size of the internal buffer used for input/output.
  61     */
  62   protected static final int PIPE_SIZE = 2048;
  63
  64   /**
  65     * This is the internal circular buffer used for storing bytes written
  66     * to the pipe and from which bytes are read by this stream
  67     */
  68   protected byte[] buffer = new byte[PIPE_SIZE];
  69
  70   /**
  71     * The index into buffer where the next byte from the connected
  72     * <code>PipedOutputStream</code> will be written. If this variable is
  73     * equal to <code>out</code>, then the buffer is full. If set to < 0,
  74     * the buffer is empty.
  75     */
  76   protected int in = -1;
  77
  78   /**
  79     * This index into the buffer where bytes will be read from.
  80     */
  81   protected int out = 0;
  82
  83   /** Buffer used to implement single-argument read/receive */
  84   private byte[] read_buf = new byte[1];
  85
  86   /**
  87     * Creates a new <code>PipedInputStream</code> that is not connected to a
  88     * <code>PipedOutputStream</code>.  It must be connected before bytes can
  89     * be read from this stream.
  90     */
  91   public PipedInputStream()
  92   {
  93   }
  94
  95   /**
  96     * This constructor creates a new <code>PipedInputStream</code> and connects
  97     * it to the passed in <code>PipedOutputStream</code>. The stream is then
  98     * ready for reading.
  99     *
 100     * @param source The <code>PipedOutputStream</code> to connect this stream to
 101     *
 102     * @exception IOException If <code>source</code> is already connected.
 103     */
 104   public PipedInputStream(PipedOutputStream source) throws IOException
 105   {
 106     connect(source);
 107   }
 108
 109   /**
 110     * This method connects this stream to the passed in <code>PipedOutputStream</code>.
 111     * This stream is then ready for reading.  If this stream is already
 112     * connected or has been previously closed, then an exception is thrown
 113     *
 114     * @param src The <code>PipedOutputStream</code> to connect this stream to
 115     *
 116     * @exception IOException If this PipedInputStream or <code>source</code>
 117     *                        has been connected already.
 118     */
 119   public void connect(PipedOutputStream source) throws IOException
 120   {
 121     // The JDK (1.3) does not appear to check for a previously closed
 122     // connection here.
 123
 124     if (this.source != null || source.sink != null)
 125       throw new IOException ("Already connected");
 126
 127     source.sink = this;
 128     this.source = source;
 129   }
 130
 131   /**
 132   * This method receives a byte of input from the source PipedOutputStream.
 133   * If the internal circular buffer is full, this method blocks.
 134   *
 135   * @param byte_received The byte to write to this stream
 136   *
 137   * @exception IOException if error occurs
 138   * @specnote Weird. This method must be some sort of accident.
 139   */
 140   protected synchronized void receive(int b) throws IOException
 141   {
 142     read_buf[0] = (byte) (b & 0xff);
 143     receive (read_buf, 0, 1);
 144   }
 145
 146   /**
 147     * This method is used by the connected <code>PipedOutputStream</code> to
 148     * write bytes into the buffer.
 149     *
 150     * @param buf The array containing bytes to write to this stream
 151     * @param offset The offset into the array to start writing from
 152     * @param len The number of bytes to write.
 153     *
 154     * @exception IOException If an error occurs
 155     * @specnote This code should be in PipedOutputStream.write, but we
 156     *           put it here in order to support that bizarre recieve(int)
 157     *           method.
 158     */
 159   synchronized void receive(byte[] buf, int offset, int len)
 160     throws IOException
 161   {
 162     if (closed)
 163       throw new IOException ("Pipe closed");
 164
 165     int bufpos = offset;
 166     int copylen;
 167
 168     while (len > 0)
 169       {
 170         try
 171           {
 172             while (in == out)
 173               {
 174                 // The pipe is full. Wake up any readers and wait for them.
 175                 notifyAll();
 176                 wait();
 177                 // The pipe could have been closed while we were waiting.
 178                 if (closed)
 179                   throw new IOException ("Pipe closed");
 180               }
 181           }
 182         catch (InterruptedException ix)
 183           {
 184             throw new InterruptedIOException ();
 185           }
 186
 187         if (in < 0) // The pipe is empty.
 188           in = 0;
 189
 190         // Figure out how many bytes from buf can be copied without
 191         // overrunning out or going past the length of the buffer.
 192         if (in < out)
 193           copylen = Math.min (len, out - in);
 194         else
 195           copylen = Math.min (len, buffer.length - in);
 196
 197         // Copy bytes until the pipe is filled, wrapping if neccessary.
 198         System.arraycopy(buf, bufpos, buffer, in, copylen);
 199         len -= copylen;
 200         bufpos += copylen;
 201         in += copylen;
 202         if (in == buffer.length)
 203           in = 0;
 204       }
 205     // Notify readers that new data is in the pipe.
 206     notifyAll();
 207   }
 208
 209   /**
 210     * This method reads bytes from the stream into a caller supplied buffer.
 211     * It starts storing bytes at position <code>offset</code> into the buffer and
 212     * reads a maximum of <cod>>len</code> bytes.  Note that this method can actually
 213     * read fewer than <code>len</code> bytes.  The actual number of bytes read is
 214     * returned.  A -1 is returned to indicated that no bytes can be read
 215     * because the end of the stream was reached.  If the stream is already
 216     * closed, a -1 will again be returned to indicate the end of the stream.
 217     * <p>
 218     * This method will block if no bytes are available to be read.
 219     *
 220     * @param buf The buffer into which bytes will be stored
 221     * @param offset The index into the buffer at which to start writing.
 222     * @param len The maximum number of bytes to read.
 223     */
 224   public int read() throws IOException
 225   {
 226     // Method operates by calling the multibyte overloaded read method
 227     // Note that read_buf is an internal instance variable.  I allocate it
 228     // there to avoid constant reallocation overhead for applications that
 229     // call this method in a loop at the cost of some unneeded overhead
 230     // if this method is never called.
 231
 232     int r = read(read_buf, 0, 1);
 233
 234     if (r == -1)
 235       return -1;
 236     else
 237       return read_buf[0];
 238   }
 239
 240   /**
 241     * This method reads bytes from the stream into a caller supplied buffer.
 242     * It starts storing bytes at position <code>offset</code> into the buffer and
 243     * reads a maximum of <cod>>len</code> bytes.  Note that this method can actually
 244     * read fewer than <code>len</code> bytes.  The actual number of bytes read is
 245     * returned.  A -1 is returned to indicated that no bytes can be read
 246     * because the end of the stream was reached - ie close() was called on the
 247     * connected PipedOutputStream.
 248     * <p>
 249     * This method will block if no bytes are available to be read.
 250     *
 251     * @param buf The buffer into which bytes will be stored
 252     * @param offset The index into the buffer at which to start writing.
 253     * @param len The maximum number of bytes to read.
 254     *
 255     * @exception IOException If <code>close()/code> was called on this Piped
 256     *                        InputStream.
 257     */
 258   public synchronized int read(byte[] buf, int offset, int len)
 259     throws IOException
 260   {
 261     if (source == null)
 262       throw new IOException ("Not connected");
 263     if (closed)
 264       throw new IOException ("Pipe closed");
 265
 266     // If the buffer is empty, wait until there is something in the pipe
 267     // to read.
 268     try
 269       {
 270         while (in < 0)
 271           {
 272             if (source.closed)
 273               return -1;
 274             wait();
 275           }
 276       }
 277     catch (InterruptedException ix)
 278       {
 279         throw new InterruptedIOException();
 280       }
 281
 282     int total = 0;
 283     int copylen;
 284
 285     while (true)
 286       {
 287         // Figure out how many bytes from the pipe can be copied without
 288         // overrunning in or going past the length of buf.
 289         if (out < in)
 290           copylen = Math.min (len, in - out);
 291         else
 292           copylen = Math.min (len, buffer.length - out);
 293
 294         System.arraycopy (buffer, out, buf, offset, copylen);
 295         offset += copylen;
 296         len -= copylen;
 297         out += copylen;
 298         total += copylen;
 299
 300         if (out == buffer.length)
 301           out = 0;
 302
 303         if (out == in)
 304           {
 305             // Pipe is now empty.
 306             in = -1;
 307             out = 0;
 308           }
 309
 310         // If output buffer is filled or the pipe is empty, we're done.
 311         if (len == 0 || in == -1)
 312           {
 313             // Notify any waiting outputstream that there is now space
 314             // to write.
 315             notifyAll();
 316             return total;
 317           }
 318       }
 319   }
 320
 321   /**
 322     * This method returns the number of bytes that can be read from this stream
 323     * before blocking could occur.  This is the number of bytes that are
 324     * currently unread in the internal circular buffer.  Note that once this
 325     * many additional bytes are read, the stream may block on a subsequent
 326     * read, but it not guaranteed to block.
 327     *
 328     * @return The number of bytes that can be read before blocking might occur
 329     *
 330     * @exception IOException If an error occurs
 331     */
 332   public synchronized int available() throws IOException
 333   {
 334     // The JDK 1.3 implementation does not appear to check for the closed or
 335     // unconnected stream conditions here.
 336
 337     if (in < 0)
 338       return 0;
 339     else if (out < in)
 340       return in - out;
 341     else
 342       return (buffer.length - out) + in;
 343   }
 344
 345   /**
 346   * This methods closes the stream so that no more data can be read
 347   * from it.
 348   *
 349   * @exception IOException If an error occurs
 350   */
 351   public synchronized void close() throws IOException
 352   {
 353     closed = true;
 354     // Wake any thread which may be in receive() waiting to write data.
 355     notifyAll();
 356   }
 357 }