001    /* PipedInputStream.java -- Read portion of piped streams.
002       Copyright (C) 1998, 1999, 2000, 2001, 2003, 2005  Free Software Foundation, Inc.
003    
004    This file is part of GNU Classpath.
005    
006    GNU Classpath is free software; you can redistribute it and/or modify
007    it under the terms of the GNU General Public License as published by
008    the Free Software Foundation; either version 2, or (at your option)
009    any later version.
010     
011    GNU Classpath is distributed in the hope that it will be useful, but
012    WITHOUT ANY WARRANTY; without even the implied warranty of
013    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
014    General Public License for more details.
015    
016    You should have received a copy of the GNU General Public License
017    along with GNU Classpath; see the file COPYING.  If not, write to the
018    Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
019    02110-1301 USA.
020    
021    Linking this library statically or dynamically with other modules is
022    making a combined work based on this library.  Thus, the terms and
023    conditions of the GNU General Public License cover the whole
024    combination.
025    
026    As a special exception, the copyright holders of this library give you
027    permission to link this library with independent modules to produce an
028    executable, regardless of the license terms of these independent
029    modules, and to copy and distribute the resulting executable under
030    terms of your choice, provided that you also meet, for each linked
031    independent module, the terms and conditions of the license of that
032    module.  An independent module is a module which is not derived from
033    or based on this library.  If you modify this library, you may extend
034    this exception to your version of the library, but you are not
035    obligated to do so.  If you do not wish to do so, delete this
036    exception statement from your version. */
037    
038    package 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      */
062    public 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 = new byte[PIPE_SIZE];
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      }
111    
112      /**
113        * This constructor creates a new <code>PipedInputStream</code> and connects
114        * it to the passed in <code>PipedOutputStream</code>. The stream is then 
115        * ready for reading.
116        *
117        * @param source The <code>PipedOutputStream</code> to connect this 
118        * stream to
119        *
120        * @exception IOException If <code>source</code> is already connected.
121        */
122      public PipedInputStream(PipedOutputStream source) throws IOException
123      {
124        connect(source);
125      }
126    
127      /**
128        * This method connects this stream to the passed in 
129        * <code>PipedOutputStream</code>.
130        * This stream is then ready for reading.  If this stream is already
131        * connected or has been previously closed, then an exception is thrown
132        *
133        * @param source The <code>PipedOutputStream</code> to connect this stream to
134        *
135        * @exception IOException If this PipedInputStream or <code>source</code> 
136        *                        has been connected already.
137        */
138      public void connect(PipedOutputStream source) throws IOException
139      {
140        // The JDK (1.3) does not appear to check for a previously closed 
141        // connection here.
142        
143        if (this.source != null || source.sink != null)
144          throw new IOException ("Already connected");
145        
146        source.sink = this;
147        this.source = source;
148      }
149      
150      /**
151      * This method receives a byte of input from the source PipedOutputStream.
152      * If the internal circular buffer is full, this method blocks.
153      *
154      * @param val The byte to write to this stream
155      *
156      * @exception IOException if error occurs
157      * @specnote Weird. This method must be some sort of accident.
158      */
159      protected synchronized void receive(int val) throws IOException
160      {
161        read_buf[0] = (byte) (val & 0xff);
162        receive (read_buf, 0, 1);
163      }
164    
165      /**
166        * This method is used by the connected <code>PipedOutputStream</code> to
167        * write bytes into the buffer.
168        *
169        * @param buf The array containing bytes to write to this stream
170        * @param offset The offset into the array to start writing from
171        * @param len The number of bytes to write.
172        *
173        * @exception IOException If an error occurs
174        * @specnote This code should be in PipedOutputStream.write, but we
175        *           put it here in order to support that bizarre recieve(int)
176        *           method.
177        */  
178      synchronized void receive(byte[] buf, int offset, int len)
179        throws IOException
180      {
181        if (closed)
182          throw new IOException ("Pipe closed");
183    
184        int bufpos = offset;
185        int copylen;
186        
187        while (len > 0)
188          {
189            try
190              {
191                while (in == out)
192                  {
193                    // The pipe is full. Wake up any readers and wait for them.
194                    notifyAll();
195                    wait();
196                    // The pipe could have been closed while we were waiting.
197                    if (closed)
198                      throw new IOException ("Pipe closed");
199                  }
200              }
201            catch (InterruptedException ix)
202              {
203                throw new InterruptedIOException ();
204              }
205    
206            if (in < 0) // The pipe is empty.
207              in = 0;
208            
209            // Figure out how many bytes from buf can be copied without 
210            // overrunning out or going past the length of the buffer.
211            if (in < out)
212              copylen = Math.min (len, out - in);
213            else
214              copylen = Math.min (len, buffer.length - in);
215    
216            // Copy bytes until the pipe is filled, wrapping if necessary.
217            System.arraycopy(buf, bufpos, buffer, in, copylen);
218            len -= copylen;
219            bufpos += copylen;
220            in += copylen;
221            if (in == buffer.length)
222              in = 0;
223          }
224        // Notify readers that new data is in the pipe.
225        notifyAll();
226      }
227      
228      /**
229        * This method reads one byte from the stream.
230        * -1 is returned to indicated that no bytes can be read
231        * because the end of the stream was reached.  If the stream is already
232        * closed, a -1 will again be returned to indicate the end of the stream.
233        * 
234        * <p>This method will block if no byte is available to be read.</p>
235        *
236        * @return the value of the read byte value, or -1 of the end of the stream
237        * was reached
238        * 
239        * @throws IOException if an error occured
240        */
241      public int read() throws IOException
242      {
243        // Method operates by calling the multibyte overloaded read method
244        // Note that read_buf is an internal instance variable.  I allocate it
245        // there to avoid constant reallocation overhead for applications that
246        // call this method in a loop at the cost of some unneeded overhead
247        // if this method is never called.
248    
249        int r = read(read_buf, 0, 1);
250        return r != -1 ? (read_buf[0] & 0xff) : -1;
251      }
252      
253      /**
254        * This method reads bytes from the stream into a caller supplied buffer.
255        * It starts storing bytes at position <code>offset</code> into the 
256        * buffer and
257        * reads a maximum of <code>len</code> bytes.  Note that this method 
258        * can actually
259        * read fewer than <code>len</code> bytes.  The actual number of bytes 
260        * read is
261        * returned.  A -1 is returned to indicated that no bytes can be read
262        * because the end of the stream was reached - ie close() was called on the
263        * connected PipedOutputStream.
264        * <p>
265        * This method will block if no bytes are available to be read.
266        *
267        * @param buf The buffer into which bytes will be stored
268        * @param offset The index into the buffer at which to start writing.
269        * @param len The maximum number of bytes to read.
270        *
271        * @exception IOException If <code>close()</code> was called on this Piped
272        *                        InputStream.
273        */  
274      public synchronized int read(byte[] buf, int offset, int len)
275        throws IOException
276      {
277        if (source == null)
278          throw new IOException ("Not connected");
279        if (closed)
280          throw new IOException ("Pipe closed");
281    
282        // Don't block if nothing was requested.
283        if (len == 0)
284          return 0;
285    
286        // If the buffer is empty, wait until there is something in the pipe 
287        // to read.
288        try
289          {
290            while (in < 0)
291              {
292                if (source.closed)
293                  return -1;
294                wait();
295              }
296          }
297        catch (InterruptedException ix)
298          {
299            throw new InterruptedIOException();
300          }
301        
302        int total = 0;
303        int copylen;
304        
305        while (true)
306          {
307            // Figure out how many bytes from the pipe can be copied without 
308            // overrunning in or going past the length of buf.
309            if (out < in)
310              copylen = Math.min (len, in - out);
311            else
312              copylen = Math.min (len, buffer.length - out);
313    
314            System.arraycopy (buffer, out, buf, offset, copylen);
315            offset += copylen;
316            len -= copylen;
317            out += copylen;
318            total += copylen;
319            
320            if (out == buffer.length)
321              out = 0;
322            
323            if (out == in)
324              {
325                // Pipe is now empty.
326                in = -1;
327                out = 0;
328              }
329    
330            // If output buffer is filled or the pipe is empty, we're done.
331            if (len == 0 || in == -1)
332              {
333                // Notify any waiting outputstream that there is now space
334                // to write.
335                notifyAll();
336                return total;
337              }
338          }
339      }
340      
341      /**
342        * This method returns the number of bytes that can be read from this stream
343        * before blocking could occur.  This is the number of bytes that are
344        * currently unread in the internal circular buffer.  Note that once this
345        * many additional bytes are read, the stream may block on a subsequent
346        * read, but it not guaranteed to block.
347        *
348        * @return The number of bytes that can be read before blocking might occur
349        *
350        * @exception IOException If an error occurs
351        */  
352      public synchronized int available() throws IOException
353      {
354        // The JDK 1.3 implementation does not appear to check for the closed or 
355        // unconnected stream conditions here.
356        
357        if (in < 0)
358          return 0;
359        else if (out < in)
360          return in - out;
361        else
362          return (buffer.length - out) + in;
363      }
364      
365      /**
366      * This methods closes the stream so that no more data can be read
367      * from it.
368      *
369      * @exception IOException If an error occurs
370      */
371      public synchronized void close() throws IOException
372      {
373        closed = true;
374        // Wake any thread which may be in receive() waiting to write data.
375        notifyAll();
376      }
377    }
378