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 io.netty.buffer.ByteBuf;
19 import io.netty.buffer.ByteBufAllocator;
20 import io.netty.channel.ChannelHandlerContext;
21 import io.netty.util.internal.ObjectUtil;
22
23 import java.io.InputStream;
24 import java.io.PushbackInputStream;
25
26 /**
27 * A {@link ChunkedInput} that fetches data from an {@link InputStream} chunk by
28 * chunk.
29 * <p>
30 * Please note that the {@link InputStream} instance that feeds data into
31 * {@link ChunkedStream} must implement {@link InputStream#available()} as
32 * accurately as possible, rather than using the default implementation.
33 * Otherwise, {@link ChunkedStream} will generate many too small chunks or
34 * block unnecessarily often.
35 */
36 public class ChunkedStream implements ChunkedInput<ByteBuf> {
37
38 static final int DEFAULT_CHUNK_SIZE = 8192;
39
40 private final PushbackInputStream in;
41 private final int chunkSize;
42 private long offset;
43 private boolean closed;
44
45 /**
46 * Creates a new instance that fetches data from the specified stream.
47 */
48 public ChunkedStream(InputStream in) {
49 this(in, DEFAULT_CHUNK_SIZE);
50 }
51
52 /**
53 * Creates a new instance that fetches data from the specified stream.
54 *
55 * @param chunkSize the number of bytes to fetch on each
56 * {@link #readChunk(ChannelHandlerContext)} call
57 */
58 public ChunkedStream(InputStream in, int chunkSize) {
59 ObjectUtil.checkNotNull(in, "in");
60 ObjectUtil.checkPositive(chunkSize, "chunkSize");
61
62 if (in instanceof PushbackInputStream) {
63 this.in = (PushbackInputStream) in;
64 } else {
65 this.in = new PushbackInputStream(in);
66 }
67 this.chunkSize = chunkSize;
68 }
69
70 /**
71 * Returns the number of transferred bytes.
72 */
73 public long transferredBytes() {
74 return offset;
75 }
76
77 @Override
78 public boolean isEndOfInput() throws Exception {
79 if (closed) {
80 return true;
81 }
82 if (in.available() > 0) {
83 return false;
84 }
85
86 int b = in.read();
87 if (b < 0) {
88 return true;
89 } else {
90 in.unread(b);
91 return false;
92 }
93 }
94
95 @Override
96 public void close() throws Exception {
97 closed = true;
98 in.close();
99 }
100
101 @Deprecated
102 @Override
103 public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
104 return readChunk(ctx.alloc());
105 }
106
107 @Override
108 public ByteBuf readChunk(ByteBufAllocator allocator) throws Exception {
109 if (isEndOfInput()) {
110 return null;
111 }
112
113 final int availableBytes = in.available();
114 final int chunkSize;
115 if (availableBytes <= 0) {
116 chunkSize = this.chunkSize;
117 } else {
118 chunkSize = Math.min(this.chunkSize, in.available());
119 }
120
121 boolean release = true;
122 ByteBuf buffer = allocator.buffer(chunkSize);
123 try {
124 // transfer to buffer
125 int written = buffer.writeBytes(in, chunkSize);
126 if (written < 0) {
127 return null;
128 }
129 offset += written;
130 release = false;
131 return buffer;
132 } finally {
133 if (release) {
134 buffer.release();
135 }
136 }
137 }
138
139 @Override
140 public long length() {
141 return -1;
142 }
143
144 @Override
145 public long progress() {
146 return offset;
147 }
148 }
以上源码来自： 即时通讯网 - 即时通讯开发者社区！