下面列出了io.netty.channel.FileRegion#count ( ) 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
@Override
protected void doWriteFileRegion(FileRegion region) throws Exception {
OutputStream os = this.os;
if (os == null) {
throw new NotYetConnectedException();
}
if (outChannel == null) {
outChannel = Channels.newChannel(os);
}
long written = 0;
for (;;) {
long localWritten = region.transferTo(outChannel, written);
if (localWritten == -1) {
checkEOF(region);
return;
}
written += localWritten;
if (written >= region.count()) {
return;
}
}
}
/**
* Write a {@link FileRegion}
* @param in the collection which contains objects to write.
* @param region the {@link FileRegion} from which the bytes should be written
* @return The value that should be decremented from the write quantum which starts at
* {@link ChannelConfig#getWriteSpinCount()}. The typical use cases are as follows:
* <ul>
* <li>0 - if no write was attempted. This is appropriate if an empty {@link ByteBuf} (or other empty content)
* is encountered</li>
* <li>1 - if a single call to write data was made to the OS</li>
* <li>{@link ChannelUtils#WRITE_STATUS_SNDBUF_FULL} - if an attempt to write data was made to the OS, but no
* data was accepted</li>
* </ul>
*/
private int writeFileRegion(ChannelOutboundBuffer in, FileRegion region) throws Exception {
if (region.transferred() >= region.count()) {
in.remove();
return 0;
}
if (byteChannel == null) {
byteChannel = new KQueueSocketWritableByteChannel();
}
final long flushedAmount = region.transferTo(byteChannel, region.transferred());
if (flushedAmount > 0) {
in.progress(flushedAmount);
if (region.transferred() >= region.count()) {
in.remove();
}
return 1;
}
return WRITE_STATUS_SNDBUF_FULL;
}
/**
* Write a {@link FileRegion}
* @param in the collection which contains objects to write.
* @param region the {@link FileRegion} from which the bytes should be written
* @return The value that should be decremented from the write quantum which starts at
* {@link ChannelConfig#getWriteSpinCount()}. The typical use cases are as follows:
* <ul>
* <li>0 - if no write was attempted. This is appropriate if an empty {@link ByteBuf} (or other empty content)
* is encountered</li>
* <li>1 - if a single call to write data was made to the OS</li>
* <li>{@link ChannelUtils#WRITE_STATUS_SNDBUF_FULL} - if an attempt to write data was made to the OS, but
* no data was accepted</li>
* </ul>
*/
private int writeFileRegion(ChannelOutboundBuffer in, FileRegion region) throws Exception {
if (region.transferred() >= region.count()) {
in.remove();
return 0;
}
if (byteChannel == null) {
byteChannel = new EpollSocketWritableByteChannel();
}
final long flushedAmount = region.transferTo(byteChannel, region.transferred());
if (flushedAmount > 0) {
in.progress(flushedAmount);
if (region.transferred() >= region.count()) {
in.remove();
}
return 1;
}
return WRITE_STATUS_SNDBUF_FULL;
}
@Override
protected void doWriteFileRegion(FileRegion region) throws Exception {
OutputStream os = this.os;
if (os == null) {
throw new NotYetConnectedException();
}
if (outChannel == null) {
outChannel = Channels.newChannel(os);
}
long written = 0;
for (;;) {
long localWritten = region.transferTo(outChannel, written);
if (localWritten == -1) {
checkEOF(region);
return;
}
written += localWritten;
if (written >= region.count()) {
return;
}
}
}
/**
* Encode a message into a {@link io.netty.buffer.ByteBuf}. This method will be called for each written message that
* can be handled by this encoder.
*
* @param ctx the {@link io.netty.channel.ChannelHandlerContext} which this {@link
* io.netty.handler.codec.MessageToByteEncoder} belongs to
* @param msg the message to encode
* @param out the {@link io.netty.buffer.ByteBuf} into which the encoded message will be written
* @throws Exception is thrown if an error occurs
*/
@Override
protected void encode(ChannelHandlerContext ctx, FileRegion msg, final ByteBuf out) throws Exception {
WritableByteChannel writableByteChannel = new WritableByteChannel() {
@Override
public int write(ByteBuffer src) throws IOException {
out.writeBytes(src);
return out.capacity();
}
@Override
public boolean isOpen() {
return true;
}
@Override
public void close() throws IOException {
}
};
long toTransfer = msg.count();
while (true) {
long transferred = msg.transfered();
if (toTransfer - transferred <= 0) {
break;
}
msg.transferTo(writableByteChannel, transferred);
}
}
/**
* Encode a message into a {@link io.netty.buffer.ByteBuf}. This method will be called for each written message that
* can be handled by this encoder.将消息编码到ByteBuf中。对于可由此编码器处理的每个书面消息，将调用此方法。
*
* @param ctx the {@link io.netty.channel.ChannelHandlerContext} which this {@link
* io.netty.handler.codec.MessageToByteEncoder} belongs to
* @param msg the message to encode
* @param out the {@link io.netty.buffer.ByteBuf} into which the encoded message will be written
* @throws Exception is thrown if an error occurs
*/
@Override
protected void encode(ChannelHandlerContext ctx, FileRegion msg, final ByteBuf out) throws Exception {
WritableByteChannel writableByteChannel = new WritableByteChannel() {
@Override
public int write(ByteBuffer src) throws IOException {
out.writeBytes(src);
return out.capacity();
}
@Override
public boolean isOpen() {
return true;
}
@Override
public void close() throws IOException {
}
};
long toTransfer = msg.count();
while (true) {
long transferred = msg.transfered();
if (toTransfer - transferred <= 0) {
break;
}
msg.transferTo(writableByteChannel, transferred);
}
}
/**
* Encode a message into a {@link io.netty.buffer.ByteBuf}. This method will be called for each written message that
* can be handled by this encoder.
*
* @param ctx the {@link io.netty.channel.ChannelHandlerContext} which this {@link
* io.netty.handler.codec.MessageToByteEncoder} belongs to
* @param msg the message to encode
* @param out the {@link io.netty.buffer.ByteBuf} into which the encoded message will be written
* @throws Exception is thrown if an error occurs
*/
@Override
protected void encode(ChannelHandlerContext ctx, FileRegion msg, final ByteBuf out) throws Exception {
WritableByteChannel writableByteChannel = new WritableByteChannel() {
@Override
public int write(ByteBuffer src) throws IOException {
out.writeBytes(src);
return out.capacity();
}
@Override
public boolean isOpen() {
return true;
}
@Override
public void close() throws IOException {
}
};
long toTransfer = msg.count();
while (true) {
long transferred = msg.transfered();
if (toTransfer - transferred <= 0) {
break;
}
msg.transferTo(writableByteChannel, transferred);
}
}
/**
* Encode a message into a {@link io.netty.buffer.ByteBuf}. This method will be called for each written message that
* can be handled by this encoder.
*
* @param ctx the {@link io.netty.channel.ChannelHandlerContext} which this {@link
* io.netty.handler.codec.MessageToByteEncoder} belongs to
* @param msg the message to encode
* @param out the {@link io.netty.buffer.ByteBuf} into which the encoded message will be written
* @throws Exception is thrown if an error occurs
*/
@Override
protected void encode(ChannelHandlerContext ctx, FileRegion msg, final ByteBuf out) throws Exception {
WritableByteChannel writableByteChannel = new WritableByteChannel() {
@Override
public int write(ByteBuffer src) throws IOException {
out.writeBytes(src);
return out.capacity();
}
@Override
public boolean isOpen() {
return true;
}
@Override
public void close() throws IOException {
}
};
long toTransfer = msg.count();
while (true) {
long transferred = msg.transfered();
if (toTransfer - transferred <= 0) {
break;
}
msg.transferTo(writableByteChannel, transferred);
}
}
/**
* Encode a message into a {@link io.netty.buffer.ByteBuf}. This method will be called for each written message that
* can be handled by this encoder.
*
* @param ctx the {@link io.netty.channel.ChannelHandlerContext} which this {@link
* io.netty.handler.codec.MessageToByteEncoder} belongs to
* @param msg the message to encode
* @param out the {@link io.netty.buffer.ByteBuf} into which the encoded message will be written
* @throws Exception is thrown if an error occurs
*/
@Override
protected void encode(ChannelHandlerContext ctx, FileRegion msg, final ByteBuf out) throws Exception {
WritableByteChannel writableByteChannel = new WritableByteChannel() {
@Override
public int write(ByteBuffer src) throws IOException {
out.writeBytes(src);
return out.capacity();
}
@Override
public boolean isOpen() {
return true;
}
@Override
public void close() throws IOException {
}
};
long toTransfer = msg.count();
while (true) {
long transferred = msg.transfered();
if (toTransfer - transferred <= 0) {
break;
}
msg.transferTo(writableByteChannel, transferred);
}
}
private static void checkEOF(FileRegion region) throws IOException {
if (region.transferred() < region.count()) {
throw new EOFException("Expected to be able to write " + region.count() + " bytes, " +
"but only wrote " + region.transferred());
}
}
@Override
public void handleMessage(Object msg) {
try {
//log.info("Got message: " + msg.getClass().getName());
if (msg instanceof HttpResponse) {
HttpResponse res = (HttpResponse) msg;
responseBuilder.setStatusCode(res.status().code());
if (request.getRequestSource() == AwsProxyRequest.RequestSource.ALB) {
responseBuilder.setStatusDescription(res.status().reasonPhrase());
}
responseBuilder.setMultiValueHeaders(new Headers());
for (String name : res.headers().names()) {
for (String v : res.headers().getAll(name)) {
responseBuilder.getMultiValueHeaders().add(name, v);
}
}
}
if (msg instanceof HttpContent) {
HttpContent content = (HttpContent) msg;
int readable = content.content().readableBytes();
if (baos == null && readable > 0) {
baos = createByteStream();
}
for (int i = 0; i < readable; i++) {
baos.write(content.content().readByte());
}
}
if (msg instanceof FileRegion) {
FileRegion file = (FileRegion) msg;
if (file.count() > 0 && file.transferred() < file.count()) {
if (baos == null)
baos = createByteStream();
if (byteChannel == null)
byteChannel = Channels.newChannel(baos);
file.transferTo(byteChannel, file.transferred());
}
}
if (msg instanceof LastHttpContent) {
if (baos != null) {
if (isBinary(responseBuilder.getMultiValueHeaders().getFirst("Content-Type"))) {
responseBuilder.setBase64Encoded(true);
responseBuilder.setBody(Base64.getMimeEncoder().encodeToString(baos.toByteArray()));
} else {
responseBuilder.setBody(new String(baos.toByteArray(), "UTF-8"));
}
}
future.complete(responseBuilder);
}
} catch (Throwable ex) {
future.completeExceptionally(ex);
} finally {
if (msg != null) {
ReferenceCountUtil.release(msg);
}
}
}
@Override
public void handleMessage(Object msg) {
try {
//log.info("Got message: " + msg.getClass().getName());
if (msg instanceof HttpResponse) {
HttpResponse res = (HttpResponse) msg;
responseBuilder = request.createResponseBuilder(HttpStatus.valueOf(res.status().code()));
for (Map.Entry<String, String> entry : res.headers()) {
responseBuilder.header(entry.getKey(), entry.getValue());
}
}
if (msg instanceof HttpContent) {
HttpContent content = (HttpContent) msg;
if (baos == null) {
// todo what is right size?
baos = createByteStream();
}
int readable = content.content().readableBytes();
for (int i = 0; i < readable; i++) {
baos.write(content.content().readByte());
}
}
if (msg instanceof FileRegion) {
FileRegion file = (FileRegion) msg;
if (file.count() > 0 && file.transferred() < file.count()) {
if (baos == null)
baos = createByteStream();
if (byteChannel == null)
byteChannel = Channels.newChannel(baos);
file.transferTo(byteChannel, file.transferred());
}
}
if (msg instanceof LastHttpContent) {
responseBuilder.body(baos.toByteArray());
future.complete(responseBuilder.build());
}
} catch (Throwable ex) {
future.completeExceptionally(ex);
} finally {
ReferenceCountUtil.release(msg);
}
}
private static void checkEOF(FileRegion region) throws IOException {
if (region.transfered() < region.count()) {
throw new EOFException("Expected to be able to write " + region.count() + " bytes, " +
"but only wrote " + region.transfered());
}
}