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

Clean up echo example / Fix a bug where closeFuture is not notified

Browse Source
This commit is contained in:
Trustin Lee 2012-05-14 14:17:40 +09:00
parent 6a0040a14e
commit 05f955ee10
5 changed files with 61 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
example/src/main/java/io/netty/example/echo/EchoClient.java
View File

@ -15,12 +15,6 @@
*/ */
package io.netty.example.echo; package io.netty.example.echo;
import io.netty.buffer.ChannelBuffer;
import io.netty.buffer.ChannelBuffers;
import io.netty.channel.ChannelBufferHolder;
import io.netty.channel.ChannelBufferHolders;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInboundHandlerContext;
import io.netty.channel.EventLoop; import io.netty.channel.EventLoop;
import io.netty.channel.MultithreadEventLoop; import io.netty.channel.MultithreadEventLoop;
import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.SocketChannel;
@ -30,7 +24,6 @@ import io.netty.handler.logging.LoggingHandler;
import io.netty.logging.InternalLogLevel; import io.netty.logging.InternalLogLevel;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicLong;
/** /**
* Sends one message when a connection is open and echoes back any received * Sends one message when a connection is open and echoes back any received
@ -43,7 +36,6 @@ public class EchoClient {
private final String host; private final String host;
private final int port; private final int port;
private final int firstMessageSize; private final int firstMessageSize;
private final AtomicLong transferredBytes = new AtomicLong();
public EchoClient(String host, int port, int firstMessageSize) { public EchoClient(String host, int port, int firstMessageSize) {
this.host = host; this.host = host;
@ -52,53 +44,23 @@ public class EchoClient {
} }
public void run() throws Exception { public void run() throws Exception {
EventLoop loop = new MultithreadEventLoop(SelectorEventLoop.FACTORY); // Create a new socket and configure it.
SocketChannel s = new NioSocketChannel(); SocketChannel s = new NioSocketChannel();
s.config().setTcpNoDelay(true); s.config().setTcpNoDelay(true);
s.pipeline().addLast("logger", new LoggingHandler(InternalLogLevel.INFO)); s.pipeline().addLast("logger", new LoggingHandler(InternalLogLevel.INFO));
s.pipeline().addLast("echoer", new ChannelInboundHandlerAdapter<Byte>() { s.pipeline().addLast("echoer", new EchoClientHandler(firstMessageSize));
private final ChannelBuffer firstMessage; // Begin the communication by registering the channel to an event loop and connecting
{ // to the peer.
if (firstMessageSize <= 0) { EventLoop loop = new MultithreadEventLoop(SelectorEventLoop.FACTORY);
throw new IllegalArgumentException(
"firstMessageSize: " + firstMessageSize);
}
firstMessage = ChannelBuffers.buffer(firstMessageSize);
for (int i = 0; i < firstMessage.capacity(); i ++) {
firstMessage.writeByte((byte) i);
}
}
@Override
public ChannelBufferHolder<Byte> newInboundBuffer(ChannelInboundHandlerContext<Byte> ctx) {
return ChannelBufferHolders.byteBuffer(ChannelBuffers.dynamicBuffer());
}
@Override
public void channelActive(ChannelInboundHandlerContext<Byte> ctx)
throws Exception {
ctx.write(firstMessage);
}
@Override
public void inboundBufferUpdated(
ChannelInboundHandlerContext<Byte> ctx) throws Exception {
ChannelBuffer in = ctx.in().byteBuffer();
ChannelBuffer out = ctx.out().byteBuffer();
transferredBytes.addAndGet(in.readableBytes());
out.discardReadBytes();
out.writeBytes(in);
in.clear();
ctx.flush();
}
});
loop.register(s).awaitUninterruptibly().rethrowIfFailed(); loop.register(s).awaitUninterruptibly().rethrowIfFailed();
s.connect(new InetSocketAddress(host, port)).awaitUninterruptibly().rethrowIfFailed(); s.connect(new InetSocketAddress(host, port));
// FIXME: Wait until the connection is closed or the connection attempt fails. // Wait until the connection is closed.
// FIXME: Show how to shut down. s.closeFuture().awaitUninterruptibly();
// Terminate the event loop.
loop.shutdown();
} }
public static void main(String[] args) throws Exception { public static void main(String[] args) throws Exception {

55
example/src/main/java/io/netty/example/echo/EchoClientHandler.java
View File

@ -15,38 +15,34 @@
*/ */
package io.netty.example.echo; package io.netty.example.echo;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;
import io.netty.buffer.ChannelBuffer; import io.netty.buffer.ChannelBuffer;
import io.netty.buffer.ChannelBuffers; import io.netty.buffer.ChannelBuffers;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelBufferHolder;
import io.netty.channel.ChannelStateEvent; import io.netty.channel.ChannelBufferHolders;
import io.netty.channel.ExceptionEvent; import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.MessageEvent; import io.netty.channel.ChannelInboundHandlerContext;
import io.netty.channel.SimpleChannelUpstreamHandler;
import java.util.logging.Level;
import java.util.logging.Logger;
/** /**
* Handler implementation for the echo client. It initiates the ping-pong * Handler implementation for the echo client. It initiates the ping-pong
* traffic between the echo client and server by sending the first message to * traffic between the echo client and server by sending the first message to
* the server. * the server.
*/ */
public class EchoClientHandler extends SimpleChannelUpstreamHandler { public class EchoClientHandler extends ChannelInboundHandlerAdapter<Byte> {
private static final Logger logger = Logger.getLogger( private static final Logger logger = Logger.getLogger(
EchoClientHandler.class.getName()); EchoClientHandler.class.getName());
private final ChannelBuffer firstMessage; private final ChannelBuffer firstMessage;
private final AtomicLong transferredBytes = new AtomicLong();
/** /**
* Creates a client-side handler. * Creates a client-side handler.
*/ */
public EchoClientHandler(int firstMessageSize) { public EchoClientHandler(int firstMessageSize) {
if (firstMessageSize <= 0) { if (firstMessageSize <= 0) {
throw new IllegalArgumentException( throw new IllegalArgumentException("firstMessageSize: " + firstMessageSize);
"firstMessageSize: " + firstMessageSize);
} }
firstMessage = ChannelBuffers.buffer(firstMessageSize); firstMessage = ChannelBuffers.buffer(firstMessageSize);
for (int i = 0; i < firstMessage.capacity(); i ++) { for (int i = 0; i < firstMessage.capacity(); i ++) {
@ -54,34 +50,31 @@ public class EchoClientHandler extends SimpleChannelUpstreamHandler {
} }
} }
public long getTransferredBytes() { @Override
return transferredBytes.get(); public ChannelBufferHolder<Byte> newInboundBuffer(ChannelInboundHandlerContext<Byte> ctx) {
return ChannelBufferHolders.byteBuffer(ChannelBuffers.dynamicBuffer());
} }
@Override @Override
public void channelConnected( public void channelActive(ChannelInboundHandlerContext<Byte> ctx) {
ChannelHandlerContext ctx, ChannelStateEvent e) { ctx.write(firstMessage);
// Send the first message. Server will not send anything here
// because the firstMessage's capacity is 0.
e.channel().write(firstMessage);
} }
@Override @Override
public void messageReceived( public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) {
ChannelHandlerContext ctx, MessageEvent e) { ChannelBuffer in = ctx.in().byteBuffer();
// Send back the received message to the remote peer. ChannelBuffer out = ctx.out().byteBuffer();
transferredBytes.addAndGet(((ChannelBuffer) e.getMessage()).readableBytes()); out.discardReadBytes();
e.channel().write(e.getMessage()); out.writeBytes(in);
in.discardReadBytes();
ctx.flush();
} }
@Override @Override
public void exceptionCaught( public void exceptionCaught(
ChannelHandlerContext ctx, ExceptionEvent e) { ChannelInboundHandlerContext<Byte> ctx, Throwable cause) {
// Close the connection when an exception is raised. // Close the connection when an exception is raised.
logger.log( logger.log(Level.WARNING, "Unexpected exception from downstream.", cause);
Level.WARNING, ctx.close();
"Unexpected exception from downstream.",
e.cause());
e.channel().close();
} }
} }

23
example/src/main/java/io/netty/example/echo/EchoServer.java
View File

@ -15,8 +15,6 @@
*/ */
package io.netty.example.echo; package io.netty.example.echo;
import io.netty.buffer.ChannelBuffer;
import io.netty.buffer.ChannelBuffers;
import io.netty.channel.ChannelBufferHolder; import io.netty.channel.ChannelBufferHolder;
import io.netty.channel.ChannelBufferHolders; import io.netty.channel.ChannelBufferHolders;
import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInboundHandlerAdapter;
@ -33,7 +31,6 @@ import io.netty.logging.InternalLogLevel;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.util.ArrayDeque; import java.util.ArrayDeque;
import java.util.Queue; import java.util.Queue;
import java.util.concurrent.atomic.AtomicLong;
/** /**
* Echoes back any received data from a client. * Echoes back any received data from a client.
@ -41,7 +38,6 @@ import java.util.concurrent.atomic.AtomicLong;
public class EchoServer { public class EchoServer {
private final int port; private final int port;
private final AtomicLong transferredBytes = new AtomicLong();
public EchoServer(int port) { public EchoServer(int port) {
this.port = port; this.port = port;
@ -73,24 +69,7 @@ public class EchoServer {
} }
s.config().setTcpNoDelay(true); s.config().setTcpNoDelay(true);
s.pipeline().addLast("logger", new LoggingHandler(InternalLogLevel.INFO)); s.pipeline().addLast("logger", new LoggingHandler(InternalLogLevel.INFO));
s.pipeline().addLast("echoer", new ChannelInboundHandlerAdapter<Byte>() { s.pipeline().addLast("echoer", new EchoServerHandler());
@Override
public ChannelBufferHolder<Byte> newInboundBuffer(ChannelInboundHandlerContext<Byte> ctx) {
return ChannelBufferHolders.byteBuffer(ChannelBuffers.dynamicBuffer());
}
@Override
public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) {
ChannelBuffer in = ctx.in().byteBuffer();
ChannelBuffer out = ctx.out().byteBuffer();
transferredBytes.addAndGet(in.readableBytes());
out.discardReadBytes();
out.writeBytes(in);
in.clear();
ctx.flush();
}
});
loop.register(s); loop.register(s);
} }
} }

45
example/src/main/java/io/netty/example/echo/EchoServerHandler.java
View File

@ -15,46 +15,43 @@
*/ */
package io.netty.example.echo; package io.netty.example.echo;
import java.util.concurrent.atomic.AtomicLong; import io.netty.buffer.ChannelBuffer;
import io.netty.buffer.ChannelBuffers;
import io.netty.channel.ChannelBufferHolder;
import io.netty.channel.ChannelBufferHolders;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInboundHandlerContext;
import java.util.logging.Level; import java.util.logging.Level;
import java.util.logging.Logger; import java.util.logging.Logger;
import io.netty.buffer.ChannelBuffer;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ExceptionEvent;
import io.netty.channel.MessageEvent;
import io.netty.channel.SimpleChannelUpstreamHandler;
/** /**
* Handler implementation for the echo server. * Handler implementation for the echo server.
*/ */
public class EchoServerHandler extends SimpleChannelUpstreamHandler { public class EchoServerHandler extends ChannelInboundHandlerAdapter<Byte> {
private static final Logger logger = Logger.getLogger( private static final Logger logger = Logger.getLogger(
EchoServerHandler.class.getName()); EchoServerHandler.class.getName());
private final AtomicLong transferredBytes = new AtomicLong(); @Override
public ChannelBufferHolder<Byte> newInboundBuffer(ChannelInboundHandlerContext<Byte> ctx) {
public long getTransferredBytes() { return ChannelBufferHolders.byteBuffer(ChannelBuffers.dynamicBuffer());
return transferredBytes.get();
} }
@Override @Override
public void messageReceived( public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) {
ChannelHandlerContext ctx, MessageEvent e) { ChannelBuffer in = ctx.in().byteBuffer();
// Send back the received message to the remote peer. ChannelBuffer out = ctx.out().byteBuffer();
transferredBytes.addAndGet(((ChannelBuffer) e.getMessage()).readableBytes()); out.discardReadBytes();
e.channel().write(e.getMessage()); out.writeBytes(in);
in.discardReadBytes();
ctx.flush();
} }
@Override @Override
public void exceptionCaught( public void exceptionCaught(ChannelInboundHandlerContext<Byte> ctx, Throwable cause) {
ChannelHandlerContext ctx, ExceptionEvent e) {
// Close the connection when an exception is raised. // Close the connection when an exception is raised.
logger.log( logger.log(Level.WARNING, "Unexpected exception from downstream.", cause);
Level.WARNING, ctx.close();
"Unexpected exception from downstream.",
e.cause());
e.channel().close();
} }
} }

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

@ -400,6 +400,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
future.setFailure(t); future.setFailure(t);
pipeline().fireExceptionCaught(t); pipeline().fireExceptionCaught(t);
closeFuture().setSuccess();
} }
} }
@ -538,7 +539,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
@Override @Override
public void close(final ChannelFuture future) { public void close(final ChannelFuture future) {
if (eventLoop().inEventLoop()) { if (eventLoop().inEventLoop()) {
if (isOpen()) { if (closeFuture.setClosed()) {
boolean wasActive = isActive(); boolean wasActive = isActive();
try { try {
doClose(); doClose();
@ -557,7 +558,6 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
pipeline().fireChannelInactive(); pipeline().fireChannelInactive();
} }
closeFuture.setClosed();
deregister(voidFuture()); deregister(voidFuture());
} else { } else {
// Closed already. // Closed already.
@ -791,9 +791,10 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
throw new IllegalStateException(); throw new IllegalStateException();
} }
void setClosed() { boolean setClosed() {
boolean set = super.setSuccess(); boolean set = super.setSuccess();
assert set; assert set;
return set;
} }
} }