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 14:00:04 +09:00
parent 8c3eaf3b56
commit b9039eaa82
17 changed files with 90 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
common/src/main/java/io/netty/util/concurrent/EventExecutor.java
View File

@ -65,14 +65,14 @@ public interface EventExecutor extends EventExecutorGroup {
<V> ProgressivePromise<V> newProgressivePromise();
/**
* Create a new {@link Future} which is marked as successes already. So {@link Future#isSuccess()}
* Create a new {@link Future} which is marked as succeeded already. So {@link Future#isSuccess()}
* will return {@code true}. All {@link FutureListener} added to it will be notified directly. Also
* every call of blocking methods will just return without blocking.
*/
<V> Future<V> newSucceededFuture(V result);
/**
* Create a new {@link Future} which is marked as fakued already. So {@link Future#isSuccess()}
* Create a new {@link Future} which is marked as failed already. So {@link Future#isSuccess()}
* will return {@code false}. All {@link FutureListener} added to it will be notified directly. Also
* every call of blocking methods will just return without blocking.
*/

2
example/src/main/java/io/netty/example/uptime/UptimeClient.java
View File

@ -18,8 +18,8 @@ package io.netty.example.uptime;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;

2
handler/src/main/java/io/netty/handler/traffic/package-info.java
View File

@ -30,7 +30,7 @@
* <li> <tt>{@link io.netty.handler.traffic.AbstractTrafficShapingHandler}</tt>: this abstract class implements
* the kernel of traffic shaping. It could be extended to fit your needs. Two classes are proposed as default
* implementations: see {@link io.netty.handler.traffic.ChannelTrafficShapingHandler} and
* see {@link io.netty.handler.traffic.GlobalTrafficShapingHandler} respectively for Channel traffic shaping and
* {@link io.netty.handler.traffic.GlobalTrafficShapingHandler} respectively for Channel traffic shaping and
* global traffic shaping.</li>
* </ul></p>
*

2
transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java
View File

@ -66,7 +66,7 @@ public final class EpollSocketChannel extends AbstractEpollChannel implements So
private volatile boolean inputShutdown;
private volatile boolean outputShutdown;
EpollSocketChannel(Channel parent, int fd) throws IOException {
EpollSocketChannel(Channel parent, int fd) {
super(parent, fd, Native.EPOLLIN, true);
config = new EpollSocketChannelConfig(this);
// Directly cache the remote and local addresses

1
transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannelOption.java
View File

@ -25,7 +25,6 @@ import static io.netty.channel.ChannelOption.*;
/**
* Option for configuring a serial port connection
*/
public final class RxtxChannelOption {
private static final Class<RxtxChannelOption> T = RxtxChannelOption.class;

5
transport-sctp/src/main/java/io/netty/channel/sctp/SctpChannelConfig.java
View File

@ -18,6 +18,7 @@ package io.netty.channel.sctp;
import com.sun.nio.sctp.SctpStandardSocketOptions.InitMaxStreams;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelOption;
import io.netty.channel.MessageSizeEstimator;
import io.netty.channel.RecvByteBufAllocator;
@ -33,9 +34,9 @@ import io.netty.channel.RecvByteBufAllocator;
* <tr>
* <th>Name</th><th>Associated setter method</th>
* </tr><tr>
* <td>{@link io.netty.channel.ChannelOption#SO_RCVBUF}</td><td>{@link #setReceiveBufferSize(int)}</td>
* <td>{@link ChannelOption#SO_RCVBUF}</td><td>{@link #setReceiveBufferSize(int)}</td>
* </tr><tr>
* <td>{@link io.netty.channel.ChannelOption#SO_SNDBUF}</td><td>{@link #setSendBufferSize(int)}</td>
* <td>{@link ChannelOption#SO_SNDBUF}</td><td>{@link #setSendBufferSize(int)}</td>
* </tr><tr>
* <td>{@link SctpChannelOption#SCTP_NODELAY}</td><td>{@link #setSctpNoDelay(boolean)}}</td>
* </tr><tr>

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

@ -23,11 +23,9 @@ 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));
}
}

57
transport-udt/src/main/java/io/netty/channel/udt/nio/NioUdtProvider.java
View File

@ -201,35 +201,35 @@ public final class NioUdtProvider<T extends UdtChannel> implements ChannelFactor
@Override
public T newChannel() {
switch (kind) {
case ACCEPTOR:
switch (type) {
case DATAGRAM:
return (T) new NioUdtMessageAcceptorChannel();
case STREAM:
return (T) new NioUdtByteAcceptorChannel();
case ACCEPTOR:
switch (type) {
case DATAGRAM:
return (T) new NioUdtMessageAcceptorChannel();
case STREAM:
return (T) new NioUdtByteAcceptorChannel();
default:
throw new IllegalStateException("wrong type=" + type);
}
case CONNECTOR:
switch (type) {
case DATAGRAM:
return (T) new NioUdtMessageConnectorChannel();
case STREAM:
return (T) new NioUdtByteConnectorChannel();
default:
throw new IllegalStateException("wrong type=" + type);
}
case RENDEZVOUS:
switch (type) {
case DATAGRAM:
return (T) new NioUdtMessageRendezvousChannel();
case STREAM:
return (T) new NioUdtByteRendezvousChannel();
default:
throw new IllegalStateException("wrong type=" + type);
}
default:
throw new IllegalStateException("wrong type=" + type);
}
case CONNECTOR:
switch (type) {
case DATAGRAM:
return (T) new NioUdtMessageConnectorChannel();
case STREAM:
return (T) new NioUdtByteConnectorChannel();
default:
throw new IllegalStateException("wrong type=" + type);
}
case RENDEZVOUS:
switch (type) {
case DATAGRAM:
return (T) new NioUdtMessageRendezvousChannel();
case STREAM:
return (T) new NioUdtByteRendezvousChannel();
default:
throw new IllegalStateException("wrong type=" + type);
}
default:
throw new IllegalStateException("wrong kind=" + kind);
throw new IllegalStateException("wrong kind=" + kind);
}
}
@ -239,5 +239,4 @@ public final class NioUdtProvider<T extends UdtChannel> implements ChannelFactor
public TypeUDT type() {
return type;
}
}

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

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

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

@ -192,6 +192,9 @@ public class ChannelHandlerAppender extends ChannelInboundHandlerAdapter {
} else {
name = e.name;
}
// Note that we do not use dctx.invoker() because it raises an IllegalStateExxception
// if the Channel is not registered yet.
pipeline.addAfter(dctx.invoker, oldName, name, e.handler);
}
} finally {

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

@ -101,7 +101,7 @@ import java.nio.channels.Channels;
* a = 1;
* }
*
* attr.set(a * (Integer) msg));
* attr.set(a * (Integer) msg);
* }
* }
*
@ -161,8 +161,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} was registered to its {@link EventLoop}.
*
* This will result in having the {@link ChannelInboundHandler#channelRegistered(ChannelHandlerContext)} method
* called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelInboundHandler#channelRegistered(ChannelHandlerContext)} method
* called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelRegistered();
@ -170,8 +170,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* 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
* 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
@ -180,8 +180,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} is active now, which means it is connected.
*
* This will result in having the {@link ChannelInboundHandler#channelActive(ChannelHandlerContext)} method
* called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelInboundHandler#channelActive(ChannelHandlerContext)} method
* called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelActive();
@ -189,8 +189,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} is inactive now, which means it is closed.
*
* This will result in having the {@link ChannelInboundHandler#channelInactive(ChannelHandlerContext)} method
* called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelInboundHandler#channelInactive(ChannelHandlerContext)} method
* called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelInactive();
@ -198,8 +198,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 ChannelInboundHandler#exceptionCaught(ChannelHandlerContext, Throwable)}
* method called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelInboundHandler#exceptionCaught(ChannelHandlerContext, Throwable)}
* method called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireExceptionCaught(Throwable cause);
@ -207,8 +207,8 @@ public interface ChannelHandlerContext extends AttributeMap {
/**
* A {@link Channel} received an user defined event.
*
* This will result in having the {@link ChannelInboundHandler#userEventTriggered(ChannelHandlerContext, Object)}
* method called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* This will result in having the {@link ChannelInboundHandler#userEventTriggered(ChannelHandlerContext, Object)}
* method called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireUserEventTriggered(Object event);
@ -217,7 +217,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* A {@link Channel} received a message.
*
* This will result in having the {@link ChannelInboundHandler#channelRead(ChannelHandlerContext, Object)}
* method called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelInboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext fireChannelRead(Object msg);
@ -240,7 +240,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#bind(ChannelHandlerContext, SocketAddress, ChannelPromise)} method
* called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture bind(SocketAddress localAddress);
@ -255,7 +255,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture connect(SocketAddress remoteAddress);
@ -267,7 +267,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress);
@ -278,7 +278,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#disconnect(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture disconnect();
@ -292,7 +292,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#close(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture close();
@ -304,7 +304,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* 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
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*
*/
@ -319,7 +319,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#bind(ChannelHandlerContext, SocketAddress, ChannelPromise)} method
* called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);
@ -337,7 +337,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise);
@ -351,7 +351,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, ChannelPromise)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
@ -364,7 +364,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#disconnect(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture disconnect(ChannelPromise promise);
@ -379,7 +379,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#close(ChannelHandlerContext, ChannelPromise)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelFuture close(ChannelPromise promise);
@ -393,7 +393,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* 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
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
@Deprecated
@ -408,7 +408,7 @@ public interface ChannelHandlerContext extends AttributeMap {
* <p>
* This will result in having the
* {@link ChannelOutboundHandler#read(ChannelHandlerContext)}
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* method called of the next {@link ChannelOutboundHandler} contained in the {@link ChannelPipeline} of the
* {@link Channel}.
*/
ChannelHandlerContext read();

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

@ -23,7 +23,7 @@ import java.net.SocketAddress;
/**
* Invokes the event handler methods of {@link ChannelInboundHandler} and {@link ChannelOutboundHandler}.
* A user can specify a {@link ChannelHandlerInvoker} to implement a custom thread model unsupported by the default
* implementation.
* implementation. Note that the methods in this interface are not intended to be called by a user.
*/
public interface ChannelHandlerInvoker {

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

@ -126,6 +126,11 @@ final class EmbeddedEventLoop extends AbstractEventLoop implements ChannelHandle
invokeChannelRegisteredNow(ctx);
}
@Override
public void invokeChannelUnregistered(ChannelHandlerContext ctx) {
invokeChannelUnregisteredNow(ctx);
}
@Override
public void invokeChannelActive(ChannelHandlerContext ctx) {
invokeChannelActiveNow(ctx);
@ -136,11 +141,6 @@ final class EmbeddedEventLoop extends AbstractEventLoop implements ChannelHandle
invokeChannelInactiveNow(ctx);
}
@Override
public void invokeChannelUnregistered(ChannelHandlerContext ctx) {
invokeChannelUnregisteredNow(ctx);
}
@Override
public void invokeExceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
invokeExceptionCaughtNow(ctx, cause);

1
transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java
View File

@ -143,6 +143,7 @@ public class DefaultDatagramChannelConfig extends DefaultChannelConfig implement
}
this.activeOnOpen = activeOnOpen;
}
@Override
public boolean isBroadcast() {
try {

2
transport/src/main/java/io/netty/channel/socket/nio/NioSocketChannel.java
View File

@ -67,7 +67,7 @@ public class NioSocketChannel extends AbstractNioByteChannel implements io.netty
* Create a new instance
*/
public NioSocketChannel() {
this(newSocket(DEFAULT_SELECTOR_PROVIDER));
this(DEFAULT_SELECTOR_PROVIDER);
}
/**

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

@ -39,8 +39,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.*;
import static org.junit.Assert.*;
public class SingleThreadEventLoopTest {

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

@ -90,7 +90,7 @@ public class LocalTransportThreadModelTest {
EventExecutorGroup e2 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e2"));
ThreadNameAuditor h1 = new ThreadNameAuditor();
ThreadNameAuditor h2 = new ThreadNameAuditor();
ThreadNameAuditor h3 = new ThreadNameAuditor();
ThreadNameAuditor h3 = new ThreadNameAuditor(true);
Channel ch = new LocalChannel();
// With no EventExecutor specified, h1 will be always invoked by EventLoop 'l'.
@ -361,6 +361,15 @@ public class LocalTransportThreadModelTest {
private final Queue<String> inboundThreadNames = new ConcurrentLinkedQueue<String>();
private final Queue<String> outboundThreadNames = new ConcurrentLinkedQueue<String>();
private final Queue<String> removalThreadNames = new ConcurrentLinkedQueue<String>();
private final boolean discard;
ThreadNameAuditor() {
this(false);
}
ThreadNameAuditor(boolean discard) {
this.discard = discard;
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
@ -370,7 +379,9 @@ public class LocalTransportThreadModelTest {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
inboundThreadNames.add(Thread.currentThread().getName());
ctx.fireChannelRead(msg);
if (!discard) {
ctx.fireChannelRead(msg);
}
}
@Override