From c149f4bcc0c0d02aa1abcd5e39c155a9e598822e Mon Sep 17 00:00:00 2001 From: bgallagher Date: Fri, 23 Aug 2013 14:52:52 -0400 Subject: [PATCH] Remove support from deregister a Channel from a EventLoop manually --- .../http/upload/HttpUploadServerHandler.java | 2 +- .../example/uptime/UptimeClientHandler.java | 6 +- .../netty/handler/logging/LoggingHandler.java | 18 --- .../java/io/netty/handler/ssl/SslHandler.java | 7 +- .../io/netty/channel/AbstractChannel.java | 45 ++---- .../main/java/io/netty/channel/Channel.java | 7 - .../netty/channel/ChannelDuplexHandler.java | 11 -- .../netty/channel/ChannelHandlerContext.java | 4 - .../netty/channel/ChannelInboundHandler.java | 12 +- .../channel/ChannelInboundHandlerAdapter.java | 13 +- .../netty/channel/ChannelInboundInvoker.java | 10 -- .../netty/channel/ChannelOutboundHandler.java | 9 -- .../ChannelOutboundHandlerAdapter.java | 11 -- .../netty/channel/ChannelOutboundInvoker.java | 31 ---- .../io/netty/channel/ChannelPipeline.java | 6 - .../channel/CombinedChannelDuplexHandler.java | 10 -- .../channel/DefaultChannelHandlerContext.java | 58 ------- .../netty/channel/DefaultChannelPipeline.java | 30 +--- .../io/netty/channel/group/ChannelGroup.java | 21 --- .../channel/group/DefaultChannelGroup.java | 27 ---- .../io/netty/channel/local/LocalChannel.java | 5 +- .../netty/channel/AbstractEventLoopTest.java | 12 +- .../io/netty/channel/BaseChannelTest.java | 14 ++ .../java/io/netty/channel/LoggingHandler.java | 17 +-- .../netty/channel/ReentrantChannelTest.java | 143 ++++++++++++++++++ .../local/LocalTransportThreadModelTest3.java | 22 +-- 26 files changed, 188 insertions(+), 363 deletions(-) diff --git a/example/src/main/java/io/netty/example/http/upload/HttpUploadServerHandler.java b/example/src/main/java/io/netty/example/http/upload/HttpUploadServerHandler.java index 878c5271f520..8856b9762e12 100644 --- a/example/src/main/java/io/netty/example/http/upload/HttpUploadServerHandler.java +++ b/example/src/main/java/io/netty/example/http/upload/HttpUploadServerHandler.java @@ -88,7 +88,7 @@ public class HttpUploadServerHandler extends SimpleChannelInboundHandlerSSL * · TLS and StartTLS support to a {@link Channel}. Please refer @@ -370,11 +372,6 @@ public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, Sock ctx.connect(remoteAddress, localAddress, promise); } - @Override - public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { - ctx.deregister(promise); - } - @Override public void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) 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 5876fc3b1a47..6871ad334945 100644 --- a/transport/src/main/java/io/netty/channel/AbstractChannel.java +++ b/transport/src/main/java/io/netty/channel/AbstractChannel.java @@ -30,6 +30,7 @@ import java.net.SocketAddress; import java.nio.channels.ClosedChannelException; import java.nio.channels.NotYetConnectedException; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; /** * A skeletal {@link Channel} implementation. @@ -66,6 +67,9 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha private boolean strValActive; private String strVal; + private static final AtomicReferenceFieldUpdater EVENT_LOOP_UPDATER = + AtomicReferenceFieldUpdater.newUpdater(AbstractChannel.class, EventLoop.class, "eventLoop"); + /** * Creates a new instance. * @@ -177,11 +181,6 @@ public ChannelFuture close() { return pipeline.close(); } - @Override - public ChannelFuture deregister() { - return pipeline.deregister(); - } - @Override public Channel flush() { pipeline.flush(); @@ -213,11 +212,6 @@ public ChannelFuture close(ChannelPromise promise) { return pipeline.close(promise); } - @Override - public ChannelFuture deregister(ChannelPromise promise) { - return pipeline.deregister(promise); - } - @Override public Channel read() { pipeline.read(); @@ -395,17 +389,16 @@ public final void register(EventLoop eventLoop, final ChannelPromise promise) { if (eventLoop == null) { throw new NullPointerException("eventLoop"); } - if (isRegistered()) { - promise.setFailure(new IllegalStateException("registered to an event loop already")); - return; - } + if (!isCompatible(eventLoop)) { - promise.setFailure( - new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName())); + promise.setFailure(new IllegalStateException("incompatible event loop type: " + + eventLoop.getClass().getName())); return; } - AbstractChannel.this.eventLoop = eventLoop; + if (!AbstractChannel.EVENT_LOOP_UPDATER.compareAndSet(AbstractChannel.this, null, eventLoop)) { + return; + } if (eventLoop.inEventLoop()) { register0(promise); @@ -560,7 +553,7 @@ public void run() { }); } - deregister(voidPromise()); + deregister(); } } @@ -573,10 +566,8 @@ public final void closeForcibly() { } } - @Override - public final void deregister(final ChannelPromise promise) { + private void deregister() { if (!registered) { - promise.setSuccess(); return; } @@ -587,18 +578,6 @@ public final void deregister(final ChannelPromise promise) { } finally { if (registered) { registered = false; - promise.setSuccess(); - invokeLater(new Runnable() { - @Override - public void run() { - pipeline.fireChannelUnregistered(); - } - }); - } else { - // Some transports like local and AIO does not allow the deregistration of - // an open channel. Their doDeregister() calls close(). Consequently, - // close() calls deregister() again - no need to fire channelUnregistered. - promise.setSuccess(); } } } diff --git a/transport/src/main/java/io/netty/channel/Channel.java b/transport/src/main/java/io/netty/channel/Channel.java index 39b8417762a4..2a9d34f3cc15 100644 --- a/transport/src/main/java/io/netty/channel/Channel.java +++ b/transport/src/main/java/io/netty/channel/Channel.java @@ -222,13 +222,6 @@ interface Unsafe { */ void closeForcibly(); - /** - * Deregister the {@link Channel} of the {@link ChannelPromise} from {@link EventLoop} and notify the - * {@link ChannelPromise} once the operation was complete. - */ - @Deprecated - void deregister(ChannelPromise promise); - /** * Schedules a read operation that fills the inbound buffer of the first {@link ChannelInboundHandler} in the * {@link ChannelPipeline}. If there's already a pending read operation, this method does nothing. diff --git a/transport/src/main/java/io/netty/channel/ChannelDuplexHandler.java b/transport/src/main/java/io/netty/channel/ChannelDuplexHandler.java index 6b6d1841db2a..afea6180c23f 100644 --- a/transport/src/main/java/io/netty/channel/ChannelDuplexHandler.java +++ b/transport/src/main/java/io/netty/channel/ChannelDuplexHandler.java @@ -73,17 +73,6 @@ public void close(ChannelHandlerContext ctx, ChannelPromise future) throws Excep ctx.close(future); } - /** - * Calls {@link ChannelHandlerContext#close(ChannelPromise)} to forward - * to the next {@link ChannelOutboundHandler} in the {@link ChannelPipeline}. - * - * Sub-classes may override this method to change behavior. - */ - @Override - public void deregister(ChannelHandlerContext ctx, ChannelPromise future) throws Exception { - ctx.deregister(future); - } - /** * Calls {@link ChannelHandlerContext#read()} to forward * to the next {@link ChannelOutboundHandler} in the {@link ChannelPipeline}. diff --git a/transport/src/main/java/io/netty/channel/ChannelHandlerContext.java b/transport/src/main/java/io/netty/channel/ChannelHandlerContext.java index 4c24ff1cc382..2d7ca803b1cb 100644 --- a/transport/src/main/java/io/netty/channel/ChannelHandlerContext.java +++ b/transport/src/main/java/io/netty/channel/ChannelHandlerContext.java @@ -160,10 +160,6 @@ public interface ChannelHandlerContext @Override ChannelHandlerContext fireChannelRegistered(); - @Override - @Deprecated - ChannelHandlerContext fireChannelUnregistered(); - @Override ChannelHandlerContext fireChannelActive(); diff --git a/transport/src/main/java/io/netty/channel/ChannelInboundHandler.java b/transport/src/main/java/io/netty/channel/ChannelInboundHandler.java index 4cccc84405ae..a4e1b97cfac9 100644 --- a/transport/src/main/java/io/netty/channel/ChannelInboundHandler.java +++ b/transport/src/main/java/io/netty/channel/ChannelInboundHandler.java @@ -16,8 +16,8 @@ package io.netty.channel; /** - * {@link ChannelHandler} which adds callbacks for state changes. This allows the user - * to hook in to state changes easily. + * {@link ChannelHandler} which adds callbacks for state changes. This allows the user to hook in to state changes + * easily. */ public interface ChannelInboundHandler extends ChannelHandler { @@ -26,14 +26,6 @@ public interface ChannelInboundHandler extends ChannelHandler { */ void channelRegistered(ChannelHandlerContext ctx) throws Exception; - /** - * The {@link Channel} of the {@link ChannelHandlerContext} was unregistered from its {@link EventLoop} - * - * @deprecated use {@link #channelInactive(ChannelHandlerContext)} - */ - @Deprecated - void channelUnregistered(ChannelHandlerContext ctx) throws Exception; - /** * The {@link Channel} of the {@link ChannelHandlerContext} is now active */ diff --git a/transport/src/main/java/io/netty/channel/ChannelInboundHandlerAdapter.java b/transport/src/main/java/io/netty/channel/ChannelInboundHandlerAdapter.java index 3a06d12170c4..580cc82f3ed6 100644 --- a/transport/src/main/java/io/netty/channel/ChannelInboundHandlerAdapter.java +++ b/transport/src/main/java/io/netty/channel/ChannelInboundHandlerAdapter.java @@ -16,7 +16,7 @@ package io.netty.channel; /** - * Abstract base class for {@link ChannelInboundHandler} implementations which provide + * Abstract base class for {@link ChannelInboundHandler} implementations that provides * implementations of all of their methods. * *

@@ -42,17 +42,6 @@ public void channelRegistered(ChannelHandlerContext ctx) throws Exception { ctx.fireChannelRegistered(); } - /** - * Calls {@link ChannelHandlerContext#fireChannelUnregistered()} to forward - * to the next {@link ChannelInboundHandler} in the {@link ChannelPipeline}. - * - * Sub-classes may override this method to change behavior. - */ - @Override - public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { - ctx.fireChannelUnregistered(); - } - /** * Calls {@link ChannelHandlerContext#fireChannelActive()} to forward * to the next {@link ChannelInboundHandler} in the {@link ChannelPipeline}. diff --git a/transport/src/main/java/io/netty/channel/ChannelInboundInvoker.java b/transport/src/main/java/io/netty/channel/ChannelInboundInvoker.java index 64086f348e2d..07717621402d 100644 --- a/transport/src/main/java/io/netty/channel/ChannelInboundInvoker.java +++ b/transport/src/main/java/io/netty/channel/ChannelInboundInvoker.java @@ -30,16 +30,6 @@ interface ChannelInboundInvoker { */ ChannelInboundInvoker fireChannelRegistered(); - /** - * A {@link Channel} was unregistered from its {@link EventLoop}. - * - * This will result in having the {@link ChannelInboundHandler#channelUnregistered(ChannelHandlerContext)} method - * called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the - * {@link Channel}. - */ - @Deprecated - ChannelInboundInvoker fireChannelUnregistered(); - /** * A {@link Channel} is active now, which means it is connected. * diff --git a/transport/src/main/java/io/netty/channel/ChannelOutboundHandler.java b/transport/src/main/java/io/netty/channel/ChannelOutboundHandler.java index 92a3943ad042..a02f780bca14 100644 --- a/transport/src/main/java/io/netty/channel/ChannelOutboundHandler.java +++ b/transport/src/main/java/io/netty/channel/ChannelOutboundHandler.java @@ -62,15 +62,6 @@ void connect( */ void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception; - /** - * Called once a deregister operation is made from the current registered {@link EventLoop}. - * - * @param ctx the {@link ChannelHandlerContext} for which the close operation is made - * @param promise the {@link ChannelPromise} to notify once the operation completes - * @throws Exception thrown if an error accour - */ - void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception; - /** * Intercepts {@link ChannelHandlerContext#read()}. */ diff --git a/transport/src/main/java/io/netty/channel/ChannelOutboundHandlerAdapter.java b/transport/src/main/java/io/netty/channel/ChannelOutboundHandlerAdapter.java index c0a68971237e..2ac728ac776f 100644 --- a/transport/src/main/java/io/netty/channel/ChannelOutboundHandlerAdapter.java +++ b/transport/src/main/java/io/netty/channel/ChannelOutboundHandlerAdapter.java @@ -71,17 +71,6 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) ctx.close(promise); } - /** - * Calls {@link ChannelHandlerContext#close(ChannelPromise)} to forward - * to the next {@link ChannelOutboundHandler} in the {@link ChannelPipeline}. - * - * Sub-classes may override this method to change behavior. - */ - @Override - public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { - ctx.deregister(promise); - } - /** * Calls {@link ChannelHandlerContext#read()} to forward * to the next {@link ChannelOutboundHandler} in the {@link ChannelPipeline}. diff --git a/transport/src/main/java/io/netty/channel/ChannelOutboundInvoker.java b/transport/src/main/java/io/netty/channel/ChannelOutboundInvoker.java index 64b5ec64cc17..f4de88cde688 100644 --- a/transport/src/main/java/io/netty/channel/ChannelOutboundInvoker.java +++ b/transport/src/main/java/io/netty/channel/ChannelOutboundInvoker.java @@ -15,8 +15,6 @@ */ package io.netty.channel; -import io.netty.util.concurrent.EventExecutor; - import java.net.ConnectException; import java.net.SocketAddress; @@ -88,20 +86,6 @@ interface ChannelOutboundInvoker { */ ChannelFuture close(); - /** - * Request to deregister this ChannelOutboundInvoker from the previous assigned {@link EventExecutor} and notify the - * {@link ChannelFuture} once the operation completes, either because the operation was successful or because of - * an error. - *

- * This will result in having the - * {@link ChannelOutboundHandler#deregister(ChannelHandlerContext, ChannelPromise)} - * method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the - * {@link Channel}. - * - */ - @Deprecated - ChannelFuture deregister(); - /** * Request to bind to the given {@link SocketAddress} and notify the {@link ChannelFuture} once the operation * completes, either because the operation was successful or because of an error. @@ -175,21 +159,6 @@ interface ChannelOutboundInvoker { */ ChannelFuture close(ChannelPromise promise); - /** - * Request to deregister this ChannelOutboundInvoker from the previous assigned {@link EventExecutor} and notify the - * {@link ChannelFuture} once the operation completes, either because the operation was successful or because of - * an error. - * - * The given {@link ChannelPromise} will be notified. - *

- * This will result in having the - * {@link ChannelOutboundHandler#deregister(ChannelHandlerContext, ChannelPromise)} - * method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the - * {@link Channel}. - */ - @Deprecated - ChannelFuture deregister(ChannelPromise promise); - /** * Request to Read data from the {@link Channel} into the first inbound buffer, triggers an * {@link ChannelInboundHandler#channelRead(ChannelHandlerContext, Object)} event if data was diff --git a/transport/src/main/java/io/netty/channel/ChannelPipeline.java b/transport/src/main/java/io/netty/channel/ChannelPipeline.java index f8e07bc9ee62..7090f5dced17 100644 --- a/transport/src/main/java/io/netty/channel/ChannelPipeline.java +++ b/transport/src/main/java/io/netty/channel/ChannelPipeline.java @@ -138,7 +138,6 @@ *

  • {@link ChannelHandlerContext#fireUserEventTriggered(Object)}
    • *
  • {@link ChannelHandlerContext#fireChannelWritabilityChanged()}
    • *
  • {@link ChannelHandlerContext#fireChannelInactive()}
    • - *
  • {@link ChannelHandlerContext#fireChannelUnregistered()}
    • * * *
  • Outbound event propagation methods: @@ -150,7 +149,6 @@ *
  • {@link ChannelHandlerContext#read()}
    • *
  • {@link ChannelHandlerContext#disconnect(ChannelPromise)}
    • *
  • {@link ChannelHandlerContext#close(ChannelPromise)}
    • - *
  • {@link ChannelHandlerContext#deregister(ChannelPromise)}
    • * * * @@ -597,10 +595,6 @@ T replace(Class oldHandlerType, String newName, @Override ChannelPipeline fireChannelRegistered(); - @Override - @Deprecated - ChannelPipeline fireChannelUnregistered(); - @Override ChannelPipeline fireChannelActive(); diff --git a/transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java b/transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java index 7ba94a1208f7..4e9e925c68b1 100644 --- a/transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java +++ b/transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java @@ -118,11 +118,6 @@ public void channelRegistered(ChannelHandlerContext ctx) throws Exception { inboundHandler.channelRegistered(ctx); } - @Override - public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { - inboundHandler.channelUnregistered(ctx); - } - @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { inboundHandler.channelActive(ctx); @@ -178,11 +173,6 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exce outboundHandler.close(ctx, promise); } - @Override - public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { - outboundHandler.deregister(ctx, promise); - } - @Override public void read(ChannelHandlerContext ctx) throws Exception { outboundHandler.read(ctx); diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java b/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java index 8f8fd8b3637f..38eab4139885 100644 --- a/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java +++ b/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java @@ -160,31 +160,6 @@ private void invokeChannelRegistered() { } } - @Override - public ChannelHandlerContext fireChannelUnregistered() { - final DefaultChannelHandlerContext next = findContextInbound(); - EventExecutor executor = next.executor(); - if (executor.inEventLoop()) { - next.invokeChannelUnregistered(); - } else { - executor.execute(new Runnable() { - @Override - public void run() { - next.invokeChannelUnregistered(); - } - }); - } - return this; - } - - private void invokeChannelUnregistered() { - try { - ((ChannelInboundHandler) handler).channelUnregistered(this); - } catch (Throwable t) { - notifyHandlerException(t); - } - } - @Override public ChannelHandlerContext fireChannelActive() { final DefaultChannelHandlerContext next = findContextInbound(); @@ -418,11 +393,6 @@ public ChannelFuture close() { return close(newPromise()); } - @Override - public ChannelFuture deregister() { - return deregister(newPromise()); - } - @Override public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) { if (localAddress == null) { @@ -558,34 +528,6 @@ private void invokeClose(ChannelPromise promise) { } } - @Override - public ChannelFuture deregister(final ChannelPromise promise) { - validatePromise(promise, false); - - final DefaultChannelHandlerContext next = findContextOutbound(); - EventExecutor executor = next.executor(); - if (executor.inEventLoop()) { - next.invokeDeregister(promise); - } else { - executor.execute(new Runnable() { - @Override - public void run() { - next.invokeDeregister(promise); - } - }); - } - - return promise; - } - - private void invokeDeregister(ChannelPromise promise) { - try { - ((ChannelOutboundHandler) handler).deregister(this, promise); - } catch (Throwable t) { - notifyOutboundHandlerException(t, promise); - } - } - @Override public ChannelHandlerContext read() { final DefaultChannelHandlerContext next = findContextOutbound(); diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java b/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java index c34682828690..a47a68a89dfd 100644 --- a/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java +++ b/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java @@ -731,17 +731,6 @@ public ChannelPipeline fireChannelRegistered() { return this; } - @Override - public ChannelPipeline fireChannelUnregistered() { - head.fireChannelUnregistered(); - - // Remove all handlers sequentially if channel is closed and unregistered. - if (!channel.isOpen()) { - teardownAll(); - } - return this; - } - /** * Removes all handlers from the pipeline one by one from tail (exclusive) to head (inclusive) to trigger * handlerRemoved(). Note that the tail handler is excluded because it's neither an outbound handler nor it @@ -765,6 +754,7 @@ public ChannelPipeline fireChannelActive() { @Override public ChannelPipeline fireChannelInactive() { head.fireChannelInactive(); + teardownAll(); return this; } @@ -826,11 +816,6 @@ public ChannelFuture close() { return tail.close(); } - @Override - public ChannelFuture deregister() { - return tail.deregister(); - } - @Override public ChannelPipeline flush() { tail.flush(); @@ -862,11 +847,6 @@ public ChannelFuture close(ChannelPromise promise) { return tail.close(promise); } - @Override - public ChannelFuture deregister(final ChannelPromise promise) { - return tail.deregister(promise); - } - @Override public ChannelPipeline read() { tail.read(); @@ -932,9 +912,6 @@ static final class TailHandler implements ChannelInboundHandler { @Override public void channelRegistered(ChannelHandlerContext ctx) throws Exception { } - @Override - public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { } - @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { } @@ -1018,11 +995,6 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exce unsafe.close(promise); } - @Override - public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { - unsafe.deregister(promise); - } - @Override public void read(ChannelHandlerContext ctx) { unsafe.beginRead(); diff --git a/transport/src/main/java/io/netty/channel/group/ChannelGroup.java b/transport/src/main/java/io/netty/channel/group/ChannelGroup.java index 88c7ab9c7561..e5b70c260daf 100644 --- a/transport/src/main/java/io/netty/channel/group/ChannelGroup.java +++ b/transport/src/main/java/io/netty/channel/group/ChannelGroup.java @@ -22,7 +22,6 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.channel.EventLoop; import io.netty.channel.ServerChannel; import io.netty.util.CharsetUtil; import io.netty.util.concurrent.GlobalEventExecutor; @@ -198,24 +197,4 @@ public interface ChannelGroup extends Set, Comparable { * the operation is done for all channels */ ChannelGroupFuture close(ChannelMatcher matcher); - - /** - * Deregister all {@link Channel}s in this group from their {@link EventLoop}. - * Please note that this operation is asynchronous as {@link Channel#deregister()} is. - * - * @return the {@link ChannelGroupFuture} instance that notifies when - * the operation is done for all channels - */ - @Deprecated - ChannelGroupFuture deregister(); - - /** - * Deregister all {@link Channel}s in this group from their {@link EventLoop} that match the given - * {@link ChannelMatcher}. Please note that this operation is asynchronous as {@link Channel#deregister()} is. - * - * @return the {@link ChannelGroupFuture} instance that notifies when - * the operation is done for all channels - */ - @Deprecated - ChannelGroupFuture deregister(ChannelMatcher matcher); } 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 2a83eaa26c37..c4666af20ffc 100644 --- a/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java +++ b/transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java @@ -170,10 +170,6 @@ public ChannelGroupFuture close() { public ChannelGroupFuture disconnect() { return disconnect(ChannelMatchers.all()); } - @Override - public ChannelGroupFuture deregister() { - return deregister(ChannelMatchers.all()); - } @Override public ChannelGroupFuture write(Object message) { @@ -268,29 +264,6 @@ public ChannelGroupFuture close(ChannelMatcher matcher) { return new DefaultChannelGroupFuture(this, futures, executor); } - @Override - public ChannelGroupFuture deregister(ChannelMatcher matcher) { - if (matcher == null) { - throw new NullPointerException("matcher"); - } - - Map futures = - new LinkedHashMap(size()); - - for (Channel c: serverChannels) { - if (matcher.matches(c)) { - futures.put(c, c.deregister()); - } - } - for (Channel c: nonServerChannels) { - if (matcher.matches(c)) { - futures.put(c, c.deregister()); - } - } - - return new DefaultChannelGroupFuture(this, futures, executor); - } - @Override public ChannelGroup flush(ChannelMatcher matcher) { for (Channel c: nonServerChannels) { diff --git a/transport/src/main/java/io/netty/channel/local/LocalChannel.java b/transport/src/main/java/io/netty/channel/local/LocalChannel.java index a52937ad389d..0fa5af4fbeaf 100644 --- a/transport/src/main/java/io/netty/channel/local/LocalChannel.java +++ b/transport/src/main/java/io/netty/channel/local/LocalChannel.java @@ -154,7 +154,7 @@ protected void doRegister() throws Exception { if (peer != null) { state = 2; - peer.remoteAddress = parent().localAddress(); + peer.remoteAddress = parent() == null ? null : parent().localAddress(); peer.state = 2; // Always call peer.eventLoop().execute() even if peer.eventLoop().inEventLoop() is true. @@ -219,9 +219,6 @@ public void run() { @Override protected void doDeregister() throws Exception { - if (isOpen()) { - unsafe().close(unsafe().voidPromise()); - } ((SingleThreadEventExecutor) eventLoop()).removeShutdownHook(shutdownHook); } diff --git a/transport/src/test/java/io/netty/channel/AbstractEventLoopTest.java b/transport/src/test/java/io/netty/channel/AbstractEventLoopTest.java index e87934e0e347..32c7c461c064 100644 --- a/transport/src/test/java/io/netty/channel/AbstractEventLoopTest.java +++ b/transport/src/test/java/io/netty/channel/AbstractEventLoopTest.java @@ -15,21 +15,14 @@ */ package io.netty.channel; -import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.socket.ServerSocketChannel; -import io.netty.channel.socket.SocketChannel; -import io.netty.util.concurrent.DefaultEventExecutorGroup; -import io.netty.util.concurrent.EventExecutor; -import io.netty.util.concurrent.EventExecutorGroup; -import org.junit.Test; - -import static org.junit.Assert.*; public abstract class AbstractEventLoopTest { /** * Test for https://github.com/netty/netty/issues/803 */ + /* @Test public void testReregister() { EventLoopGroup group = newEventLoopGroup(); @@ -53,7 +46,6 @@ public void initChannel(ServerSocketChannel ch) throws Exception { EventExecutor executor = future.channel().pipeline().context(TestChannelHandler2.class).executor(); EventExecutor executor1 = future.channel().pipeline().context(TestChannelHandler.class).executor(); - future.channel().deregister().awaitUninterruptibly(); Channel channel = group2.register(future.channel()).awaitUninterruptibly().channel(); EventExecutor executorNew = channel.pipeline().context(TestChannelHandler.class).executor(); assertNotSame(executor1, executorNew); @@ -66,7 +58,7 @@ private static final class TestChannelHandler2 extends ChannelDuplexHandler { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { } } - +*/ protected abstract EventLoopGroup newEventLoopGroup(); protected abstract Class newChannel(); } diff --git a/transport/src/test/java/io/netty/channel/BaseChannelTest.java b/transport/src/test/java/io/netty/channel/BaseChannelTest.java index 907ed2def489..b3aaf3d1f458 100644 --- a/transport/src/test/java/io/netty/channel/BaseChannelTest.java +++ b/transport/src/test/java/io/netty/channel/BaseChannelTest.java @@ -25,6 +25,8 @@ import io.netty.channel.local.LocalEventLoopGroup; import io.netty.channel.local.LocalServerChannel; +import java.io.UnsupportedEncodingException; + class BaseChannelTest { private final LoggingHandler loggingHandler; @@ -64,6 +66,18 @@ static ByteBuf createTestBuf(int len) { return buf; } + static Object createTestBuf(String string) throws UnsupportedEncodingException { + byte[] buf = string.getBytes("US-ASCII"); + return createTestBuf(buf); + } + + static Object createTestBuf(byte[] buf) { + ByteBuf ret = createTestBuf(buf.length); + ret.clear(); + ret.writeBytes(buf); + return ret; + } + void assertLog(String expected) { String actual = this.loggingHandler.getLog(); assertEquals(expected, actual); diff --git a/transport/src/test/java/io/netty/channel/LoggingHandler.java b/transport/src/test/java/io/netty/channel/LoggingHandler.java index 1fc72d8f40de..c58050439ab2 100644 --- a/transport/src/test/java/io/netty/channel/LoggingHandler.java +++ b/transport/src/test/java/io/netty/channel/LoggingHandler.java @@ -21,9 +21,8 @@ final class LoggingHandler implements ChannelInboundHandler, ChannelOutboundHandler { - static enum Event { WRITE, FLUSH, BIND, CONNECT, DISCONNECT, CLOSE, DEREGISTER, READ, WRITABILITY, - HANDLER_ADDED, HANDLER_REMOVED, EXCEPTION, READ_COMPLETE, REGISTERED, UNREGISTERED, ACTIVE, INACTIVE, - USER }; + static enum Event { WRITE, FLUSH, BIND, CONNECT, DISCONNECT, CLOSE, READ, WRITABILITY, HANDLER_ADDED, + HANDLER_REMOVED, EXCEPTION, READ_COMPLETE, REGISTERED, ACTIVE, INACTIVE, USER }; private StringBuilder log = new StringBuilder(); @@ -67,12 +66,6 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exce ctx.close(promise); } - @Override - public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { - log(Event.DEREGISTER); - ctx.deregister(promise); - } - @Override public void read(ChannelHandlerContext ctx) throws Exception { log(Event.READ); @@ -106,12 +99,6 @@ public void channelRegistered(ChannelHandlerContext ctx) throws Exception { ctx.fireChannelRegistered(); } - @Override - public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { - log(Event.UNREGISTERED); - ctx.fireChannelUnregistered(); - } - @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { log(Event.ACTIVE); diff --git a/transport/src/test/java/io/netty/channel/ReentrantChannelTest.java b/transport/src/test/java/io/netty/channel/ReentrantChannelTest.java index eb35ef8ee4b2..67cd5fbe1777 100644 --- a/transport/src/test/java/io/netty/channel/ReentrantChannelTest.java +++ b/transport/src/test/java/io/netty/channel/ReentrantChannelTest.java @@ -15,11 +15,17 @@ */ package io.netty.channel; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import io.netty.bootstrap.Bootstrap; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.LoggingHandler.Event; import io.netty.channel.local.LocalAddress; +import io.netty.util.concurrent.Future; +import io.netty.util.concurrent.GenericFutureListener; + +import java.nio.channels.ClosedChannelException; import org.junit.Test; @@ -91,4 +97,141 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exceptio "WRITABILITY: writable=true\n"); } + @Test + public void testWriteFlushPingPong() throws Exception { + + LocalAddress addr = new LocalAddress("testWriteFlushPingPong"); + + ServerBootstrap sb = getLocalServerBootstrap(); + sb.bind(addr).sync().channel(); + + Bootstrap cb = getLocalClientBootstrap(); + + setInterest(Event.WRITE, Event.FLUSH, Event.CLOSE, Event.EXCEPTION); + + Channel clientChannel = cb.connect(addr).sync().channel(); + + clientChannel.pipeline().addLast(new ChannelOutboundHandlerAdapter() { + + int writeCount; + int flushCount; + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + if (writeCount < 5) { + writeCount++; + ctx.channel().flush(); + } + super.write(ctx, msg, promise); + } + + @Override + public void flush(ChannelHandlerContext ctx) throws Exception { + if (flushCount < 5) { + flushCount++; + ctx.channel().write(createTestBuf(2000)); + } + super.flush(ctx); + } + }); + + clientChannel.writeAndFlush(createTestBuf(2000)); + clientChannel.close().sync(); + + assertLog( + "WRITE\n" + + "FLUSH\n" + + "WRITE\n" + + "FLUSH\n" + + "WRITE\n" + + "FLUSH\n" + + "WRITE\n" + + "FLUSH\n" + + "WRITE\n" + + "FLUSH\n" + + "WRITE\n" + + "FLUSH\n" + + "CLOSE\n"); + } + + @Test + public void testCloseInFlush() throws Exception { + + LocalAddress addr = new LocalAddress("testCloseInFlush"); + + ServerBootstrap sb = getLocalServerBootstrap(); + sb.bind(addr).sync().channel(); + + Bootstrap cb = getLocalClientBootstrap(); + + setInterest(Event.WRITE, Event.FLUSH, Event.CLOSE, Event.EXCEPTION); + + Channel clientChannel = cb.connect(addr).sync().channel(); + + clientChannel.pipeline().addLast(new ChannelOutboundHandlerAdapter() { + + @Override + public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + promise.addListener(new GenericFutureListener>() { + @Override + public void operationComplete(Future future) throws Exception { + ctx.channel().close(); + } + }); + super.write(ctx, msg, promise); + ctx.channel().flush(); + } + }); + + clientChannel.write(createTestBuf(2000)).sync(); + clientChannel.closeFuture().sync(); + + assertLog( + "WRITE\n" + + "FLUSH\n" + + "CLOSE\n"); + } + + @Test + public void testFlushFailure() throws Exception { + + LocalAddress addr = new LocalAddress("testFlushFailure"); + + ServerBootstrap sb = getLocalServerBootstrap(); + sb.bind(addr).sync().channel(); + + Bootstrap cb = getLocalClientBootstrap(); + + setInterest(Event.WRITE, Event.FLUSH, Event.CLOSE, Event.EXCEPTION); + + Channel clientChannel = cb.connect(addr).sync().channel(); + + clientChannel.pipeline().addLast(new ChannelOutboundHandlerAdapter() { + + @Override + public void flush(ChannelHandlerContext ctx) throws Exception { + throw new Exception("intentional failure"); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + ctx.close(); + } + }); + + try { + clientChannel.writeAndFlush(createTestBuf(2000)).sync(); + fail(); + } catch (Throwable cce) { + // FIXME: shouldn't this contain the "intentional failure" exception? + assertEquals(ClosedChannelException.class, cce.getClass()); + } + + clientChannel.closeFuture().sync(); + + assertLog( + "WRITE\n" + + "CLOSE\n"); + } + } diff --git a/transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest3.java b/transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest3.java index bfb3b4bd673a..7e9292a878ec 100644 --- a/transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest3.java +++ b/transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest3.java @@ -28,11 +28,6 @@ import io.netty.util.concurrent.DefaultEventExecutorGroup; import io.netty.util.concurrent.DefaultThreadFactory; import io.netty.util.concurrent.EventExecutorGroup; -import org.junit.AfterClass; -import org.junit.Assert; -import org.junit.BeforeClass; -import org.junit.Ignore; -import org.junit.Test; import java.util.Deque; import java.util.LinkedList; @@ -41,6 +36,12 @@ import java.util.UUID; import java.util.concurrent.ConcurrentLinkedDeque; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Ignore; +import org.junit.Test; + public class LocalTransportThreadModelTest3 { enum EventType { @@ -49,7 +50,6 @@ enum EventType { MESSAGE_RECEIVED_LAST, INACTIVE, ACTIVE, - UNREGISTERED, REGISTERED, MESSAGE_RECEIVED, WRITE, @@ -198,14 +198,9 @@ public void run() { ch.close().sync(); - while (events.peekLast() != EventType.UNREGISTERED) { - Thread.sleep(10); - } - expectedEvents.addFirst(EventType.ACTIVE); expectedEvents.addFirst(EventType.REGISTERED); expectedEvents.addLast(EventType.INACTIVE); - expectedEvents.addLast(EventType.UNREGISTERED); for (;;) { EventType event = events.poll(); @@ -292,11 +287,6 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception { events.add(EventType.ACTIVE); } - @Override - public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { - events.add(EventType.UNREGISTERED); - } - @Override public void channelRegistered(ChannelHandlerContext ctx) throws Exception { events.add(EventType.REGISTERED);