View Javadoc
1   /*
2    * Copyright 2012 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    *   http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  package io.netty.buffer;
17  
18  import io.netty.util.ReferenceCounted;
19  
20  import java.io.DataInput;
21  import java.io.DataInputStream;
22  import java.io.EOFException;
23  import java.io.IOException;
24  import java.io.InputStream;
25  
26  /**
27   * An {@link InputStream} which reads data from a {@link ByteBuf}.
28   * <p>
29   * A read operation against this stream will occur at the {@code readerIndex}
30   * of its underlying buffer and the {@code readerIndex} will increase during
31   * the read operation.  Please note that it only reads up to the number of
32   * readable bytes determined at the moment of construction.  Therefore,
33   * updating {@link ByteBuf#writerIndex()} will not affect the return
34   * value of {@link #available()}.
35   * <p>
36   * This stream implements {@link DataInput} for your convenience.
37   * The endianness of the stream is not always big endian but depends on
38   * the endianness of the underlying buffer.
39   *
40   * @see ByteBufOutputStream
41   */
42  public class ByteBufInputStream extends InputStream implements DataInput {
43      private final ByteBuf buffer;
44      private final int startIndex;
45      private final int endIndex;
46      private boolean closed;
47      /**
48       * To preserve backwards compatibility (which didn't transfer ownership) we support a conditional flag which
49       * indicates if {@link #buffer} should be released when this {@link InputStream} is closed.
50       * However in future releases ownership should always be transferred and callers of this class should call
51       * {@link ReferenceCounted#retain()} if necessary.
52       */
53      private final boolean releaseOnClose;
54  
55      /**
56       * Creates a new stream which reads data from the specified {@code buffer}
57       * starting at the current {@code readerIndex} and ending at the current
58       * {@code writerIndex}.
59       * @param buffer The buffer which provides the content for this {@link InputStream}.
60       */
61      public ByteBufInputStream(ByteBuf buffer) {
62          this(buffer, buffer.readableBytes());
63      }
64  
65      /**
66       * Creates a new stream which reads data from the specified {@code buffer}
67       * starting at the current {@code readerIndex} and ending at
68       * {@code readerIndex + length}.
69       * @param buffer The buffer which provides the content for this {@link InputStream}.
70       * @param length The length of the buffer to use for this {@link InputStream}.
71       * @throws IndexOutOfBoundsException
72       *         if {@code readerIndex + length} is greater than
73       *            {@code writerIndex}
74       */
75      public ByteBufInputStream(ByteBuf buffer, int length) {
76          this(buffer, length, false);
77      }
78  
79      /**
80       * Creates a new stream which reads data from the specified {@code buffer}
81       * starting at the current {@code readerIndex} and ending at the current
82       * {@code writerIndex}.
83       * @param buffer The buffer which provides the content for this {@link InputStream}.
84       * @param releaseOnClose {@code true} means that when {@link #close()} is called then {@link ByteBuf#release()} will
85       *                       be called on {@code buffer}.
86       */
87      public ByteBufInputStream(ByteBuf buffer, boolean releaseOnClose) {
88          this(buffer, buffer.readableBytes(), releaseOnClose);
89      }
90  
91      /**
92       * Creates a new stream which reads data from the specified {@code buffer}
93       * starting at the current {@code readerIndex} and ending at
94       * {@code readerIndex + length}.
95       * @param buffer The buffer which provides the content for this {@link InputStream}.
96       * @param length The length of the buffer to use for this {@link InputStream}.
97       * @param releaseOnClose {@code true} means that when {@link #close()} is called then {@link ByteBuf#release()} will
98       *                       be called on {@code buffer}.
99       * @throws IndexOutOfBoundsException
100      *         if {@code readerIndex + length} is greater than
101      *            {@code writerIndex}
102      */
103     public ByteBufInputStream(ByteBuf buffer, int length, boolean releaseOnClose) {
104         if (buffer == null) {
105             throw new NullPointerException("buffer");
106         }
107         if (length < 0) {
108             if (releaseOnClose) {
109                 buffer.release();
110             }
111             throw new IllegalArgumentException("length: " + length);
112         }
113         if (length > buffer.readableBytes()) {
114             if (releaseOnClose) {
115                 buffer.release();
116             }
117             throw new IndexOutOfBoundsException("Too many bytes to be read - Needs "
118                     + length + ", maximum is " + buffer.readableBytes());
119         }
120 
121         this.releaseOnClose = releaseOnClose;
122         this.buffer = buffer;
123         startIndex = buffer.readerIndex();
124         endIndex = startIndex + length;
125         buffer.markReaderIndex();
126     }
127 
128     /**
129      * Returns the number of read bytes by this stream so far.
130      */
131     public int readBytes() {
132         return buffer.readerIndex() - startIndex;
133     }
134 
135     @Override
136     public void close() throws IOException {
137         try {
138             super.close();
139         } finally {
140             // The Closable interface says "If the stream is already closed then invoking this method has no effect."
141             if (releaseOnClose && !closed) {
142                 closed = true;
143                 buffer.release();
144             }
145         }
146     }
147 
148     @Override
149     public int available() throws IOException {
150         return endIndex - buffer.readerIndex();
151     }
152 
153     @Override
154     public void mark(int readlimit) {
155         buffer.markReaderIndex();
156     }
157 
158     @Override
159     public boolean markSupported() {
160         return true;
161     }
162 
163     @Override
164     public int read() throws IOException {
165         if (!buffer.isReadable()) {
166             return -1;
167         }
168         return buffer.readByte() & 0xff;
169     }
170 
171     @Override
172     public int read(byte[] b, int off, int len) throws IOException {
173         int available = available();
174         if (available == 0) {
175             return -1;
176         }
177 
178         len = Math.min(available, len);
179         buffer.readBytes(b, off, len);
180         return len;
181     }
182 
183     @Override
184     public void reset() throws IOException {
185         buffer.resetReaderIndex();
186     }
187 
188     @Override
189     public long skip(long n) throws IOException {
190         if (n > Integer.MAX_VALUE) {
191             return skipBytes(Integer.MAX_VALUE);
192         } else {
193             return skipBytes((int) n);
194         }
195     }
196 
197     @Override
198     public boolean readBoolean() throws IOException {
199         checkAvailable(1);
200         return read() != 0;
201     }
202 
203     @Override
204     public byte readByte() throws IOException {
205         if (!buffer.isReadable()) {
206             throw new EOFException();
207         }
208         return buffer.readByte();
209     }
210 
211     @Override
212     public char readChar() throws IOException {
213         return (char) readShort();
214     }
215 
216     @Override
217     public double readDouble() throws IOException {
218         return Double.longBitsToDouble(readLong());
219     }
220 
221     @Override
222     public float readFloat() throws IOException {
223         return Float.intBitsToFloat(readInt());
224     }
225 
226     @Override
227     public void readFully(byte[] b) throws IOException {
228         readFully(b, 0, b.length);
229     }
230 
231     @Override
232     public void readFully(byte[] b, int off, int len) throws IOException {
233         checkAvailable(len);
234         buffer.readBytes(b, off, len);
235     }
236 
237     @Override
238     public int readInt() throws IOException {
239         checkAvailable(4);
240         return buffer.readInt();
241     }
242 
243     private final StringBuilder lineBuf = new StringBuilder();
244 
245     @Override
246     public String readLine() throws IOException {
247         lineBuf.setLength(0);
248 
249         loop: while (true) {
250             if (!buffer.isReadable()) {
251                 return lineBuf.length() > 0 ? lineBuf.toString() : null;
252             }
253 
254             int c = buffer.readUnsignedByte();
255             switch (c) {
256                 case '\n':
257                     break loop;
258 
259                 case '\r':
260                     if (buffer.isReadable() && (char) buffer.getUnsignedByte(buffer.readerIndex()) == '\n') {
261                         buffer.skipBytes(1);
262                     }
263                     break loop;
264 
265                 default:
266                     lineBuf.append((char) c);
267             }
268         }
269 
270         return lineBuf.toString();
271     }
272 
273     @Override
274     public long readLong() throws IOException {
275         checkAvailable(8);
276         return buffer.readLong();
277     }
278 
279     @Override
280     public short readShort() throws IOException {
281         checkAvailable(2);
282         return buffer.readShort();
283     }
284 
285     @Override
286     public String readUTF() throws IOException {
287         return DataInputStream.readUTF(this);
288     }
289 
290     @Override
291     public int readUnsignedByte() throws IOException {
292         return readByte() & 0xff;
293     }
294 
295     @Override
296     public int readUnsignedShort() throws IOException {
297         return readShort() & 0xffff;
298     }
299 
300     @Override
301     public int skipBytes(int n) throws IOException {
302         int nBytes = Math.min(available(), n);
303         buffer.skipBytes(nBytes);
304         return nBytes;
305     }
306 
307     private void checkAvailable(int fieldSize) throws IOException {
308         if (fieldSize < 0) {
309             throw new IndexOutOfBoundsException("fieldSize cannot be a negative number");
310         }
311         if (fieldSize > available()) {
312             throw new EOFException("fieldSize is too long! Length is " + fieldSize
313                     + ", but maximum is " + available());
314         }
315     }
316 }