001/* PipedInputStream.java -- Read portion of piped streams.
002   Copyright (C) 1998, 1999, 2000, 2001, 2003, 2005  Free Software Foundation, Inc.
003
004This file is part of GNU Classpath.
005
006GNU Classpath is free software; you can redistribute it and/or modify
007it under the terms of the GNU General Public License as published by
008the Free Software Foundation; either version 2, or (at your option)
009any later version.
010
011GNU Classpath is distributed in the hope that it will be useful, but
012WITHOUT ANY WARRANTY; without even the implied warranty of
013MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
014General Public License for more details.
015
016You should have received a copy of the GNU General Public License
017along with GNU Classpath; see the file COPYING.  If not, write to the
018Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
01902110-1301 USA.
020
021Linking this library statically or dynamically with other modules is
022making a combined work based on this library.  Thus, the terms and
023conditions of the GNU General Public License cover the whole
024combination.
025
026As a special exception, the copyright holders of this library give you
027permission to link this library with independent modules to produce an
028executable, regardless of the license terms of these independent
029modules, and to copy and distribute the resulting executable under
030terms of your choice, provided that you also meet, for each linked
031independent module, the terms and conditions of the license of that
032module.  An independent module is a module which is not derived from
033or based on this library.  If you modify this library, you may extend
034this exception to your version of the library, but you are not
035obligated to do so.  If you do not wish to do so, delete this
036exception statement from your version. */
037
038package java.io;
039
040// NOTE: This implementation is very similar to that of PipedReader.  If you
041// fix a bug in here, chances are you should make a similar change to the
042// PipedReader code.
043
044/**
045  * An input stream that reads its bytes from an output stream
046  * to which it is connected.
047  * <p>
048  * Data is read and written to an internal buffer.  It is highly recommended
049  * that the <code>PipedInputStream</code> and connected
050  * <code>PipedOutputStream</code>
051  * be part of different threads.  If they are not, the read and write
052  * operations could deadlock their thread.
053  *
054  * @specnote The JDK implementation appears to have some undocumented
055  *           functionality where it keeps track of what thread is writing
056  *           to pipe and throws an IOException if that thread susequently
057  *           dies. This behaviour seems dubious and unreliable - we don't
058  *           implement it.
059  *
060  * @author Aaron M. Renn (arenn@urbanophile.com)
061  */
062public class PipedInputStream extends InputStream
063{
064  /** PipedOutputStream to which this is connected. Null only if this
065    * InputStream hasn't been connected yet. */
066  PipedOutputStream source;
067
068  /** Set to true if close() has been called on this InputStream. */
069  boolean closed;
070
071
072  /**
073   * The size of the internal buffer used for input/output.
074   */
075  /* The "Constant Field Values" Javadoc of the Sun J2SE 1.4
076   * specifies 1024.
077   */
078  protected static final int PIPE_SIZE = 1024;
079
080
081  /**
082    * This is the internal circular buffer used for storing bytes written
083    * to the pipe and from which bytes are read by this stream
084    */
085  protected byte[] buffer = null;
086
087  /**
088    * The index into buffer where the next byte from the connected
089    * <code>PipedOutputStream</code> will be written. If this variable is
090    * equal to <code>out</code>, then the buffer is full. If set to < 0,
091    * the buffer is empty.
092    */
093  protected int in = -1;
094
095  /**
096    * This index into the buffer where bytes will be read from.
097    */
098  protected int out = 0;
099
100  /** Buffer used to implement single-argument read/receive */
101  private byte[] read_buf = new byte[1];
102
103  /**
104    * Creates a new <code>PipedInputStream</code> that is not connected to a
105    * <code>PipedOutputStream</code>.  It must be connected before bytes can
106    * be read from this stream.
107    */
108  public PipedInputStream()
109  {
110    this(PIPE_SIZE);
111  }
112
113  /**
114   * Creates a new <code>PipedInputStream</code> of the given size that is not
115   * connected to a <code>PipedOutputStream</code>.
116   * It must be connected before bytes can be read from this stream.
117   *
118   * @since 1.6
119   * @since IllegalArgumentException If pipeSize <= 0.
120   */
121  public PipedInputStream(int pipeSize) throws IllegalArgumentException
122  {
123    if (pipeSize <= 0)
124      throw new IllegalArgumentException("pipeSize must be > 0");
125
126    this.buffer = new byte[pipeSize];
127  }
128
129  /**
130    * This constructor creates a new <code>PipedInputStream</code> and connects
131    * it to the passed in <code>PipedOutputStream</code>. The stream is then
132    * ready for reading.
133    *
134    * @param source The <code>PipedOutputStream</code> to connect this
135    * stream to
136    *
137    * @exception IOException If <code>source</code> is already connected.
138    */
139  public PipedInputStream(PipedOutputStream source) throws IOException
140  {
141    this();
142    connect(source);
143  }
144
145  /**
146   * This constructor creates a new <code>PipedInputStream</code> of the given
147   * size and connects it to the passed in <code>PipedOutputStream</code>.
148   * The stream is then ready for reading.
149   *
150   * @param source The <code>PipedOutputStream</code> to connect this
151   * stream to
152   *
153   * @since 1.6
154   * @exception IOException If <code>source</code> is already connected.
155   */
156 public PipedInputStream(PipedOutputStream source, int pipeSize)
157   throws IOException
158 {
159   this(pipeSize);
160   connect(source);
161 }
162
163  /**
164    * This method connects this stream to the passed in
165    * <code>PipedOutputStream</code>.
166    * This stream is then ready for reading.  If this stream is already
167    * connected or has been previously closed, then an exception is thrown
168    *
169    * @param source The <code>PipedOutputStream</code> to connect this stream to
170    *
171    * @exception IOException If this PipedInputStream or <code>source</code>
172    *                        has been connected already.
173    */
174  public void connect(PipedOutputStream source) throws IOException
175  {
176    // The JDK (1.3) does not appear to check for a previously closed
177    // connection here.
178
179    if (this.source != null || source.sink != null)
180      throw new IOException ("Already connected");
181
182    source.sink = this;
183    this.source = source;
184  }
185
186  /**
187  * This method receives a byte of input from the source PipedOutputStream.
188  * If the internal circular buffer is full, this method blocks.
189  *
190  * @param val The byte to write to this stream
191  *
192  * @exception IOException if error occurs
193  * @specnote Weird. This method must be some sort of accident.
194  */
195  protected synchronized void receive(int val) throws IOException
196  {
197    read_buf[0] = (byte) (val & 0xff);
198    receive (read_buf, 0, 1);
199  }
200
201  /**
202    * This method is used by the connected <code>PipedOutputStream</code> to
203    * write bytes into the buffer.
204    *
205    * @param buf The array containing bytes to write to this stream
206    * @param offset The offset into the array to start writing from
207    * @param len The number of bytes to write.
208    *
209    * @exception IOException If an error occurs
210    * @specnote This code should be in PipedOutputStream.write, but we
211    *           put it here in order to support that bizarre recieve(int)
212    *           method.
213    */
214  synchronized void receive(byte[] buf, int offset, int len)
215    throws IOException
216  {
217    if (closed)
218      throw new IOException ("Pipe closed");
219
220    int bufpos = offset;
221    int copylen;
222
223    while (len > 0)
224      {
225        try
226          {
227            while (in == out)
228              {
229                // The pipe is full. Wake up any readers and wait for them.
230                notifyAll();
231                wait();
232                // The pipe could have been closed while we were waiting.
233                if (closed)
234                  throw new IOException ("Pipe closed");
235              }
236          }
237        catch (InterruptedException ix)
238          {
239            throw new InterruptedIOException ();
240          }
241
242        if (in < 0) // The pipe is empty.
243          in = 0;
244
245        // Figure out how many bytes from buf can be copied without
246        // overrunning out or going past the length of the buffer.
247        if (in < out)
248          copylen = Math.min (len, out - in);
249        else
250          copylen = Math.min (len, buffer.length - in);
251
252        // Copy bytes until the pipe is filled, wrapping if necessary.
253        System.arraycopy(buf, bufpos, buffer, in, copylen);
254        len -= copylen;
255        bufpos += copylen;
256        in += copylen;
257        if (in == buffer.length)
258          in = 0;
259      }
260    // Notify readers that new data is in the pipe.
261    notifyAll();
262  }
263
264  /**
265    * This method reads one byte from the stream.
266    * -1 is returned to indicated that no bytes can be read
267    * because the end of the stream was reached.  If the stream is already
268    * closed, a -1 will again be returned to indicate the end of the stream.
269    *
270    * <p>This method will block if no byte is available to be read.</p>
271    *
272    * @return the value of the read byte value, or -1 of the end of the stream
273    * was reached
274    *
275    * @throws IOException if an error occured
276    */
277  public int read() throws IOException
278  {
279    // Method operates by calling the multibyte overloaded read method
280    // Note that read_buf is an internal instance variable.  I allocate it
281    // there to avoid constant reallocation overhead for applications that
282    // call this method in a loop at the cost of some unneeded overhead
283    // if this method is never called.
284
285    int r = read(read_buf, 0, 1);
286    return r != -1 ? (read_buf[0] & 0xff) : -1;
287  }
288
289  /**
290    * This method reads bytes from the stream into a caller supplied buffer.
291    * It starts storing bytes at position <code>offset</code> into the
292    * buffer and
293    * reads a maximum of <code>len</code> bytes.  Note that this method
294    * can actually
295    * read fewer than <code>len</code> bytes.  The actual number of bytes
296    * read is
297    * returned.  A -1 is returned to indicated that no bytes can be read
298    * because the end of the stream was reached - ie close() was called on the
299    * connected PipedOutputStream.
300    * <p>
301    * This method will block if no bytes are available to be read.
302    *
303    * @param buf The buffer into which bytes will be stored
304    * @param offset The index into the buffer at which to start writing.
305    * @param len The maximum number of bytes to read.
306    *
307    * @exception IOException If <code>close()</code> was called on this Piped
308    *                        InputStream.
309    */
310  public synchronized int read(byte[] buf, int offset, int len)
311    throws IOException
312  {
313    if (source == null)
314      throw new IOException ("Not connected");
315    if (closed)
316      throw new IOException ("Pipe closed");
317
318    // Don't block if nothing was requested.
319    if (len == 0)
320      return 0;
321
322    // If the buffer is empty, wait until there is something in the pipe
323    // to read.
324    try
325      {
326        while (in < 0)
327          {
328            if (source.closed)
329              return -1;
330            wait();
331          }
332      }
333    catch (InterruptedException ix)
334      {
335        throw new InterruptedIOException();
336      }
337
338    int total = 0;
339    int copylen;
340
341    while (true)
342      {
343        // Figure out how many bytes from the pipe can be copied without
344        // overrunning in or going past the length of buf.
345        if (out < in)
346          copylen = Math.min (len, in - out);
347        else
348          copylen = Math.min (len, buffer.length - out);
349
350        System.arraycopy (buffer, out, buf, offset, copylen);
351        offset += copylen;
352        len -= copylen;
353        out += copylen;
354        total += copylen;
355
356        if (out == buffer.length)
357          out = 0;
358
359        if (out == in)
360          {
361            // Pipe is now empty.
362            in = -1;
363            out = 0;
364          }
365
366        // If output buffer is filled or the pipe is empty, we're done.
367        if (len == 0 || in == -1)
368          {
369            // Notify any waiting outputstream that there is now space
370            // to write.
371            notifyAll();
372            return total;
373          }
374      }
375  }
376
377  /**
378    * This method returns the number of bytes that can be read from this stream
379    * before blocking could occur.  This is the number of bytes that are
380    * currently unread in the internal circular buffer.  Note that once this
381    * many additional bytes are read, the stream may block on a subsequent
382    * read, but it not guaranteed to block.
383    *
384    * @return The number of bytes that can be read before blocking might occur
385    *
386    * @exception IOException If an error occurs
387    */
388  public synchronized int available() throws IOException
389  {
390    // The JDK 1.3 implementation does not appear to check for the closed or
391    // unconnected stream conditions here.
392
393    if (in < 0)
394      return 0;
395    else if (out < in)
396      return in - out;
397    else
398      return (buffer.length - out) + in;
399  }
400
401  /**
402  * This methods closes the stream so that no more data can be read
403  * from it.
404  *
405  * @exception IOException If an error occurs
406  */
407  public synchronized void close() throws IOException
408  {
409    closed = true;
410    // Wake any thread which may be in receive() waiting to write data.
411    notifyAll();
412  }
413}