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