KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > HTTPClient > RespInputStream


1 /*
2  * @(#)RespInputStream.java 0.3-2 18/06/1999
3  *
4  * This file is part of the HTTPClient package
5  * Copyright (C) 1996-1999 Ronald Tschalär
6  *
7  * This library is free software; you can redistribute it and/or
8  * modify it under the terms of the GNU Lesser General Public
9  * License as published by the Free Software Foundation; either
10  * version 2 of the License, or (at your option) any later version.
11  *
12  * This library is distributed in the hope that it will be useful,
13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15  * Lesser General Public License for more details.
16  *
17  * You should have received a copy of the GNU Lesser General Public
18  * License along with this library; if not, write to the Free
19  * Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
20  * MA 02111-1307, USA
21  *
22  * For questions, suggestions, bug-reports, enhancement-requests etc.
23  * I may be contacted at:
24  *
25  * ronald@innovation.ch
26  *
27  */
28
29 package HTTPClient;
30
31 import java.io.InputStream JavaDoc;
32 import java.io.IOException JavaDoc;
33 import java.io.InterruptedIOException JavaDoc;
34
35 /**
36  * This is the InputStream that gets returned to the user. The extensions
37  * consist of the capability to have the data pushed into a buffer if the
38  * stream demux needs to.
39  *
40  * @version 0.3-2 18/06/1999
41  * @author Ronald Tschalär
42  * @since V0.2
43  */
44 final class RespInputStream extends InputStream JavaDoc implements GlobalConstants
45 {
46     /** the stream demultiplexor */
47     private StreamDemultiplexor demux = null;
48
49     /** our response handler */
50     private ResponseHandler resph;
51
52     /** signals that the user has closed the stream and will therefore
53     not read any further data */
54         boolean closed = false;
55
56     /** signals that the connection may not be closed prematurely */
57     private boolean dont_truncate = false;
58
59     /** this buffer is used to buffer data that the demux has to get rid of */
60     private byte[] buffer = null;
61
62     /** signals that we were interrupted and that the buffer is not complete */
63     private boolean interrupted = false;
64
65     /** the offset at which the unread data starts in the buffer */
66     private int offset = 0;
67
68     /** the end of the data in the buffer */
69     private int end = 0;
70
71     /** the total number of bytes of entity data read from the demux so far */
72             int count = 0;
73
74
75     // Constructors
76
77     RespInputStream(StreamDemultiplexor demux, ResponseHandler resph)
78     {
79     this.demux = demux;
80     this.resph = resph;
81     }
82
83
84     // public Methods
85
86     private byte[] ch = new byte[1];
87     /**
88      * Reads a single byte.
89      *
90      * @return the byte read, or -1 if EOF.
91      * @exception IOException if any exception occured on the connection.
92      */
93     public synchronized int read() throws IOException JavaDoc
94     {
95     int rcvd = read(ch, 0, 1);
96     if (rcvd == 1)
97         return ch[0] & 0xff;
98     else
99         return -1;
100     }
101
102
103     /**
104      * Reads <var>len</var> bytes into <var>b</var>, starting at offset
105      * <var>off</var>.
106      *
107      * @return the number of bytes actually read, or -1 if EOF.
108      * @exception IOException if any exception occured on the connection.
109      */
110     public synchronized int read(byte[] b, int off, int len) throws IOException JavaDoc
111     {
112     if (closed)
113         return -1;
114
115     int left = end - offset;
116     if (buffer != null && !(left == 0 && interrupted))
117     {
118         if (left == 0) return -1;
119
120         len = (len > left ? left : len);
121         System.arraycopy(buffer, offset, b, off, len);
122         offset += len;
123
124         return len;
125     }
126     else
127     {
128         if (DebugDemux)
129         {
130         if (resph.resp.cd_type != CD_HDRS)
131             System.err.println("RspIS: Reading stream " +
132                        this.hashCode() +
133                        " (" + Thread.currentThread() + ")");
134         }
135
136         int rcvd;
137         if (resph.resp.cd_type == CD_HDRS)
138         rcvd = demux.read(b, off, len, resph, resph.resp.timeout);
139         else
140         rcvd = demux.read(b, off, len, resph, 0);
141         if (rcvd != -1 && resph.resp.got_headers)
142         count += rcvd;
143
144         return rcvd;
145     }
146     }
147
148
149     /**
150      * skips <var>num</var> bytes.
151      *
152      * @return the number of bytes actually skipped.
153      * @exception IOException if any exception occured on the connection.
154      */
155     public synchronized long skip(long num) throws IOException JavaDoc
156     {
157     if (closed)
158         return 0;
159
160     int left = end - offset;
161     if (buffer != null && !(left == 0 && interrupted))
162     {
163         num = (num > left ? left : num);
164         offset += num;
165         return num;
166     }
167     else
168     {
169         long skpd = demux.skip(num, resph);
170         if (resph.resp.got_headers)
171         count += skpd;
172         return skpd;
173     }
174     }
175
176
177     /**
178      * gets the number of bytes available for reading without blocking.
179      *
180      * @return the number of bytes available.
181      * @exception IOException if any exception occured on the connection.
182      */
183     public synchronized int available() throws IOException JavaDoc
184     {
185     if (closed)
186         return 0;
187
188     if (buffer != null && !(end-offset == 0 && interrupted))
189         return end-offset;
190     else
191         return demux.available(resph);
192     }
193
194
195     /**
196      * closes the stream.
197      *
198      * @exception if any exception occured on the connection before or
199      * during close.
200      */
201     public synchronized void close() throws IOException JavaDoc
202     {
203     if (!closed)
204     {
205         closed = true;
206
207         if (dont_truncate && (buffer == null || interrupted))
208         readAll(resph.resp.timeout);
209
210         if (DebugDemux)
211         System.err.println("RspIS: User closed stream " + hashCode() +
212                    " (" + Thread.currentThread() + ")");
213
214         demux.closeSocketIfAllStreamsClosed();
215
216         if (dont_truncate)
217         {
218         try
219             { resph.resp.http_resp.invokeTrailerHandlers(false); }
220         catch (ModuleException me)
221             { throw new IOException JavaDoc(me.toString()); }
222         }
223     }
224     }
225
226
227     /**
228      * A safety net to clean up.
229      */
230     protected void finalize() throws Throwable JavaDoc
231     {
232     try
233         { close(); }
234     finally
235         { super.finalize(); }
236     }
237
238
239     // local Methods
240
241     /**
242      * Reads all remainings data into buffer. This is used to force a read
243      * of upstream responses.
244      *
245      * <P>This is probably the most tricky and buggy method around. It's the
246      * only one that really violates the strict top-down method invocation
247      * from the Response through the ResponseStream to the StreamDemultiplexor.
248      * This means we need to be awfully careful about what is synchronized
249      * and what parameters are passed to whom.
250      *
251      * @param timeout the timeout to use for reading from the demux
252      * @exception IOException If any exception occurs while reading stream.
253      */
254     void readAll(int timeout) throws IOException JavaDoc
255     {
256     if (DebugDemux)
257         System.err.println("RspIS: Read-all on stream " + this.hashCode() +
258                    " (" + Thread.currentThread() + ")");
259
260     synchronized(resph.resp)
261     {
262         if (!resph.resp.got_headers) // force headers to be read
263 {
264         int sav_to = resph.resp.timeout;
265         resph.resp.timeout = timeout;
266         resph.resp.getStatusCode();
267         resph.resp.timeout = sav_to;
268         }
269     }
270
271     synchronized(this)
272     {
273         if (buffer != null && !interrupted) return;
274
275         int rcvd = 0;
276         try
277         {
278         if (closed) // throw away
279 {
280             buffer = new byte[10000];
281             do
282             {
283             count += rcvd;
284             rcvd = demux.read(buffer, 0, buffer.length, resph,
285                         timeout);
286             } while (rcvd != -1);
287             buffer = null;
288         }
289         else
290         {
291             if (buffer == null)
292             {
293             buffer = new byte[10000];
294             offset = 0;
295             end = 0;
296             }
297
298             do
299             {
300             rcvd = demux.read(buffer, end, buffer.length-end, resph,
301                       timeout);
302             if (rcvd < 0) break;
303
304             count += rcvd;
305             end += rcvd;
306             buffer = Util.resizeArray(buffer, end+10000);
307             } while (true);
308         }
309         }
310         catch (InterruptedIOException JavaDoc iioe)
311         {
312         interrupted = true;
313         throw iioe;
314         }
315         catch (IOException JavaDoc ioe)
316         {
317         buffer = null; // force a read on demux for exception
318 }
319
320         interrupted = false;
321     }
322     }
323
324
325     /**
326      * Sometime the full response body must be read, i.e. the connection may
327      * not be closed prematurely (by us). Currently this is needed when the
328      * chunked encoding with trailers is used in a response.
329      */
330     synchronized void dontTruncate()
331     {
332     dont_truncate = true;
333     }
334 }
335
336
Popular Tags