io.netty.channel.ChannelOutboundBuffer#progress ( )源码实例Demo

/**
 * Write a {@link DefaultFileRegion}
 * @param in the collection which contains objects to write.
 * @param region the {@link DefaultFileRegion} 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 writeDefaultFileRegion(ChannelOutboundBuffer in, DefaultFileRegion region) throws Exception {
    final long regionCount = region.count();
    if (region.transferred() >= regionCount) {
        in.remove();
        return 0;
    }

    final long offset = region.transferred();
    final long flushedAmount = socket.sendFile(region, region.position(), offset, regionCount - offset);
    if (flushedAmount > 0) {
        in.progress(flushedAmount);
        if (region.transferred() >= regionCount) {
            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 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 DefaultFileRegion}
 * @param in the collection which contains objects to write.
 * @param region the {@link DefaultFileRegion} 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 writeDefaultFileRegion(ChannelOutboundBuffer in, DefaultFileRegion region) throws Exception {
    final long regionCount = region.count();
    if (region.transferred() >= regionCount) {
        in.remove();
        return 0;
    }

    final long offset = region.transferred();
    final long flushedAmount = socket.sendFile(region, region.position(), offset, regionCount - offset);
    if (flushedAmount > 0) {
        in.progress(flushedAmount);
        if (region.transferred() >= regionCount) {
            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 doWrite(ChannelOutboundBuffer in) throws Exception {
    for (;;) {
        Object msg = in.current();
        if (msg == null) {
            // nothing left to write
            break;
        }
        if (msg instanceof ByteBuf) {
            ByteBuf buf = (ByteBuf) msg;
            int readableBytes = buf.readableBytes();
            while (readableBytes > 0) {
                doWriteBytes(buf);
                int newReadableBytes = buf.readableBytes();
                in.progress(readableBytes - newReadableBytes);
                readableBytes = newReadableBytes;
            }
            in.remove();
        } else if (msg instanceof FileRegion) {
            FileRegion region = (FileRegion) msg;
            long transferred = region.transferred();
            doWriteFileRegion(region);
            in.progress(region.transferred() - transferred);
            in.remove();
        } else {
            in.remove(new UnsupportedOperationException(
                    "unsupported message type: " + StringUtil.simpleClassName(msg)));
        }
    }
}
 

@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
    for (;;) {
        Object msg = in.current();
        if (msg == null) {
            // nothing left to write
            break;
        }
        if (msg instanceof ByteBuf) {
            ByteBuf buf = (ByteBuf) msg;
            int readableBytes = buf.readableBytes();
            while (readableBytes > 0) {
                doWriteBytes(buf);
                int newReadableBytes = buf.readableBytes();
                in.progress(readableBytes - newReadableBytes);
                readableBytes = newReadableBytes;
            }
            in.remove();
        } else if (msg instanceof FileRegion) {
            FileRegion region = (FileRegion) msg;
            long transfered = region.transfered();
            doWriteFileRegion(region);
            in.progress(region.transfered() - transfered);
            in.remove();
        } else {
            in.remove(new UnsupportedOperationException(
                    "unsupported message type: " + StringUtil.simpleClassName(msg)));
        }
    }
}
 

/**
 * Write a {@link DefaultFileRegion}
 *
 * @param region        the {@link DefaultFileRegion} from which the bytes should be written
 * @return amount       the amount of written bytes
 */
private boolean writeFileRegion(
        ChannelOutboundBuffer in, DefaultFileRegion region, int writeSpinCount) throws Exception {
    final long regionCount = region.count();
    if (region.transfered() >= regionCount) {
        in.remove();
        return true;
    }

    final long baseOffset = region.position();
    boolean done = false;
    long flushedAmount = 0;

    for (int i = writeSpinCount - 1; i >= 0; i--) {
        final long offset = region.transfered();
        final long localFlushedAmount =
                Native.sendfile(fd().intValue(), region, baseOffset, offset, regionCount - offset);
        if (localFlushedAmount == 0) {
            break;
        }

        flushedAmount += localFlushedAmount;
        if (region.transfered() >= regionCount) {
            done = true;
            break;
        }
    }

    if (flushedAmount > 0) {
        in.progress(flushedAmount);
    }

    if (done) {
        in.remove();
    } else {
        // Returned EAGAIN need to set EPOLLOUT
        setFlag(Native.EPOLLOUT);
    }
    return done;
}