andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Revert as it should be in nio2 branch "Commit first round of classes to support nio2/async channel api. Still work in progress.. See #396"

Browse Source 
This reverts commit 18aaae3c2e.
This commit is contained in:
Norman Maurer 2012-06-13 22:24:32 +02:00 committed by Trustin Lee
parent 41a9c66f66
commit c165a38e15
6 changed files with 22 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
pom.xml
View File

@ -276,12 +276,14 @@
<ignores> <ignores>
<ignore>sun.misc.Unsafe</ignore> <ignore>sun.misc.Unsafe</ignore>
<ignore>java.util.zip.Deflater</ignore> <ignore>java.util.zip.Deflater</ignore>
<!-- Used for NIO UDP multicast --> <!-- Used for NIO UDP multicast -->
<ignore>java.nio.channels.DatagramChannel</ignore> <ignore>java.nio.channels.DatagramChannel</ignore>
<ignore>java.nio.channels.MembershipKey</ignore> <ignore>java.nio.channels.MembershipKey</ignore>
<ignore>java.net.StandardSocketOptions</ignore> <ignore>java.net.StandardSocketOptions</ignore>
<ignore>java.net.StandardProtocolFamily</ignore> <ignore>java.net.StandardProtocolFamily</ignore>
<!-- Used for NIO2 -->
<!-- Used for NIO. 2 -->
<ignore>java.nio.channels.AsynchronousChannel</ignore> <ignore>java.nio.channels.AsynchronousChannel</ignore>
<ignore>java.nio.channels.AsynchronousSocketChannel</ignore> <ignore>java.nio.channels.AsynchronousSocketChannel</ignore>
<ignore>java.nio.channels.AsynchronousServerSocketChannel</ignore> <ignore>java.nio.channels.AsynchronousServerSocketChannel</ignore>

30
transport/src/main/java/io/netty/channel/AbstractChannel.java
View File

@ -85,7 +85,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
private ClosedChannelException closedChannelException; private ClosedChannelException closedChannelException;
private final Deque<FlushCheckpoint> flushCheckpoints = new ArrayDeque<FlushCheckpoint>(); private final Deque<FlushCheckpoint> flushCheckpoints = new ArrayDeque<FlushCheckpoint>();
private long writeCounter; private long writeCounter;
protected boolean inFlushNow; private boolean inFlushNow;
private boolean flushNowPending; private boolean flushNowPending;
/** Cache for the string representation of this channel */ /** Cache for the string representation of this channel */
@ -623,7 +623,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
} }
@Override @Override
public void flushNow() { public final void flushNow() {
if (inFlushNow) { if (inFlushNow) {
return; return;
} }
@ -631,13 +631,12 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
inFlushNow = true; inFlushNow = true;
ChannelHandlerContext ctx = directOutboundContext(); ChannelHandlerContext ctx = directOutboundContext();
Throwable cause = null; Throwable cause = null;
boolean handleFlush = true;
try { try {
if (ctx.hasOutboundByteBuffer()) { if (ctx.hasOutboundByteBuffer()) {
ByteBuf out = ctx.outboundByteBuffer(); ByteBuf out = ctx.outboundByteBuffer();
int oldSize = out.readableBytes(); int oldSize = out.readableBytes();
try { try {
handleFlush = doFlushByteBuffer(out); doFlushByteBuffer(out);
} catch (Throwable t) { } catch (Throwable t) {
cause = t; cause = t;
} finally { } finally {
@ -658,15 +657,14 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
writeCounter += oldSize - out.size(); writeCounter += oldSize - out.size();
} }
} }
if (handleFlush) {
if (cause == null) { if (cause == null) {
notifyFlushFutures(); notifyFlushFutures();
} else { } else {
notifyFlushFutures(cause); notifyFlushFutures(cause);
pipeline.fireExceptionCaught(cause); pipeline.fireExceptionCaught(cause);
if (cause instanceof IOException) { if (cause instanceof IOException) {
close(voidFuture()); close(voidFuture());
}
} }
} }
} finally { } finally {
@ -715,7 +713,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
protected abstract void doClose() throws Exception; protected abstract void doClose() throws Exception;
protected abstract void doDeregister() throws Exception; protected abstract void doDeregister() throws Exception;
protected boolean doFlushByteBuffer(ByteBuf buf) throws Exception { protected void doFlushByteBuffer(ByteBuf buf) throws Exception {
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
} }
protected void doFlushMessageBuffer(MessageBuf<Object> buf) throws Exception { protected void doFlushMessageBuffer(MessageBuf<Object> buf) throws Exception {
@ -724,7 +722,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
protected abstract boolean isFlushPending(); protected abstract boolean isFlushPending();
protected final void notifyFlushFutures() { private void notifyFlushFutures() {
if (flushCheckpoints.isEmpty()) { if (flushCheckpoints.isEmpty()) {
return; return;
} }
@ -762,7 +760,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
} }
} }
protected final void notifyFlushFutures(Throwable cause) { private void notifyFlushFutures(Throwable cause) {
notifyFlushFutures(); notifyFlushFutures();
for (;;) { for (;;) {
FlushCheckpoint cp = flushCheckpoints.poll(); FlushCheckpoint cp = flushCheckpoints.poll();

2
transport/src/main/java/io/netty/channel/AbstractServerChannel.java
View File

@ -77,7 +77,7 @@ public abstract class AbstractServerChannel extends AbstractChannel implements S
} }
@Override @Override
protected boolean doFlushByteBuffer(ByteBuf buf) throws Exception { protected void doFlushByteBuffer(ByteBuf buf) throws Exception {
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
} }

3
transport/src/main/java/io/netty/channel/embedded/EmbeddedByteChannel.java
View File

@ -71,11 +71,10 @@ public class EmbeddedByteChannel extends AbstractEmbeddedChannel {
} }
@Override @Override
protected boolean doFlushByteBuffer(ByteBuf buf) throws Exception { protected void doFlushByteBuffer(ByteBuf buf) throws Exception {
if (!lastOutboundBuffer().readable()) { if (!lastOutboundBuffer().readable()) {
lastOutboundBuffer().discardReadBytes(); lastOutboundBuffer().discardReadBytes();
} }
lastOutboundBuffer().writeBytes(buf); lastOutboundBuffer().writeBytes(buf);
return true;
} }
} }

5
transport/src/main/java/io/netty/channel/socket/nio/AbstractNioByteChannel.java
View File

@ -85,11 +85,11 @@ abstract class AbstractNioByteChannel extends AbstractNioChannel {
} }
@Override @Override
protected boolean doFlushByteBuffer(ByteBuf buf) throws Exception { protected void doFlushByteBuffer(ByteBuf buf) throws Exception {
if (!buf.readable()) { if (!buf.readable()) {
// Reset reader/writerIndex to 0 if the buffer is empty. // Reset reader/writerIndex to 0 if the buffer is empty.
buf.clear(); buf.clear();
return true; return;
} }
for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) { for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) {
@ -103,7 +103,6 @@ abstract class AbstractNioByteChannel extends AbstractNioChannel {
break; break;
} }
} }
return true;
} }
protected abstract int doReadBytes(ByteBuf buf) throws Exception; protected abstract int doReadBytes(ByteBuf buf) throws Exception;

3
transport/src/main/java/io/netty/channel/socket/oio/AbstractOioByteChannel.java
View File

@ -86,12 +86,11 @@ abstract class AbstractOioByteChannel extends AbstractOioChannel {
} }
@Override @Override
protected boolean doFlushByteBuffer(ByteBuf buf) throws Exception { protected void doFlushByteBuffer(ByteBuf buf) throws Exception {
while (buf.readable()) { while (buf.readable()) {
doWriteBytes(buf); doWriteBytes(buf);
} }
buf.clear(); buf.clear();
return true;
} }
protected abstract int available(); protected abstract int available();