View Javadoc

1   /*
2    * ====================================================================
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *   http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing,
14   * software distributed under the License is distributed on an
15   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16   * KIND, either express or implied.  See the License for the
17   * specific language governing permissions and limitations
18   * under the License.
19   * ====================================================================
20   *
21   * This software consists of voluntary contributions made by many
22   * individuals on behalf of the Apache Software Foundation.  For more
23   * information on the Apache Software Foundation, please see
24   * <http://www.apache.org/>.
25   *
26   */
27  
28  package org.apache.http.impl.io;
29  
30  import java.io.IOException;
31  import java.io.InputStream;
32  
33  import org.apache.http.ConnectionClosedException;
34  import org.apache.http.io.BufferInfo;
35  import org.apache.http.io.SessionInputBuffer;
36  import org.apache.http.util.Args;
37  
38  /**
39   * Input stream that cuts off after a defined number of bytes. This class
40   * is used to receive content of HTTP messages where the end of the content
41   * entity is determined by the value of the {@code Content-Length header}.
42   * Entities transferred using this stream can be maximum {@link Long#MAX_VALUE}
43   * long.
44   * <p>
45   * Note that this class NEVER closes the underlying stream, even when close
46   * gets called.  Instead, it will read until the "end" of its limit on
47   * close, which allows for the seamless execution of subsequent HTTP 1.1
48   * requests, while not requiring the client to remember to read the entire
49   * contents of the response.
50   *
51   *
52   * @since 4.0
53   */
54  public class ContentLengthInputStream extends InputStream {
55  
56      private static final int BUFFER_SIZE = 2048;
57      /**
58       * The maximum number of bytes that can be read from the stream. Subsequent
59       * read operations will return -1.
60       */
61      private final long contentLength;
62  
63      /** The current position */
64      private long pos = 0;
65  
66      /** True if the stream is closed. */
67      private boolean closed = false;
68  
69      /**
70       * Wrapped input stream that all calls are delegated to.
71       */
72      private SessionInputBuffer in = null;
73  
74      /**
75       * Wraps a session input buffer and cuts off output after a defined number
76       * of bytes.
77       *
78       * @param in The session input buffer
79       * @param contentLength The maximum number of bytes that can be read from
80       * the stream. Subsequent read operations will return -1.
81       */
82      public ContentLengthInputStream(final SessionInputBuffer in, final long contentLength) {
83          super();
84          this.in = Args.notNull(in, "Session input buffer");
85          this.contentLength = Args.notNegative(contentLength, "Content length");
86      }
87  
88      /**
89       * <p>Reads until the end of the known length of content.</p>
90       *
91       * <p>Does not close the underlying socket input, but instead leaves it
92       * primed to parse the next response.</p>
93       * @throws IOException If an IO problem occurs.
94       */
95      @Override
96      public void close() throws IOException {
97          if (!closed) {
98              try {
99                  if (pos < contentLength) {
100                     final byte buffer[] = new byte[BUFFER_SIZE];
101                     while (read(buffer) >= 0) {
102                     }
103                 }
104             } finally {
105                 // close after above so that we don't throw an exception trying
106                 // to read after closed!
107                 closed = true;
108             }
109         }
110     }
111 
112     @Override
113     public int available() throws IOException {
114         if (this.in instanceof BufferInfo) {
115             final int len = ((BufferInfo) this.in).length();
116             return Math.min(len, (int) (this.contentLength - this.pos));
117         } else {
118             return 0;
119         }
120     }
121 
122     /**
123      * Read the next byte from the stream
124      * @return The next byte or -1 if the end of stream has been reached.
125      * @throws IOException If an IO problem occurs
126      * @see java.io.InputStream#read()
127      */
128     @Override
129     public int read() throws IOException {
130         if (closed) {
131             throw new IOException("Attempted read from closed stream.");
132         }
133 
134         if (pos >= contentLength) {
135             return -1;
136         }
137         final int b = this.in.read();
138         if (b == -1) {
139             if (pos < contentLength) {
140                 throw new ConnectionClosedException(
141                         "Premature end of Content-Length delimited message body (expected: "
142                         + contentLength + "; received: " + pos);
143             }
144         } else {
145             pos++;
146         }
147         return b;
148     }
149 
150     /**
151      * Does standard {@link InputStream#read(byte[], int, int)} behavior, but
152      * also notifies the watcher when the contents have been consumed.
153      *
154      * @param b     The byte array to fill.
155      * @param off   Start filling at this position.
156      * @param len   The number of bytes to attempt to read.
157      * @return The number of bytes read, or -1 if the end of content has been
158      *  reached.
159      *
160      * @throws java.io.IOException Should an error occur on the wrapped stream.
161      */
162     @Override
163     public int read (final byte[] b, final int off, final int len) throws java.io.IOException {
164         if (closed) {
165             throw new IOException("Attempted read from closed stream.");
166         }
167 
168         if (pos >= contentLength) {
169             return -1;
170         }
171 
172         int chunk = len;
173         if (pos + len > contentLength) {
174             chunk = (int) (contentLength - pos);
175         }
176         final int count = this.in.read(b, off, chunk);
177         if (count == -1 && pos < contentLength) {
178             throw new ConnectionClosedException(
179                     "Premature end of Content-Length delimited message body (expected: "
180                     + contentLength + "; received: " + pos);
181         }
182         if (count > 0) {
183             pos += count;
184         }
185         return count;
186     }
187 
188 
189     /**
190      * Read more bytes from the stream.
191      * @param b The byte array to put the new data in.
192      * @return The number of bytes read into the buffer.
193      * @throws IOException If an IO problem occurs
194      * @see java.io.InputStream#read(byte[])
195      */
196     @Override
197     public int read(final byte[] b) throws IOException {
198         return read(b, 0, b.length);
199     }
200 
201     /**
202      * Skips and discards a number of bytes from the input stream.
203      * @param n The number of bytes to skip.
204      * @return The actual number of bytes skipped. &le; 0 if no bytes
205      * are skipped.
206      * @throws IOException If an error occurs while skipping bytes.
207      * @see InputStream#skip(long)
208      */
209     @Override
210     public long skip(final long n) throws IOException {
211         if (n <= 0) {
212             return 0;
213         }
214         final byte[] buffer = new byte[BUFFER_SIZE];
215         // make sure we don't skip more bytes than are
216         // still available
217         long remaining = Math.min(n, this.contentLength - this.pos);
218         // skip and keep track of the bytes actually skipped
219         long count = 0;
220         while (remaining > 0) {
221             final int l = read(buffer, 0, (int)Math.min(BUFFER_SIZE, remaining));
222             if (l == -1) {
223                 break;
224             }
225             count += l;
226             remaining -= l;
227         }
228         return count;
229     }
230 }