下面列出了io.netty.channel.ChannelOutboundBuffer#current ( ) 实例代码,或者点击链接到github查看源代码,也可以在右侧发表评论。
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
int writeSpinCount = config().getWriteSpinCount();
do {
Object msg = in.current();
if (msg == null) {
// Wrote all messages.
clearOpWrite();
// Directly return here so incompleteWrite(...) is not called.
return;
}
writeSpinCount -= doWriteInternal(in, msg);
} while (writeSpinCount > 0);
incompleteWrite(writeSpinCount < 0);
}
/**
* Attempt to write a single object.
* @param in the collection which contains objects to write.
* @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>
* @throws Exception If an I/O error occurs.
*/
protected int doWriteSingle(ChannelOutboundBuffer in) throws Exception {
// The outbound buffer contains only one message or it contains a file region.
Object msg = in.current();
if (msg instanceof ByteBuf) {
return writeBytes(in, (ByteBuf) msg);
} else if (msg instanceof DefaultFileRegion) {
return writeDefaultFileRegion(in, (DefaultFileRegion) msg);
} else if (msg instanceof FileRegion) {
return writeFileRegion(in, (FileRegion) msg);
} else if (msg instanceof SpliceOutTask) {
if (!((SpliceOutTask) msg).spliceOut()) {
return WRITE_STATUS_SNDBUF_FULL;
}
in.remove();
return 1;
} else {
// Should never reach here.
throw new Error();
}
}
protected boolean doWriteSingle(ChannelOutboundBuffer in, int writeSpinCount) throws Exception {
// The outbound buffer contains only one message or it contains a file region.
Object msg = in.current();
if (msg instanceof ByteBuf) {
ByteBuf buf = (ByteBuf) msg;
if (!writeBytes(in, buf, writeSpinCount)) {
// was not able to write everything so break here we will get notified later again once
// the network stack can handle more writes.
return false;
}
} else if (msg instanceof DefaultFileRegion) {
DefaultFileRegion region = (DefaultFileRegion) msg;
if (!writeFileRegion(in, region, writeSpinCount)) {
// was not able to write everything so break here we will get notified later again once
// the network stack can handle more writes.
return false;
}
} else {
// Should never reach here.
throw new Error();
}
return true;
}
public Object consume() {
ChannelOutboundBuffer buf = unsafe().outboundBuffer();
if (buf != null) {
Object msg = buf.current();
if (msg != null) {
ReferenceCountUtil.retain(msg);
buf.remove();
return msg;
}
}
return null;
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
for (;;) {
Object msg = in.current();
if (msg == null) {
break;
}
ReferenceCountUtil.retain(msg);
outboundMessages.add(msg);
in.remove();
}
}
@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 {
int writeSpinCount = config().getWriteSpinCount();
for (;;) {
final int msgCount = in.size();
if (msgCount == 0) {
// Wrote all messages.
clearFlag(Native.EPOLLOUT);
break;
}
// Do gathering write if the outbounf buffer entries start with more than one ByteBuf.
if (msgCount > 1 && in.current() instanceof ByteBuf) {
if (!doWriteMultiple(in, writeSpinCount)) {
break;
}
// We do not break the loop here even if the outbound buffer was flushed completely,
// because a user might have triggered another write and flush when we notify his or her
// listeners.
} else { // msgCount == 1
if (!doWriteSingle(in, writeSpinCount)) {
break;
}
}
}
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
int writeSpinCount = config().getWriteSpinCount();
do {
final int msgCount = in.size();
// Do gathering write if the outbound buffer entries start with more than one ByteBuf.
if (msgCount > 1 && in.current() instanceof ByteBuf) {
writeSpinCount -= doWriteMultiple(in);
} else if (msgCount == 0) {
// Wrote all messages.
writeFilter(false);
// Return here so we don't set the WRITE flag.
return;
} else { // msgCount == 1
writeSpinCount -= doWriteSingle(in);
}
// We do not break the loop here even if the outbound buffer was flushed completely,
// because a user might have triggered another write and flush when we notify his or her
// listeners.
} while (writeSpinCount > 0);
if (writeSpinCount == 0) {
// It is possible that we have set the write filter, woken up by KQUEUE because the socket is writable, and
// then use our write quantum. In this case we no longer want to set the write filter because the socket is
// still writable (as far as we know). We will find out next time we attempt to write if the socket is
// writable and set the write filter if necessary.
writeFilter(false);
// We used our writeSpin quantum, and should try to write again later.
eventLoop().execute(flushTask);
} else {
// Underlying descriptor can not accept all data currently, so set the WRITE flag to be woken up
// when it can accept more data.
writeFilter(true);
}
}
/**
* Attempt to write a single object.
* @param in the collection which contains objects to write.
* @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>
* @throws Exception If an I/O error occurs.
*/
protected int doWriteSingle(ChannelOutboundBuffer in) throws Exception {
// The outbound buffer contains only one message or it contains a file region.
Object msg = in.current();
if (msg instanceof ByteBuf) {
return writeBytes(in, (ByteBuf) msg);
} else if (msg instanceof DefaultFileRegion) {
return writeDefaultFileRegion(in, (DefaultFileRegion) msg);
} else if (msg instanceof FileRegion) {
return writeFileRegion(in, (FileRegion) msg);
} else {
// Should never reach here.
throw new Error();
}
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
for (;;) {
Object msg = in.current();
if (msg == null) {
// Wrote all messages.
writeFilter(false);
break;
}
try {
boolean done = false;
for (int i = config().getWriteSpinCount(); i > 0; --i) {
if (doWriteMessage(msg)) {
done = true;
break;
}
}
if (done) {
in.remove();
} else {
// Did not write all messages.
writeFilter(true);
break;
}
} catch (IOException e) {
// Continue on write error as a DatagramChannel can write to multiple remote peers
//
// See https://github.com/netty/netty/issues/2665
in.remove(e);
}
}
}
@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)));
}
}
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
int writeSpinCount = config().getWriteSpinCount();
do {
final int msgCount = in.size();
// Do gathering write if the outbound buffer entries start with more than one ByteBuf.
if (msgCount > 1 && in.current() instanceof ByteBuf) {
writeSpinCount -= doWriteMultiple(in);
} else if (msgCount == 0) {
// Wrote all messages.
clearFlag(Native.EPOLLOUT);
// Return here so we not set the EPOLLOUT flag.
return;
} else { // msgCount == 1
writeSpinCount -= doWriteSingle(in);
}
// We do not break the loop here even if the outbound buffer was flushed completely,
// because a user might have triggered another write and flush when we notify his or her
// listeners.
} while (writeSpinCount > 0);
if (writeSpinCount == 0) {
// It is possible that we have set EPOLLOUT, woken up by EPOLL because the socket is writable, and then use
// our write quantum. In this case we no longer want to set the EPOLLOUT flag because the socket is still
// writable (as far as we know). We will find out next time we attempt to write if the socket is writable
// and set the EPOLLOUT if necessary.
clearFlag(Native.EPOLLOUT);
// We used our writeSpin quantum, and should try to write again later.
eventLoop().execute(flushTask);
} else {
// Underlying descriptor can not accept all data currently, so set the EPOLLOUT flag to be woken up
// when it can accept more data.
setFlag(Native.EPOLLOUT);
}
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
boolean sent = false;
for (; ; ) {
Object msg = in.current();
if (msg == null) {
flushPending = false;
break;
}
try {
boolean done = false;
if (kcpSend((ByteBuf) msg)) {
done = true;
sent = true;
}
if (done) {
in.remove();
} else {
flushPending = true;
break;
}
} catch (IOException e) {
throw e; // throw exception and close channel
}
}
if (sent) {
// update kcp
if (ukcp.isFastFlush()) {
updateKcp();
} else {
kcpTsUpdate(-1);
}
}
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
boolean sent = false;
for (; ; ) {
Object msg = in.current();
if (msg == null) {
flushPending = false;
break;
}
try {
boolean done = false;
if (kcpSend((ByteBuf) msg)) {
done = true;
sent = true;
}
if (done) {
in.remove();
} else {
flushPending = true;
break;
}
} catch (IOException e) {
throw e; // throw exception and close channel
}
}
if (sent) {
// update kcp
if (ukcp.isFastFlush()) {
parent().updateChildKcp(this);
} else {
kcpTsUpdate(-1);
}
}
}
@Override
protected boolean doWriteSingle(ChannelOutboundBuffer in, int writeSpinCount) throws Exception {
Object msg = in.current();
if (msg instanceof FileDescriptor && Native.sendFd(fd().intValue(), ((FileDescriptor) msg).intValue()) > 0) {
// File descriptor was written, so remove it.
in.remove();
return true;
}
return super.doWriteSingle(in, writeSpinCount);
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
switch (state) {
case OPEN:
case BOUND:
throw new NotYetConnectedException();
case CLOSED:
throw DO_WRITE_CLOSED_CHANNEL_EXCEPTION;
case CONNECTED:
break;
}
final LocalChannel peer = this.peer;
writeInProgress = true;
try {
for (;;) {
Object msg = in.current();
if (msg == null) {
break;
}
try {
// It is possible the peer could have closed while we are writing, and in this case we should
// simulate real socket behavior and ensure the write operation is failed.
if (peer.state == State.CONNECTED) {
peer.inboundBuffer.add(ReferenceCountUtil.retain(msg));
in.remove();
} else {
in.remove(DO_WRITE_CLOSED_CHANNEL_EXCEPTION);
}
} catch (Throwable cause) {
in.remove(cause);
}
}
} finally {
// The following situation may cause trouble:
// 1. Write (with promise X)
// 2. promise X is completed when in.remove() is called, and a listener on this promise calls close()
// 3. Then the close event will be executed for the peer before the write events, when the write events
// actually happened before the close event.
writeInProgress = false;
}
finishPeerRead(peer);
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
final SelectionKey key = selectionKey();
final int interestOps = key.interestOps();
for (;;) {
Object msg = in.current();
if (msg == null) {
// Wrote all messages.
if ((interestOps & SelectionKey.OP_WRITE) != 0) {
key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
}
break;
}
try {
boolean done = false;
for (int i = config().getWriteSpinCount() - 1; i >= 0; i--) {
if (doWriteMessage(msg, in)) {
done = true;
break;
}
}
if (done) {
in.remove();
} else {
// Did not write all messages.
if ((interestOps & SelectionKey.OP_WRITE) == 0) {
key.interestOps(interestOps | SelectionKey.OP_WRITE);
}
break;
}
} catch (Exception e) {
if (continueOnWriteError()) {
in.remove(e);
} else {
throw e;
}
}
}
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
if (!writeSelector.isOpen()) {
return;
}
final int size = in.size();
final int selectedKeys = writeSelector.select(SO_TIMEOUT);
if (selectedKeys > 0) {
final Set<SelectionKey> writableKeys = writeSelector.selectedKeys();
if (writableKeys.isEmpty()) {
return;
}
Iterator<SelectionKey> writableKeysIt = writableKeys.iterator();
int written = 0;
for (;;) {
if (written == size) {
// all written
return;
}
writableKeysIt.next();
writableKeysIt.remove();
SctpMessage packet = (SctpMessage) in.current();
if (packet == null) {
return;
}
ByteBuf data = packet.content();
int dataLen = data.readableBytes();
ByteBuffer nioData;
if (data.nioBufferCount() != -1) {
nioData = data.nioBuffer();
} else {
nioData = ByteBuffer.allocate(dataLen);
data.getBytes(data.readerIndex(), nioData);
nioData.flip();
}
final MessageInfo mi = MessageInfo.createOutgoing(association(), null, packet.streamIdentifier());
mi.payloadProtocolID(packet.protocolIdentifier());
mi.streamNumber(packet.streamIdentifier());
mi.unordered(packet.isUnordered());
ch.send(nioData, mi);
written ++;
in.remove();
if (!writableKeysIt.hasNext()) {
return;
}
}
}
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
for (;;) {
Object msg = in.current();
if (msg == null) {
// Wrote all messages.
clearFlag(Native.EPOLLOUT);
break;
}
try {
// Check if sendmmsg(...) is supported which is only the case for GLIBC 2.14+
if (Native.IS_SUPPORTING_SENDMMSG && in.size() > 1) {
NativeDatagramPacketArray array = NativeDatagramPacketArray.getInstance(in);
int cnt = array.count();
if (cnt >= 1) {
// Try to use gathering writes via sendmmsg(...) syscall.
int offset = 0;
NativeDatagramPacketArray.NativeDatagramPacket[] packets = array.packets();
while (cnt > 0) {
int send = Native.sendmmsg(socket.intValue(), packets, offset, cnt);
if (send == 0) {
// Did not write all messages.
setFlag(Native.EPOLLOUT);
return;
}
for (int i = 0; i < send; i++) {
in.remove();
}
cnt -= send;
offset += send;
}
continue;
}
}
boolean done = false;
for (int i = config().getWriteSpinCount(); i > 0; --i) {
if (doWriteMessage(msg)) {
done = true;
break;
}
}
if (done) {
in.remove();
} else {
// Did not write all messages.
setFlag(Native.EPOLLOUT);
break;
}
} catch (IOException e) {
// Continue on write error as a DatagramChannel can write to multiple remote peers
//
// See https://github.com/netty/netty/issues/2665
in.remove(e);
}
}
}
@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
if (!writeSelector.isOpen()) {
return;
}
final int size = in.size();
final int selectedKeys = writeSelector.select(SO_TIMEOUT);
if (selectedKeys > 0) {
final Set<SelectionKey> writableKeys = writeSelector.selectedKeys();
if (writableKeys.isEmpty()) {
return;
}
Iterator<SelectionKey> writableKeysIt = writableKeys.iterator();
int written = 0;
for (;;) {
if (written == size) {
// all written
return;
}
writableKeysIt.next();
writableKeysIt.remove();
SctpMessage packet = (SctpMessage) in.current();
if (packet == null) {
return;
}
ByteBuf data = packet.content();
int dataLen = data.readableBytes();
ByteBuffer nioData;
if (data.nioBufferCount() != -1) {
nioData = data.nioBuffer();
} else {
nioData = ByteBuffer.allocate(dataLen);
data.getBytes(data.readerIndex(), nioData);
nioData.flip();
}
final MessageInfo mi = MessageInfo.createOutgoing(association(), null, packet.streamIdentifier());
mi.payloadProtocolID(packet.protocolIdentifier());
mi.streamNumber(packet.streamIdentifier());
ch.send(nioData, mi);
written ++;
in.remove();
if (!writableKeysIt.hasNext()) {
return;
}
}
}
}