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

Synchronized between 4.1 and master again (part 2)

Browse Source 
Motivation:
4 and 5 were diverged long time ago and we recently reverted some of the
early commits in master.  We must make sure 4.1 and master are not very
different now.

Modification:
Remove ChannelHandlerInvoker.writeAndFlush(...) and the related
implementations.

Result:
4.1 and master got closer.
This commit is contained in:
Trustin Lee 2014-04-25 13:44:04 +09:00
parent 8a7d8a8ab7
commit 872d4c5bc1
17 changed files with 102 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
codec-http/src/main/java/io/netty/handler/codec/http/DefaultHttpHeaders.java
View File

@ -90,10 +90,10 @@ public class DefaultHttpHeaders extends HttpHeaders {
CharSequence strVal;
if (validate) {
validateHeaderName0(name);
strVal = toCharsequence(value);
strVal = toCharSequence(value);
validateHeaderValue(strVal);
} else {
strVal = toCharsequence(value);
strVal = toCharSequence(value);
}
int h = hash(name);
int i = index(h);
@ -109,7 +109,7 @@ public class DefaultHttpHeaders extends HttpHeaders {
int h = hash(name);
int i = index(h);
for (Object v: values) {
CharSequence vstr = toCharsequence(v);
CharSequence vstr = toCharSequence(v);
if (validate) {
validateHeaderValue(vstr);
}
@ -181,10 +181,10 @@ public class DefaultHttpHeaders extends HttpHeaders {
CharSequence strVal;
if (validate) {
validateHeaderName0(name);
strVal = toCharsequence(value);
strVal = toCharSequence(value);
validateHeaderValue(strVal);
} else {
strVal = toCharsequence(value);
strVal = toCharSequence(value);
}
int h = hash(name);
int i = index(h);
@ -210,7 +210,7 @@ public class DefaultHttpHeaders extends HttpHeaders {
if (v == null) {
break;
}
CharSequence strVal = toCharsequence(v);
CharSequence strVal = toCharSequence(v);
if (validate) {
validateHeaderValue(strVal);
}
@ -344,7 +344,7 @@ public class DefaultHttpHeaders extends HttpHeaders {
}
}
private static CharSequence toCharsequence(Object value) {
private static CharSequence toCharSequence(Object value) {
if (value == null) {
return null;
}

6
codec-http/src/main/java/io/netty/handler/codec/http/DefaultLastHttpContent.java
View File

@ -114,9 +114,9 @@ public class DefaultLastHttpContent extends DefaultHttpContent implements LastHt
@Override
void validateHeaderName0(CharSequence name) {
super.validateHeaderName0(name);
if (equalsIgnoreCase(name, HttpHeaders.Names.CONTENT_LENGTH) ||
equalsIgnoreCase(name, HttpHeaders.Names.TRANSFER_ENCODING) ||
equalsIgnoreCase(name, HttpHeaders.Names.TRAILER)) {
if (equalsIgnoreCase(HttpHeaders.Names.CONTENT_LENGTH, name) ||
equalsIgnoreCase(HttpHeaders.Names.TRANSFER_ENCODING, name) ||
equalsIgnoreCase(HttpHeaders.Names.TRAILER, name)) {
throw new IllegalArgumentException(
"prohibited trailing header: " + name);
}

2
codec-http/src/main/java/io/netty/handler/codec/http/HttpHeaders.java
View File

@ -933,7 +933,7 @@ public abstract class HttpHeaders implements Iterable<Map.Entry<String, String>>
/**
* Sets the {@code "Host"} header.
*/
public static void setHost(HttpMessage message, String value) {
public static void setHost(HttpMessage message, CharSequence value) {
message.headers().set(Names.HOST, value);
}

2
example/src/main/java/io/netty/example/spdy/server/SpdyServer.java
View File

@ -38,7 +38,7 @@ import io.netty.channel.socket.nio.NioServerSocketChannel;
* <p>
* Once started, you can test the server with your
* <a href="http://en.wikipedia.org/wiki/SPDY#Browser_support_and_usage">SPDY enabled web browser</a> by navigating
* to to <a href="https://localhost:8443/">https://localhost:8443/</a>
* to <a href="https://localhost:8443/">https://localhost:8443/</a>
*/
public class SpdyServer {

2
transport-sctp/src/main/java/io/netty/channel/sctp/nio/NioSctpServerChannel.java
View File

@ -64,7 +64,7 @@ public class NioSctpServerChannel extends AbstractNioMessageChannel
*/
public NioSctpServerChannel() {
super(null, newSocket(), SelectionKey.OP_ACCEPT);
config = new DefaultSctpServerChannelConfig(this, javaChannel());
config = new NioSctpServerChannelConfig(this, javaChannel());
}
@Override

3
transport-udt/src/main/java/io/netty/channel/udt/nio/NioUdtMessageRendezvousChannel.java
View File

@ -23,8 +23,7 @@ import io.netty.channel.udt.UdtMessage;
* <p>
* Note: send/receive must use {@link UdtMessage} in the pipeline
*/
public class NioUdtMessageRendezvousChannel extends
NioUdtMessageConnectorChannel {
public class NioUdtMessageRendezvousChannel extends NioUdtMessageConnectorChannel {
public NioUdtMessageRendezvousChannel() {
super(NioUdtProvider.newRendezvousChannelUDT(TypeUDT.DATAGRAM));

6
transport/src/main/java/io/netty/bootstrap/AbstractBootstrap.java
View File

@ -24,9 +24,9 @@ import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPromise;
import io.netty.channel.DefaultChannelPromise;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import io.netty.util.internal.StringUtil;
@ -217,7 +217,7 @@ public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C ext
public abstract B clone();
/**
* Create a new {@link Channel} and register it with an {@link EventExecutorGroup}.
* Create a new {@link Channel} and register it with an {@link EventLoop}.
*/
public ChannelFuture register() {
validate();
@ -448,7 +448,7 @@ public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C ext
@Override
public String toString() {
return clazz.getSimpleName() + ".class";
return StringUtil.simpleClassName(clazz) + ".class";
}
}
}

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

@ -416,8 +416,8 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
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;
}
@ -493,6 +493,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
closeIfClosed();
return;
}
if (!wasActive && isActive()) {
invokeLater(new OneTimeTask() {
@Override
@ -501,6 +502,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
}
});
}
safeSetSuccess(promise);
}
@ -518,6 +520,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
closeIfClosed();
return;
}
if (wasActive && !isActive()) {
invokeLater(new OneTimeTask() {
@Override
@ -526,6 +529,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
}
});
}
safeSetSuccess(promise);
closeIfClosed(); // doDisconnect() might have closed the channel
}
@ -749,18 +753,22 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
}
private void invokeLater(Runnable task) {
// This method is used by outbound operation implementations to trigger an inbound event later.
// They do not trigger an inbound event immediately because an outbound operation might have been
// triggered by another inbound event handler method. If fired immediately, the call stack
// will look like this for example:
//
// handlerA.inboundBufferUpdated() - (1) an inbound handler method closes a connection.
// -> handlerA.ctx.close()
// -> channel.unsafe.close()
// -> handlerA.channelInactive() - (2) another inbound handler method called while in (1) yet
//
// which means the execution of two inbound handler methods of the same handler overlap undesirably.
eventLoop().execute(task);
try {
// This method is used by outbound operation implementations to trigger an inbound event later.
// They do not trigger an inbound event immediately because an outbound operation might have been
// triggered by another inbound event handler method. If fired immediately, the call stack
// will look like this for example:
//
// handlerA.inboundBufferUpdated() - (1) an inbound handler method closes a connection.
// -> handlerA.ctx.close()
// -> channel.unsafe.close()
// -> handlerA.channelInactive() - (2) another inbound handler method called while in (1) yet
//
// which means the execution of two inbound handler methods of the same handler overlap undesirably.
eventLoop().execute(task);
} catch (RejectedExecutionException e) {
logger.warn("Can't invoke task later as EventLoop rejected it", e);
}
}
}
@ -787,8 +795,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
protected abstract SocketAddress remoteAddress0();
/**
* Is called after the {@link Channel} is registered with its {@link EventLoop} as part of the register
* process.
* Is called after the {@link Channel} is registered with its {@link EventLoop} as part of the register process.
*
* Sub-classes may override this method
*/

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

@ -432,7 +432,7 @@ public interface Channel extends AttributeMap, Comparable<Channel> {
* <li>{@link #localAddress()}</li>
* <li>{@link #remoteAddress()}</li>
* <li>{@link #closeForcibly()}</li>
* <li>{@link #register(ChannelPromise)}</li>
* <li>{@link #register(EventLoop, ChannelPromise)}</li>
* <li>{@link #deregister(ChannelPromise)}</li>
* <li>{@link #voidPromise()}</li>
* </ul>

12
transport/src/main/java/io/netty/channel/ChannelHandlerAppender.java
View File

@ -181,9 +181,9 @@ public class ChannelHandlerAppender extends ChannelHandlerAdapter {
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
added = true;
DefaultChannelPipeline pipeline = (DefaultChannelPipeline) ctx.pipeline();
String name = ctx.name();
DefaultChannelHandlerContext dctx = (DefaultChannelHandlerContext) ctx;
DefaultChannelPipeline pipeline = (DefaultChannelPipeline) dctx.pipeline();
String name = dctx.name();
try {
for (Entry e: handlers) {
String oldName = name;
@ -192,10 +192,10 @@ public class ChannelHandlerAppender extends ChannelHandlerAdapter {
} else {
name = e.name;
}
// Pass in direct the invoker to eliminate the possibility of an IllegalStateException
// Note that we do not use dctx.invoker() because it raises an IllegalStateExxception
// if the Channel is not registered yet.
DefaultChannelHandlerContext context = (DefaultChannelHandlerContext) ctx;
pipeline.addAfter(context.invoker, oldName, name, e.handler);
pipeline.addAfter(dctx.invoker, oldName, name, e.handler);
}
} finally {
if (selfRemoval) {

56
transport/src/main/java/io/netty/channel/ChannelHandlerContext.java
View File

@ -35,8 +35,8 @@ import java.nio.channels.Channels;
*
* <h3>Notify</h3>
*
* You can notify the closest handler in the
* same {@link ChannelPipeline} by calling one of the various methods provided here.
* You can notify the closest handler in the same {@link ChannelPipeline} by calling one of the various methods
* provided here.
*
* Please refer to {@link ChannelPipeline} to understand how an event flows.
*
@ -169,8 +169,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} was registered to its {@link EventLoop}.
*
* This will result in having the {@link ChannelHandler#channelRegistered(ChannelHandlerContext)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelHandler#channelRegistered(ChannelHandlerContext)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelRegistered();
@ -178,8 +178,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} was unregistered from its {@link EventLoop}.
*
* This will result in having the {@link ChannelHandler#channelUnregistered(ChannelHandlerContext)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelHandler#channelUnregistered(ChannelHandlerContext)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelUnregistered();
@ -187,8 +187,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} is active now, which means it is connected.
*
* This will result in having the {@link ChannelHandler#channelActive(ChannelHandlerContext)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelHandler#channelActive(ChannelHandlerContext)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelActive();
@ -196,8 +196,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} is inactive now, which means it is closed.
*
* This will result in having the {@link ChannelHandler#channelInactive(ChannelHandlerContext)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelHandler#channelInactive(ChannelHandlerContext)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelInactive();
@ -205,8 +205,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} received an {@link Throwable} in one of its inbound operations.
*
* This will result in having the {@link ChannelHandler#exceptionCaught(ChannelHandlerContext, Throwable)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelHandler#exceptionCaught(ChannelHandlerContext, Throwable)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireExceptionCaught(Throwable cause);
@ -214,8 +214,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} received an user defined event.
*
* This will result in having the {@link ChannelHandler#userEventTriggered(ChannelHandlerContext, Object)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelHandler#userEventTriggered(ChannelHandlerContext, Object)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireUserEventTriggered(Object event);
@ -224,7 +224,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* A {@link Channel} received a message.
*
* This will result in having the {@link ChannelHandler#channelRead(ChannelHandlerContext, Object)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelRead(Object msg);
@ -247,7 +247,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#bind(ChannelHandlerContext, SocketAddress, ChannelPromise)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture bind(SocketAddress localAddress);
@ -262,7 +262,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture connect(SocketAddress remoteAddress);
@ -274,7 +274,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress);
@ -285,7 +285,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#disconnect(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture disconnect();
@ -299,7 +299,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#close(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture close();
@ -311,7 +311,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#deregister(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*
*/
@ -325,7 +325,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#bind(ChannelHandlerContext, SocketAddress, ChannelPromise)} method
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);
@ -343,7 +343,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise);
@ -357,7 +357,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
@ -370,7 +370,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#disconnect(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture disconnect(ChannelPromise promise);
@ -385,7 +385,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#close(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture close(ChannelPromise promise);
@ -399,7 +399,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#deregister(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture deregister(ChannelPromise promise);
@ -413,7 +413,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelHandler#read(ChannelHandlerContext)}
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext read();

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

@ -67,6 +67,9 @@ public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutor
return (EventLoop) super.next();
}
@Override
protected abstract EventLoop newChild(Executor executor, Object... args) throws Exception;
@Override
public ChannelFuture register(Channel channel) {
return next().register(channel);

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

@ -92,7 +92,7 @@ final class EmbeddedEventLoop extends AbstractEventLoop implements ChannelHandle
@Override
public ChannelFuture register(Channel channel) {
return register(channel, channel.newPromise());
return register(channel, new DefaultChannelPromise(channel, this));
}
@Override

1
transport/src/main/java/io/netty/channel/oio/AbstractOioMessageChannel.java
View File

@ -18,6 +18,7 @@ package io.netty.channel.oio;
import io.netty.channel.Channel;
import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelPipeline;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

6
transport/src/test/java/io/netty/channel/SingleThreadEventLoopTest.java
View File

@ -137,7 +137,7 @@ public class SingleThreadEventLoopTest {
testScheduleTask(loopB);
}
private static void testScheduleTask(EventExecutor loopA) throws InterruptedException, ExecutionException {
private static void testScheduleTask(EventLoop loopA) throws InterruptedException, ExecutionException {
long startTime = System.nanoTime();
final AtomicLong endTime = new AtomicLong();
loopA.schedule(new Runnable() {
@ -159,7 +159,7 @@ public class SingleThreadEventLoopTest {
testScheduleTaskAtFixedRate(loopB);
}
private static void testScheduleTaskAtFixedRate(EventExecutor loopA) throws InterruptedException {
private static void testScheduleTaskAtFixedRate(EventLoop loopA) throws InterruptedException {
final Queue<Long> timestamps = new LinkedBlockingQueue<Long>();
ScheduledFuture<?> f = loopA.scheduleAtFixedRate(new Runnable() {
@Override
@ -199,7 +199,7 @@ public class SingleThreadEventLoopTest {
testScheduleLaggyTaskAtFixedRate(loopB);
}
private static void testScheduleLaggyTaskAtFixedRate(EventExecutor loopA) throws InterruptedException {
private static void testScheduleLaggyTaskAtFixedRate(EventLoop loopA) throws InterruptedException {
final Queue<Long> timestamps = new LinkedBlockingQueue<Long>();
ScheduledFuture<?> f = loopA.scheduleAtFixedRate(new Runnable() {
@Override

16
transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest.java
View File

@ -26,7 +26,9 @@ import io.netty.channel.ChannelPromise;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.EventLoopGroup;
import io.netty.util.ReferenceCountUtil;
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;
@ -83,8 +85,8 @@ public class LocalTransportThreadModelTest {
@Test(timeout = 5000)
public void testStagedExecution() throws Throwable {
EventLoopGroup l = new DefaultEventLoopGroup(4, new DefaultThreadFactory("l"));
EventLoopGroup e1 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e1"));
EventLoopGroup e2 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e2"));
EventExecutorGroup e1 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e1"));
EventExecutorGroup e2 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e2"));
ThreadNameAuditor h1 = new ThreadNameAuditor();
ThreadNameAuditor h2 = new ThreadNameAuditor();
ThreadNameAuditor h3 = new ThreadNameAuditor(true);
@ -226,11 +228,11 @@ public class LocalTransportThreadModelTest {
@Ignore
public void testConcurrentMessageBufferAccess() throws Throwable {
EventLoopGroup l = new DefaultEventLoopGroup(4, new DefaultThreadFactory("l"));
EventLoopGroup e1 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e1"));
EventLoopGroup e2 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e2"));
EventLoopGroup e3 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e3"));
EventLoopGroup e4 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e4"));
EventLoopGroup e5 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e5"));
EventExecutorGroup e1 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e1"));
EventExecutorGroup e2 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e2"));
EventExecutorGroup e3 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e3"));
EventExecutorGroup e4 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e4"));
EventExecutorGroup e5 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e5"));
try {
final MessageForwarder1 h1 = new MessageForwarder1();

18
transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest3.java
View File

@ -25,7 +25,9 @@ import io.netty.channel.ChannelPromise;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.EventLoopGroup;
import io.netty.util.ReferenceCountUtil;
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;
@ -114,13 +116,13 @@ public class LocalTransportThreadModelTest3 {
private static void testConcurrentAddRemove(boolean inbound) throws Exception {
EventLoopGroup l = new DefaultEventLoopGroup(4, new DefaultThreadFactory("l"));
EventLoopGroup e1 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e1"));
EventLoopGroup e2 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e2"));
EventLoopGroup e3 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e3"));
EventLoopGroup e4 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e4"));
EventLoopGroup e5 = new DefaultEventLoopGroup(4, new DefaultThreadFactory("e5"));
EventExecutorGroup e1 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e1"));
EventExecutorGroup e2 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e2"));
EventExecutorGroup e3 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e3"));
EventExecutorGroup e4 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e4"));
EventExecutorGroup e5 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e5"));
final EventLoopGroup[] groups = { e1, e2, e3, e4, e5 };
final EventExecutorGroup[] groups = { e1, e2, e3, e4, e5 };
try {
Deque<EventType> events = new ConcurrentLinkedDeque<EventType>();
final EventForwarder h1 = new EventForwarder();
@ -202,7 +204,7 @@ public class LocalTransportThreadModelTest3 {
for (;;) {
EventType event = events.poll();
if (event == null) {
Assert.assertTrue("Missing events:" + expectedEvents.toString(), expectedEvents.isEmpty());
Assert.assertTrue("Missing events:" + expectedEvents, expectedEvents.isEmpty());
break;
}
Assert.assertEquals(event, expectedEvents.poll());
@ -250,7 +252,7 @@ public class LocalTransportThreadModelTest3 {
private final Queue<EventType> events;
private final boolean inbound;
public EventRecorder(Queue<EventType> events, boolean inbound) {
EventRecorder(Queue<EventType> events, boolean inbound) {
this.events = events;
this.inbound = inbound;
}