From cb718a07c82c6a50821ae0eb805983bb8ec8b3e5 Mon Sep 17 00:00:00 2001 From: Trustin Lee Date: Fri, 11 May 2012 00:57:42 +0900 Subject: [PATCH] Move ChannelFutureFactory.newVoidFuture() to Channel.Unsafe() / Cleanup --- .../io/netty/example/echo/EchoServer.java | 2 +- .../io/netty/channel/AbstractChannel.java | 53 +++++--------- .../main/java/io/netty/channel/Channel.java | 1 + .../netty/channel/ChannelFutureFactory.java | 1 - .../netty/channel/DefaultChannelPipeline.java | 69 +++++-------------- .../netty/channel/SingleThreadEventLoop.java | 7 +- .../channel/group/DefaultChannelGroup.java | 24 ++----- 7 files changed, 47 insertions(+), 110 deletions(-) diff --git a/example/src/main/java/io/netty/example/echo/EchoServer.java b/example/src/main/java/io/netty/example/echo/EchoServer.java index 264fb16989..b318c3f8fc 100644 --- a/example/src/main/java/io/netty/example/echo/EchoServer.java +++ b/example/src/main/java/io/netty/example/echo/EchoServer.java @@ -92,7 +92,7 @@ public class EchoServer { }); loop.register(ssc).awaitUninterruptibly().rethrowIfFailed(); - ssc.bind(new InetSocketAddress(port), ssc.newFuture()).awaitUninterruptibly().rethrowIfFailed(); + ssc.bind(new InetSocketAddress(port)).awaitUninterruptibly().rethrowIfFailed(); } public static void main(String[] args) throws Exception { diff --git a/transport/src/main/java/io/netty/channel/AbstractChannel.java b/transport/src/main/java/io/netty/channel/AbstractChannel.java index d29e619f62..a85fdf3355 100644 --- a/transport/src/main/java/io/netty/channel/AbstractChannel.java +++ b/transport/src/main/java/io/netty/channel/AbstractChannel.java @@ -73,7 +73,6 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha private final ChannelPipeline pipeline = new DefaultChannelPipeline(this); private final List closureListeners = new ArrayList(4); private final ChannelFuture succeededFuture = new SucceededChannelFuture(this); - private final ChannelFuture voidFuture = new VoidChannelFuture(this); private volatile EventLoop eventLoop; private volatile boolean registered; @@ -159,58 +158,42 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha @Override public ChannelFuture bind(SocketAddress localAddress) { - ChannelFuture f = newFuture(); - pipeline().bind(localAddress, f); - return f; + return pipeline().bind(localAddress, newFuture()); } @Override public ChannelFuture connect(SocketAddress remoteAddress) { - ChannelFuture f = newFuture(); - pipeline().connect(remoteAddress, f); - return f; + return pipeline().connect(remoteAddress, newFuture()); } @Override public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) { - ChannelFuture f = newFuture(); - pipeline().connect(remoteAddress, localAddress, f); - return f; + return pipeline().connect(remoteAddress, localAddress, newFuture()); } @Override public ChannelFuture disconnect() { - ChannelFuture f = newFuture(); - pipeline().disconnect(f); - return f; + return pipeline().disconnect(newFuture()); } @Override public ChannelFuture close() { - ChannelFuture f = newFuture(); - pipeline().close(f); - return f; + return pipeline().close(newFuture()); } @Override public ChannelFuture deregister() { - ChannelFuture f = newFuture(); - pipeline().deregister(f); - return f; + return pipeline().deregister(newFuture()); } @Override public ChannelFuture flush() { - ChannelFuture f = newFuture(); - pipeline().flush(f); - return f; + return pipeline().flush(newFuture()); } @Override public ChannelFuture write(Object message) { - ChannelFuture f = newFuture(); - pipeline().write(message, f); - return f; + return pipeline().write(message, newFuture()); } @Override @@ -273,11 +256,6 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha return new FailedChannelFuture(this, cause); } - @Override - public ChannelFuture newVoidFuture() { - return voidFuture; - } - @Override public void addClosureListener(final ChannelFutureListener listener) { if (listener == null) { @@ -445,6 +423,8 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha private class DefaultUnsafe implements Unsafe { + private final ChannelFuture voidFuture = new VoidChannelFuture(AbstractChannel.this); + @Override public java.nio.channels.Channel ch() { return javaChannel(); @@ -455,6 +435,11 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha return firstOut(); } + @Override + public ChannelFuture voidFuture() { + return voidFuture; + } + @Override public SocketAddress localAddress() { return localAddress0(); @@ -624,7 +609,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha pipeline().fireChannelInactive(); } - deregister(newVoidFuture()); + deregister(voidFuture()); } else { // Closed already. future.setSuccess(); @@ -689,12 +674,12 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha } if (closed) { - close(newVoidFuture()); + close(voidFuture()); } } catch (Throwable t) { pipeline().fireExceptionCaught(t); if (t instanceof IOException) { - close(newVoidFuture()); + close(voidFuture()); } } } @@ -737,7 +722,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha if (isOpen()) { return; } - close(newVoidFuture()); + close(voidFuture()); } } diff --git a/transport/src/main/java/io/netty/channel/Channel.java b/transport/src/main/java/io/netty/channel/Channel.java index ba9e6b1b80..8bce3295f0 100644 --- a/transport/src/main/java/io/netty/channel/Channel.java +++ b/transport/src/main/java/io/netty/channel/Channel.java @@ -174,6 +174,7 @@ public interface Channel extends AttributeMap, ChannelOutboundInvoker, ChannelFu public interface Unsafe { java.nio.channels.Channel ch(); ChannelBufferHolder out(); + ChannelFuture voidFuture(); SocketAddress localAddress(); SocketAddress remoteAddress(); diff --git a/transport/src/main/java/io/netty/channel/ChannelFutureFactory.java b/transport/src/main/java/io/netty/channel/ChannelFutureFactory.java index 857b0623f4..c07bc03d8d 100644 --- a/transport/src/main/java/io/netty/channel/ChannelFutureFactory.java +++ b/transport/src/main/java/io/netty/channel/ChannelFutureFactory.java @@ -4,5 +4,4 @@ public interface ChannelFutureFactory { ChannelFuture newFuture(); ChannelFuture newSucceededFuture(); ChannelFuture newFailedFuture(Throwable cause); - ChannelFuture newVoidFuture(); } diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java b/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java index 940a6bb22e..92cd29c78f 100644 --- a/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java +++ b/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java @@ -658,58 +658,42 @@ public class DefaultChannelPipeline implements ChannelPipeline { @Override public ChannelFuture bind(SocketAddress localAddress) { - ChannelFuture f = channel().newFuture(); - bind(localAddress, f); - return f; + return bind(localAddress, channel().newFuture()); } @Override public ChannelFuture connect(SocketAddress remoteAddress) { - ChannelFuture f = channel().newFuture(); - connect(remoteAddress, f); - return f; + return connect(remoteAddress, channel().newFuture()); } @Override public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) { - ChannelFuture f = channel().newFuture(); - connect(remoteAddress, localAddress, f); - return f; + return connect(remoteAddress, localAddress, channel().newFuture()); } @Override public ChannelFuture disconnect() { - ChannelFuture f = channel().newFuture(); - disconnect(f); - return f; + return disconnect(channel().newFuture()); } @Override public ChannelFuture close() { - ChannelFuture f = channel().newFuture(); - close(f); - return f; + return close(channel().newFuture()); } @Override public ChannelFuture deregister() { - ChannelFuture f = channel().newFuture(); - deregister(f); - return f; + return deregister(channel().newFuture()); } @Override public ChannelFuture flush() { - ChannelFuture f = channel().newFuture(); - flush(f); - return f; + return flush(channel().newFuture()); } @Override public ChannelFuture write(Object message) { - ChannelFuture f = channel().newFuture(); - write(message, f); - return f; + return write(message, channel().newFuture()); } @Override @@ -1148,58 +1132,42 @@ public class DefaultChannelPipeline implements ChannelPipeline { @Override public ChannelFuture bind(SocketAddress localAddress) { - ChannelFuture f = newFuture(); - bind(localAddress, f); - return f; + return bind(localAddress, newFuture()); } @Override public ChannelFuture connect(SocketAddress remoteAddress) { - ChannelFuture f = newFuture(); - connect(remoteAddress, f); - return f; + return connect(remoteAddress, newFuture()); } @Override public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) { - ChannelFuture f = newFuture(); - connect(remoteAddress, localAddress, f); - return f; + return connect(remoteAddress, localAddress, newFuture()); } @Override public ChannelFuture disconnect() { - ChannelFuture f = newFuture(); - disconnect(f); - return f; + return disconnect(newFuture()); } @Override public ChannelFuture close() { - ChannelFuture f = newFuture(); - close(f); - return f; + return close(newFuture()); } @Override public ChannelFuture deregister() { - ChannelFuture f = newFuture(); - deregister(f); - return f; + return deregister(newFuture()); } @Override public ChannelFuture flush() { - ChannelFuture f = newFuture(); - flush(f); - return f; + return flush(newFuture()); } @Override public ChannelFuture write(Object message) { - ChannelFuture f = newFuture(); - write(message, f); - return f; + return write(message, newFuture()); } @Override @@ -1262,10 +1230,5 @@ public class DefaultChannelPipeline implements ChannelPipeline { public ChannelFuture newFailedFuture(Throwable cause) { return channel().newFailedFuture(cause); } - - @Override - public ChannelFuture newVoidFuture() { - return channel().newVoidFuture(); - } } } diff --git a/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java b/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java index 4c3ea3a4ae..c474b58aaa 100644 --- a/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java +++ b/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java @@ -51,9 +51,10 @@ public abstract class SingleThreadEventLoop extends AbstractExecutorService impl @Override public ChannelFuture register(Channel channel) { - ChannelFuture future = channel.newFuture(); - register(channel, future); - return future; + if (channel == null) { + throw new NullPointerException("channel"); + } + return register(channel, channel.newFuture()); } @Override diff --git a/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java b/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java index d21446b34a..241b3cafb2 100644 --- a/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java +++ b/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java @@ -181,14 +181,10 @@ public class DefaultChannelGroup extends AbstractSet implements Channel new LinkedHashMap(size()); for (Channel c: serverChannels.values()) { - ChannelFuture f = c.newFuture(); - c.close(f); - futures.put(c.id(), f.awaitUninterruptibly()); + futures.put(c.id(), c.close().awaitUninterruptibly()); } for (Channel c: nonServerChannels.values()) { - ChannelFuture f = c.newFuture(); - c.close(f); - futures.put(c.id(), f); + futures.put(c.id(), c.close()); } return new DefaultChannelGroupFuture(this, futures); @@ -200,14 +196,10 @@ public class DefaultChannelGroup extends AbstractSet implements Channel new LinkedHashMap(size()); for (Channel c: serverChannels.values()) { - ChannelFuture f = c.newFuture(); - c.disconnect(f); - futures.put(c.id(), f.awaitUninterruptibly()); + futures.put(c.id(), c.disconnect()); } for (Channel c: nonServerChannels.values()) { - ChannelFuture f = c.newFuture(); - c.disconnect(f); - futures.put(c.id(), f); + futures.put(c.id(), c.disconnect()); } return new DefaultChannelGroupFuture(this, futures); @@ -220,15 +212,11 @@ public class DefaultChannelGroup extends AbstractSet implements Channel if (message instanceof ChannelBuffer) { ChannelBuffer buf = (ChannelBuffer) message; for (Channel c: nonServerChannels.values()) { - ChannelFuture f = c.newFuture(); - c.write(buf.duplicate(), f); - futures.put(c.id(), f); + futures.put(c.id(), c.write(buf.duplicate())); } } else { for (Channel c: nonServerChannels.values()) { - ChannelFuture f = c.newFuture(); - c.write(message, f); - futures.put(c.id(), f); + futures.put(c.id(), c.write(message)); } } return new DefaultChannelGroupFuture(this, futures);