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

Fixed incorrect or missing event order in Bootstraps and NIO UDP transport

Browse Source
This commit is contained in:
Trustin Lee 2009-06-12 01:43:26 +00:00
parent 0a5019385c
commit d42ea03799
5 changed files with 93 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/main/java/org/jboss/netty/bootstrap/ClientBootstrap.java
View File

@ -273,10 +273,13 @@ public class ClientBootstrap extends Bootstrap {
public void channelOpen(
ChannelHandlerContext context,
ChannelStateEvent event) {
context.sendUpstream(event);
// Apply options.
event.getChannel().getConfig().setOptions(bootstrap.getOptions());
try {
// Apply options.
event.getChannel().getConfig().setOptions(bootstrap.getOptions());
} finally {
context.sendUpstream(event);
}
// Bind or connect.
if (localAddress != null) {

24
src/main/java/org/jboss/netty/bootstrap/ConnectionlessBootstrap.java
View File

@ -339,29 +339,33 @@ public class ConnectionlessBootstrap extends Bootstrap {
@ChannelPipelineCoverage("one")
private final class ConnectionlessBinder extends SimpleChannelUpstreamHandler {
private final SocketAddress localAddress;
private final BlockingQueue<ChannelFuture> futureQueue;
ConnectionlessBinder(SocketAddress localAddress, BlockingQueue<ChannelFuture> futureQueue) {
this.localAddress = localAddress;
this.futureQueue = futureQueue;
}
@Override
public void channelOpen(
ChannelHandlerContext ctx,
ChannelStateEvent evt) {
evt.getChannel().getConfig().setPipelineFactory(getPipelineFactory());
// Apply options.
evt.getChannel().getConfig().setOptions(getOptions());
try {
evt.getChannel().getConfig().setPipelineFactory(getPipelineFactory());
// Apply options.
evt.getChannel().getConfig().setOptions(getOptions());
} finally {
ctx.sendUpstream(evt);
}
boolean finished = futureQueue.offer(evt.getChannel().bind(localAddress));
assert finished;
ctx.sendUpstream(evt);
}
@Override
public void exceptionCaught(
ChannelHandlerContext ctx, ExceptionEvent e)

34
src/main/java/org/jboss/netty/bootstrap/ServerBootstrap.java
View File

@ -311,27 +311,31 @@ public class ServerBootstrap extends Bootstrap {
public void channelOpen(
ChannelHandlerContext ctx,
ChannelStateEvent evt) {
evt.getChannel().getConfig().setPipelineFactory(getPipelineFactory());
// Split options into two categories: parent and child.
Map<String, Object> allOptions = getOptions();
Map<String, Object> parentOptions = new HashMap<String, Object>();
for (Entry<String, Object> e: allOptions.entrySet()) {
if (e.getKey().startsWith("child.")) {
childOptions.put(
e.getKey().substring(6),
e.getValue());
} else if (!e.getKey().equals("pipelineFactory")) {
parentOptions.put(e.getKey(), e.getValue());
try {
evt.getChannel().getConfig().setPipelineFactory(getPipelineFactory());
// Split options into two categories: parent and child.
Map<String, Object> allOptions = getOptions();
Map<String, Object> parentOptions = new HashMap<String, Object>();
for (Entry<String, Object> e: allOptions.entrySet()) {
if (e.getKey().startsWith("child.")) {
childOptions.put(
e.getKey().substring(6),
e.getValue());
} else if (!e.getKey().equals("pipelineFactory")) {
parentOptions.put(e.getKey(), e.getValue());
}
}
}
// Apply parent options.
evt.getChannel().getConfig().setOptions(parentOptions);
// Apply parent options.
evt.getChannel().getConfig().setOptions(parentOptions);
} finally {
ctx.sendUpstream(evt);
}
boolean finished = futureQueue.offer(evt.getChannel().bind(localAddress));
assert finished;
ctx.sendUpstream(evt);
}
@Override

43
src/main/java/org/jboss/netty/channel/socket/nio/NioDatagramPipelineSink.java
View File

@ -25,12 +25,14 @@ package org.jboss.netty.channel.socket.nio;
import static org.jboss.netty.channel.Channels.*;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import org.jboss.netty.channel.AbstractChannelSink;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelState;
import org.jboss.netty.channel.ChannelStateEvent;
@ -100,11 +102,10 @@ class NioDatagramPipelineSink extends AbstractChannelSink {
}
break;
case CONNECTED:
// TODO Implement me
if (value != null) {
//connect(channel, future, (SocketAddress) value);
connect(channel, future, (InetSocketAddress) value);
} else {
//NioUdpWorker.disconnect(channel, future);
NioUdpWorker.disconnect(channel, future);
}
break;
case INTEREST_OPS:
@ -164,6 +165,42 @@ class NioDatagramPipelineSink extends AbstractChannelSink {
}
}
private void connect(
NioDatagramChannel channel, ChannelFuture future,
SocketAddress remoteAddress) {
boolean bound = channel.isBound();
boolean connected = false;
boolean workerStarted = false;
future.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
try {
channel.getDatagramChannel().connect(remoteAddress);
connected = true;
// Fire events.
future.setSuccess();
if (!bound) {
fireChannelBound(channel, channel.getLocalAddress());
}
fireChannelConnected(channel, channel.getRemoteAddress());
if (!bound) {
channel.worker.register(channel, future);
}
workerStarted = true;
} catch (Throwable t) {
future.setFailure(t);
fireExceptionCaught(channel, t);
} finally {
if (connected && !workerStarted) {
NioUdpWorker.close(channel, future);
}
}
}
NioUdpWorker nextWorker() {
return workers[Math.abs(workerIndex.getAndIncrement() % workers.length)];
}

16
src/main/java/org/jboss/netty/channel/socket/nio/NioUdpWorker.java
View File

@ -618,6 +618,20 @@ class NioUdpWorker implements Runnable {
}
}
static void disconnect(NioDatagramChannel channel, ChannelFuture future) {
boolean connected = channel.isConnected();
try {
channel.getDatagramChannel().disconnect();
future.setSuccess();
if (connected) {
fireChannelDisconnected(channel);
}
} catch (Throwable t) {
future.setFailure(t);
fireExceptionCaught(channel, t);
}
}
static void close(final NioDatagramChannel channel,
final ChannelFuture future) {
NioUdpWorker worker = channel.worker;
@ -820,8 +834,6 @@ class NioUdpWorker implements Runnable {
throw new ChannelException(
"Failed to register a socket to the selector.", e);
}
// XXX: Perhaps channelBind?
fireChannelConnected(channel, localAddress);
}
}
}