Skip to content

Commit

Permalink
DefaultChannelPipeline will not invoke handler if events are fired fr…
Browse files Browse the repository at this point in the history
…om handlerAdded

Motiviation:
DefaultChannelPipeline and AbstractChannelHandlerContext maintain state
which indicates if a ChannelHandler should be invoked or not. However
the state is updated to allow the handler to be invoked only after the
handlerAdded method completes. If the handlerAdded method generates
events which may result in other methods being invoked on that handler
they will be missed.

Modifications:
- DefaultChannelPipeline should set the state before calling
handlerAdded

Result:
DefaultChannelPipeline will allow events to be processed during the
handlerAdded process.
  • Loading branch information
Scottmitch authored and normanmaurer committed Jan 24, 2018
1 parent 27ff153 commit 2d815fa
Show file tree
Hide file tree
Showing 2 changed files with 68 additions and 2 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -611,8 +611,10 @@ private static void checkMultiplicity(ChannelHandler handler) {

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
try {
ctx.handler().handlerAdded(ctx);
// We must call setAddComplete before calling handlerAdded. Otherwise if the handlerAdded method generates
// any pipeline events ctx.handler() will miss them because the state will not allow it.
ctx.setAddComplete();
ctx.handler().handlerAdded(ctx);
} catch (Throwable t) {
boolean removed = false;
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,14 @@
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;

import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

public class DefaultChannelPipelineTest {

Expand Down Expand Up @@ -1102,6 +1109,63 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
}
}

@Test(timeout = 5000)
public void handlerAddedStateUpdatedBeforeHandlerAddedDoneForceEventLoop() throws InterruptedException {
handlerAddedStateUpdatedBeforeHandlerAddedDone(true);
}

@Test(timeout = 5000)
public void handlerAddedStateUpdatedBeforeHandlerAddedDoneOnCallingThread() throws InterruptedException {
handlerAddedStateUpdatedBeforeHandlerAddedDone(false);
}

private static void handlerAddedStateUpdatedBeforeHandlerAddedDone(boolean executeInEventLoop)
throws InterruptedException {
final ChannelPipeline pipeline = new LocalChannel().pipeline();
final Object userEvent = new Object();
final Object writeObject = new Object();
final CountDownLatch doneLatch = new CountDownLatch(1);

group.register(pipeline.channel());

Runnable r = new Runnable() {
@Override
public void run() {
pipeline.addLast(new ChannelInboundHandlerAdapter() {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
if (evt == userEvent) {
ctx.write(writeObject);
}
ctx.fireUserEventTriggered(evt);
}
});
pipeline.addFirst(new ChannelDuplexHandler() {
@Override
public void handlerAdded(ChannelHandlerContext ctx) {
ctx.fireUserEventTriggered(userEvent);
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
if (msg == writeObject) {
doneLatch.countDown();
}
ctx.write(msg, promise);
}
});
}
};

if (executeInEventLoop) {
pipeline.channel().eventLoop().execute(r);
} else {
r.run();
}

doneLatch.await();
}

private static final class TestTask implements Runnable {

private final ChannelPipeline pipeline;
Expand Down

0 comments on commit 2d815fa

Please sign in to comment.