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