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.handler.stream;
17
18 import static io.netty.util.internal.ObjectUtil.checkPositive;
19 import static io.netty.util.internal.ObjectUtil.checkNotNull;
20
21 import io.netty.buffer.ByteBuf;
22 import io.netty.buffer.ByteBufAllocator;
23 import io.netty.channel.ChannelHandlerContext;
24
25 import java.nio.ByteBuffer;
26 import java.nio.channels.ReadableByteChannel;
27
28 /**
29 * A {@link ChunkedInput} that fetches data from a {@link ReadableByteChannel}
30 * chunk by chunk. Please note that the {@link ReadableByteChannel} must
31 * operate in blocking mode. Non-blocking mode channels are not supported.
32 */
33 public class ChunkedNioStream implements ChunkedInput<ByteBuf> {
34
35 private final ReadableByteChannel in;
36
37 private final int chunkSize;
38 private long offset;
39
40 /**
41 * Associated ByteBuffer
42 */
43 private final ByteBuffer byteBuffer;
44
45 /**
46 * Creates a new instance that fetches data from the specified channel.
47 */
48 public ChunkedNioStream(ReadableByteChannel in) {
49 this(in, ChunkedStream.DEFAULT_CHUNK_SIZE);
50 }
51
52 /**
53 * Creates a new instance that fetches data from the specified channel.
54 *
55 * @param chunkSize the number of bytes to fetch on each
56 * {@link #readChunk(ChannelHandlerContext)} call
57 */
58 public ChunkedNioStream(ReadableByteChannel in, int chunkSize) {
59 this.in = checkNotNull(in, "in");
60 this.chunkSize = checkPositive(chunkSize, "chunkSize");
61 byteBuffer = ByteBuffer.allocate(chunkSize);
62 }
63
64 /**
65 * Returns the number of transferred bytes.
66 */
67 public long transferredBytes() {
68 return offset;
69 }
70
71 @Override
72 public boolean isEndOfInput() throws Exception {
73 if (byteBuffer.position() > 0) {
74 // A previous read was not over, so there is a next chunk in the buffer at least
75 return false;
76 }
77 if (in.isOpen()) {
78 // Try to read a new part, and keep this part (no rewind)
79 int b = in.read(byteBuffer);
80 if (b < 0) {
81 return true;
82 } else {
83 offset += b;
84 return false;
85 }
86 }
87 return true;
88 }
89
90 @Override
91 public void close() throws Exception {
92 in.close();
93 }
94
95 @Deprecated
96 @Override
97 public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
98 return readChunk(ctx.alloc());
99 }
100
101 @Override
102 public ByteBuf readChunk(ByteBufAllocator allocator) throws Exception {
103 if (isEndOfInput()) {
104 return null;
105 }
106 // buffer cannot be not be empty from there
107 int readBytes = byteBuffer.position();
108 for (;;) {
109 int localReadBytes = in.read(byteBuffer);
110 if (localReadBytes < 0) {
111 break;
112 }
113 readBytes += localReadBytes;
114 offset += localReadBytes;
115 if (readBytes == chunkSize) {
116 break;
117 }
118 }
119 byteBuffer.flip();
120 boolean release = true;
121 ByteBuf buffer = allocator.buffer(byteBuffer.remaining());
122 try {
123 buffer.writeBytes(byteBuffer);
124 byteBuffer.clear();
125 release = false;
126 return buffer;
127 } finally {
128 if (release) {
129 buffer.release();
130 }
131 }
132 }
133
134 @Override
135 public long length() {
136 return -1;
137 }
138
139 @Override
140 public long progress() {
141 return offset;
142 }
143 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！