Source Code Cross Referenced for PipedInputStream.java in  » 6.0-JDK-Core » io-nio » java » io » Java Source Code / Java DocumentationJava Source Code and Java Documentation

Home
Java Source Code / Java Documentation
1.6.0 JDK Core
2.6.0 JDK Modules
3.6.0 JDK Modules com.sun
4.6.0 JDK Modules com.sun.java
5.6.0 JDK Modules sun
6.6.0 JDK Platform
7.Ajax
8.Apache Harmony Java SE
9.Aspect oriented
10.Authentication Authorization
11.Blogger System
12.Build
13.Byte Code
14.Cache
15.Chart
16.Chat
17.Code Analyzer
18.Collaboration
19.Content Management System
20.Database Client
21.Database DBMS
22.Database JDBC Connection Pool
23.Database ORM
24.Development
25.EJB Server
26.ERP CRM Financial
27.ESB
28.Forum
29.Game
30.GIS
31.Graphic 3D
32.Graphic Library
33.Groupware
34.HTML Parser
35.IDE
36.IDE Eclipse
37.IDE Netbeans
38.Installer
39.Internationalization Localization
40.Inversion of Control
41.Issue Tracking
42.J2EE
43.J2ME
44.JBoss
45.JMS
46.JMX
47.Library
48.Mail Clients
49.Music
50.Net
51.Parser
52.PDF
53.Portal
54.Profiler
55.Project Management
56.Report
57.RSS RDF
58.Rule Engine
59.Science
60.Scripting
61.Search Engine
62.Security
63.Sevlet Container
64.Source Control
65.Swing Library
66.Template Engine
67.Test Coverage
68.Testing
69.UML
70.Web Crawler
71.Web Framework
72.Web Mail
73.Web Server
74.Web Services
75.Web Services apache cxf 2.2.6
76.Web Services AXIS2
77.Wiki Engine
78.Workflow Engines
79.XML
80.XML UI
Java Source Code / Java Documentation » 6.0 JDK Core » io nio » java.io 
Source Cross Referenced  Class Diagram Java Document (Java Doc) 


001        /*
002         * Copyright 1995-2006 Sun Microsystems, Inc.  All Rights Reserved.
003         * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
004         *
005         * This code is free software; you can redistribute it and/or modify it
006         * under the terms of the GNU General Public License version 2 only, as
007         * published by the Free Software Foundation.  Sun designates this
008         * particular file as subject to the "Classpath" exception as provided
009         * by Sun in the LICENSE file that accompanied this code.
010         *
011         * This code is distributed in the hope that it will be useful, but WITHOUT
012         * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
013         * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
014         * version 2 for more details (a copy is included in the LICENSE file that
015         * accompanied this code).
016         *
017         * You should have received a copy of the GNU General Public License version
018         * 2 along with this work; if not, write to the Free Software Foundation,
019         * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
020         *
021         * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
022         * CA 95054 USA or visit www.sun.com if you need additional information or
023         * have any questions.
024         */
025
026        package java.io;
027
028        /**
029         * A piped input stream should be connected
030         * to a piped output stream; the piped  input
031         * stream then provides whatever data bytes
032         * are written to the piped output  stream.
033         * Typically, data is read from a <code>PipedInputStream</code>
034         * object by one thread  and data is written
035         * to the corresponding <code>PipedOutputStream</code>
036         * by some  other thread. Attempting to use
037         * both objects from a single thread is not
038         * recommended, as it may deadlock the thread.
039         * The piped input stream contains a buffer,
040         * decoupling read operations from write operations,
041         * within limits.
042         * A pipe is said to be <a name=BROKEN> <i>broken</i> </a> if a
043         * thread that was providing data bytes to the connected
044         * piped output stream is no longer alive.
045         *
046         * @author  James Gosling
047         * @version 1.40, 12/01/05
048         * @see     java.io.PipedOutputStream
049         * @since   JDK1.0
050         */
051        public class PipedInputStream extends InputStream {
052            boolean closedByWriter = false;
053            volatile boolean closedByReader = false;
054            boolean connected = false;
055
056            /* REMIND: identification of the read and write sides needs to be
057               more sophisticated.  Either using thread groups (but what about
058               pipes within a thread?) or using finalization (but it may be a
059               long time until the next GC). */
060            Thread readSide;
061            Thread writeSide;
062
063            private static final int DEFAULT_PIPE_SIZE = 1024;
064
065            /**
066             * The default size of the pipe's circular input buffer.
067             * @since   JDK1.1
068             */
069            // This used to be a constant before the pipe size was allowed
070            // to change. This field will continue to be maintained
071            // for backward compatibility. 
072            protected static final int PIPE_SIZE = DEFAULT_PIPE_SIZE;
073
074            /**
075             * The circular buffer into which incoming data is placed.
076             * @since   JDK1.1
077             */
078            protected byte buffer[];
079
080            /**
081             * The index of the position in the circular buffer at which the
082             * next byte of data will be stored when received from the connected
083             * piped output stream. <code>in&lt;0</code> implies the buffer is empty,
084             * <code>in==out</code> implies the buffer is full
085             * @since   JDK1.1
086             */
087            protected int in = -1;
088
089            /**
090             * The index of the position in the circular buffer at which the next
091             * byte of data will be read by this piped input stream.
092             * @since   JDK1.1
093             */
094            protected int out = 0;
095
096            /**
097             * Creates a <code>PipedInputStream</code> so
098             * that it is connected to the piped output
099             * stream <code>src</code>. Data bytes written
100             * to <code>src</code> will then be  available
101             * as input from this stream.
102             *
103             * @param      src   the stream to connect to.
104             * @exception  IOException  if an I/O error occurs.
105             */
106            public PipedInputStream(PipedOutputStream src) throws IOException {
107                this (src, DEFAULT_PIPE_SIZE);
108            }
109
110            /**
111             * Creates a <code>PipedInputStream</code> so that it is
112             * connected to the piped output stream
113             * <code>src</code> and uses the specified pipe size for
114             * the pipe's buffer.
115             * Data bytes written to <code>src</code> will then
116             * be available as input from this stream.
117             *
118             * @param      src   the stream to connect to.
119             * @param      pipeSize the size of the pipe's buffer.
120             * @exception  IOException  if an I/O error occurs.
121             * @exception  IllegalArgumentException if <code>pipeSize <= 0</code>.
122             * @since	   1.6
123             */
124            public PipedInputStream(PipedOutputStream src, int pipeSize)
125                    throws IOException {
126                initPipe(pipeSize);
127                connect(src);
128            }
129
130            /**
131             * Creates a <code>PipedInputStream</code> so
132             * that it is not yet {@linkplain #connect(java.io.PipedOutputStream)
133             * connected}.
134             * It must be {@linkplain java.io.PipedOutputStream#connect(
135             * java.io.PipedInputStream) connected} to a
136             * <code>PipedOutputStream</code> before being used.
137             */
138            public PipedInputStream() {
139                initPipe(DEFAULT_PIPE_SIZE);
140            }
141
142            /**
143             * Creates a <code>PipedInputStream</code> so that it is not yet
144             * {@linkplain #connect(java.io.PipedOutputStream) connected} and
145             * uses the specified pipe size for the pipe's buffer.
146             * It must be {@linkplain java.io.PipedOutputStream#connect(
147             * java.io.PipedInputStream)
148             * connected} to a <code>PipedOutputStream</code> before being used.
149             *
150             * @param      pipeSize the size of the pipe's buffer.
151             * @exception  IllegalArgumentException if <code>pipeSize <= 0</code>.
152             * @since	   1.6
153             */
154            public PipedInputStream(int pipeSize) {
155                initPipe(pipeSize);
156            }
157
158            private void initPipe(int pipeSize) {
159                if (pipeSize <= 0) {
160                    throw new IllegalArgumentException("Pipe Size <= 0");
161                }
162                buffer = new byte[pipeSize];
163            }
164
165            /**
166             * Causes this piped input stream to be connected
167             * to the piped  output stream <code>src</code>.
168             * If this object is already connected to some
169             * other piped output  stream, an <code>IOException</code>
170             * is thrown.
171             * <p>
172             * If <code>src</code> is an
173             * unconnected piped output stream and <code>snk</code>
174             * is an unconnected piped input stream, they
175             * may be connected by either the call:
176             * <p>
177             * <pre><code>snk.connect(src)</code> </pre>
178             * <p>
179             * or the call:
180             * <p>
181             * <pre><code>src.connect(snk)</code> </pre>
182             * <p>
183             * The two
184             * calls have the same effect.
185             *
186             * @param      src   The piped output stream to connect to.
187             * @exception  IOException  if an I/O error occurs.
188             */
189            public void connect(PipedOutputStream src) throws IOException {
190                src.connect(this );
191            }
192
193            /**
194             * Receives a byte of data.  This method will block if no input is
195             * available.
196             * @param b the byte being received
197             * @exception IOException If the pipe is <a href=#BROKEN> <code>broken</code></a>,
198             *		{@link #connect(java.io.PipedOutputStream) unconnected},
199             *		closed, or if an I/O error occurs.
200             * @since     JDK1.1
201             */
202            protected synchronized void receive(int b) throws IOException {
203                checkStateForReceive();
204                writeSide = Thread.currentThread();
205                if (in == out)
206                    awaitSpace();
207                if (in < 0) {
208                    in = 0;
209                    out = 0;
210                }
211                buffer[in++] = (byte) (b & 0xFF);
212                if (in >= buffer.length) {
213                    in = 0;
214                }
215            }
216
217            /**
218             * Receives data into an array of bytes.  This method will
219             * block until some input is available.
220             * @param b the buffer into which the data is received
221             * @param off the start offset of the data
222             * @param len the maximum number of bytes received
223             * @exception IOException If the pipe is <a href=#BROKEN> broken</a>,
224             * 		 {@link #connect(java.io.PipedOutputStream) unconnected},
225             *		 closed,or if an I/O error occurs.
226             */
227            synchronized void receive(byte b[], int off, int len)
228                    throws IOException {
229                checkStateForReceive();
230                writeSide = Thread.currentThread();
231                int bytesToTransfer = len;
232                while (bytesToTransfer > 0) {
233                    if (in == out)
234                        awaitSpace();
235                    int nextTransferAmount = 0;
236                    if (out < in) {
237                        nextTransferAmount = buffer.length - in;
238                    } else if (in < out) {
239                        if (in == -1) {
240                            in = out = 0;
241                            nextTransferAmount = buffer.length - in;
242                        } else {
243                            nextTransferAmount = out - in;
244                        }
245                    }
246                    if (nextTransferAmount > bytesToTransfer)
247                        nextTransferAmount = bytesToTransfer;
248                    assert (nextTransferAmount > 0);
249                    System.arraycopy(b, off, buffer, in, nextTransferAmount);
250                    bytesToTransfer -= nextTransferAmount;
251                    off += nextTransferAmount;
252                    in += nextTransferAmount;
253                    if (in >= buffer.length) {
254                        in = 0;
255                    }
256                }
257            }
258
259            private void checkStateForReceive() throws IOException {
260                if (!connected) {
261                    throw new IOException("Pipe not connected");
262                } else if (closedByWriter || closedByReader) {
263                    throw new IOException("Pipe closed");
264                } else if (readSide != null && !readSide.isAlive()) {
265                    throw new IOException("Read end dead");
266                }
267            }
268
269            private void awaitSpace() throws IOException {
270                while (in == out) {
271                    checkStateForReceive();
272
273                    /* full: kick any waiting readers */
274                    notifyAll();
275                    try {
276                        wait(1000);
277                    } catch (InterruptedException ex) {
278                        throw new java.io.InterruptedIOException();
279                    }
280                }
281            }
282
283            /**
284             * Notifies all waiting threads that the last byte of data has been
285             * received.
286             */
287            synchronized void receivedLast() {
288                closedByWriter = true;
289                notifyAll();
290            }
291
292            /**
293             * Reads the next byte of data from this piped input stream. The
294             * value byte is returned as an <code>int</code> in the range
295             * <code>0</code> to <code>255</code>. 
296             * This method blocks until input data is available, the end of the
297             * stream is detected, or an exception is thrown.
298             *
299             * @return     the next byte of data, or <code>-1</code> if the end of the
300             *             stream is reached.
301             * @exception  IOException  if the pipe is
302             *		 {@link #connect(java.io.PipedOutputStream) unconnected},
303             *		 <a href=#BROKEN> <code>broken</code></a>, closed,
304             *		 or if an I/O error occurs.
305             */
306            public synchronized int read() throws IOException {
307                if (!connected) {
308                    throw new IOException("Pipe not connected");
309                } else if (closedByReader) {
310                    throw new IOException("Pipe closed");
311                } else if (writeSide != null && !writeSide.isAlive()
312                        && !closedByWriter && (in < 0)) {
313                    throw new IOException("Write end dead");
314                }
315
316                readSide = Thread.currentThread();
317                int trials = 2;
318                while (in < 0) {
319                    if (closedByWriter) {
320                        /* closed by writer, return EOF */
321                        return -1;
322                    }
323                    if ((writeSide != null) && (!writeSide.isAlive())
324                            && (--trials < 0)) {
325                        throw new IOException("Pipe broken");
326                    }
327                    /* might be a writer waiting */
328                    notifyAll();
329                    try {
330                        wait(1000);
331                    } catch (InterruptedException ex) {
332                        throw new java.io.InterruptedIOException();
333                    }
334                }
335                int ret = buffer[out++] & 0xFF;
336                if (out >= buffer.length) {
337                    out = 0;
338                }
339                if (in == out) {
340                    /* now empty */
341                    in = -1;
342                }
343
344                return ret;
345            }
346
347            /**
348             * Reads up to <code>len</code> bytes of data from this piped input
349             * stream into an array of bytes. Less than <code>len</code> bytes
350             * will be read if the end of the data stream is reached or if 
351             * <code>len</code> exceeds the pipe's buffer size.
352             * If <code>len </code> is zero, then no bytes are read and 0 is returned; 
353             * otherwise, the method blocks until at least 1 byte of input is 
354             * available, end of the stream has been detected, or an exception is
355             * thrown.
356             *
357             * @param      b     the buffer into which the data is read.
358             * @param      off   the start offset in the destination array <code>b</code>
359             * @param      len   the maximum number of bytes read.
360             * @return     the total number of bytes read into the buffer, or
361             *             <code>-1</code> if there is no more data because the end of
362             *             the stream has been reached.
363             * @exception  NullPointerException If <code>b</code> is <code>null</code>.
364             * @exception  IndexOutOfBoundsException If <code>off</code> is negative, 
365             * <code>len</code> is negative, or <code>len</code> is greater than 
366             * <code>b.length - off</code>
367             * @exception  IOException if the pipe is <a href=#BROKEN> <code>broken</code></a>,
368             *		 {@link #connect(java.io.PipedOutputStream) unconnected},
369             *		 closed, or if an I/O error occurs.
370             */
371            public synchronized int read(byte b[], int off, int len)
372                    throws IOException {
373                if (b == null) {
374                    throw new NullPointerException();
375                } else if (off < 0 || len < 0 || len > b.length - off) {
376                    throw new IndexOutOfBoundsException();
377                } else if (len == 0) {
378                    return 0;
379                }
380
381                /* possibly wait on the first character */
382                int c = read();
383                if (c < 0) {
384                    return -1;
385                }
386                b[off] = (byte) c;
387                int rlen = 1;
388                while ((in >= 0) && (len > 1)) {
389
390                    int available;
391
392                    if (in > out) {
393                        available = Math.min((buffer.length - out), (in - out));
394                    } else {
395                        available = buffer.length - out;
396                    }
397
398                    // A byte is read beforehand outside the loop
399                    if (available > (len - 1)) {
400                        available = len - 1;
401                    }
402                    System.arraycopy(buffer, out, b, off + rlen, available);
403                    out += available;
404                    rlen += available;
405                    len -= available;
406
407                    if (out >= buffer.length) {
408                        out = 0;
409                    }
410                    if (in == out) {
411                        /* now empty */
412                        in = -1;
413                    }
414                }
415                return rlen;
416            }
417
418            /**
419             * Returns the number of bytes that can be read from this input
420             * stream without blocking.
421             *
422             * @return the number of bytes that can be read from this input stream
423             *         without blocking, or {@code 0} if this input stream has been
424             *         closed by invoking its {@link #close()} method, or if the pipe
425             *	       is {@link #connect(java.io.PipedOutputStream) unconnected}, or
426             *		<a href=#BROKEN> <code>broken</code></a>.
427             *
428             * @exception  IOException  if an I/O error occurs.
429             * @since   JDK1.0.2
430             */
431            public synchronized int available() throws IOException {
432                if (in < 0)
433                    return 0;
434                else if (in == out)
435                    return buffer.length;
436                else if (in > out)
437                    return in - out;
438                else
439                    return in + buffer.length - out;
440            }
441
442            /**
443             * Closes this piped input stream and releases any system resources
444             * associated with the stream.
445             *
446             * @exception  IOException  if an I/O error occurs.
447             */
448            public void close() throws IOException {
449                closedByReader = true;
450                synchronized (this ) {
451                    in = -1;
452                }
453            }
454        }
www.java2java.com | Contact Us
Copyright 2009 - 12 Demo Source and Support. All rights reserved.
All other trademarks are property of their respective owners.