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

Fix #153: Add ChannelFuture.rethrowIfFailed()

Browse Source
This commit is contained in:
Trustin Lee 2012-01-19 13:34:28 +09:00
parent fafeae7aa3
commit c95f9314f3
6 changed files with 110 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/org/jboss/netty/channel/ChannelFuture.java
View File

@ -261,6 +261,12 @@ public interface ChannelFuture {
*/
void removeListener(ChannelFutureListener listener);
/**
* Rethrows the exception that caused this future fail if this future is
* complete and failed.
*/
ChannelFuture rethrowIfFailed() throws Exception;
/**
* Waits for this future to be completed.
*

21
src/main/java/org/jboss/netty/channel/DefaultChannelFuture.java
View File

@ -168,6 +168,27 @@ public class DefaultChannelFuture implements ChannelFuture {
}
}
public ChannelFuture rethrowIfFailed() throws Exception {
if (!isDone()) {
return this;
}
Throwable cause = getCause();
if (cause == null) {
return this;
}
if (cause instanceof Exception) {
throw (Exception) cause;
}
if (cause instanceof Error) {
throw (Error) cause;
}
throw new RuntimeException(cause);
}
public ChannelFuture await() throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();

12
src/main/java/org/jboss/netty/channel/FailedChannelFuture.java
View File

@ -45,4 +45,16 @@ public class FailedChannelFuture extends CompleteChannelFuture {
public boolean isSuccess() {
return false;
}
public ChannelFuture rethrowIfFailed() throws Exception {
if (cause instanceof Exception) {
throw (Exception) cause;
}
if (cause instanceof Error) {
throw (Error) cause;
}
throw new RuntimeException(cause);
}
}

4
src/main/java/org/jboss/netty/channel/SucceededChannelFuture.java
View File

@ -38,4 +38,8 @@ public class SucceededChannelFuture extends CompleteChannelFuture {
public boolean isSuccess() {
return true;
}
public ChannelFuture rethrowIfFailed() throws Exception {
return this;
}
}

127
src/main/java/org/jboss/netty/example/http/websocketx/client/WebSocketClient.java
View File

@ -23,7 +23,6 @@ package org.jboss.netty.example.http.websocketx.client;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.Executors;
@ -39,11 +38,9 @@ import org.jboss.netty.handler.codec.http.HttpRequestEncoder;
import org.jboss.netty.handler.codec.http.HttpResponseDecoder;
import org.jboss.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketFrame;
import org.jboss.netty.handler.codec.http.websocketx.WebSocketVersion;
public class WebSocketClient {
@ -61,70 +58,72 @@ public class WebSocketClient {
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
String protocol = uri.getScheme();
if (!protocol.equals("ws")) {
throw new IllegalArgumentException("Unsupported protocol: " + protocol);
}
HashMap<String, String> customHeaders = new HashMap<String, String>();
customHeaders.put("MyHeader", "MyValue");
// Connect with V13 (RFC 6455). You can change it to V08 or V00.
// If you change it to V00, ping is not supported and remember to change
// HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
final WebSocketClientHandshaker handshaker =
new WebSocketClientHandshakerFactory().newHandshaker(
uri, WebSocketVersion.V13, null, false, customHeaders);
Channel ch = null;
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline pipeline = Channels.pipeline();
// If you wish to support HyBi V00, you need to use
// WebSocketHttpResponseDecoder instead for
// HttpResponseDecoder.
pipeline.addLast("decoder", new HttpResponseDecoder());
pipeline.addLast("encoder", new HttpRequestEncoder());
pipeline.addLast("ws-handler", new WebSocketClientHandler(handshaker));
return pipeline;
try {
String protocol = uri.getScheme();
if (!protocol.equals("ws")) {
throw new IllegalArgumentException("Unsupported protocol: " + protocol);
}
});
// Connect
System.out.println("WebSocket Client connecting");
ChannelFuture future =
bootstrap.connect(
new InetSocketAddress(uri.getHost(), uri.getPort()));
future.awaitUninterruptibly();
Channel ch = future.getChannel();
handshaker.handshake(ch);
Thread.sleep(1000);
// Send 10 messages and wait for responses
System.out.println("WebSocket Client sending message");
for (int i = 0; i < 10; i++) {
ch.write(new TextWebSocketFrame("Message #" + i));
HashMap<String, String> customHeaders = new HashMap<String, String>();
customHeaders.put("MyHeader", "MyValue");
// Connect with V13 (RFC 6455). You can change it to V08 or V00.
// If you change it to V00, ping is not supported and remember to change
// HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
final WebSocketClientHandshaker handshaker =
new WebSocketClientHandshakerFactory().newHandshaker(
uri, WebSocketVersion.V13, null, false, customHeaders);
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() throws Exception {
ChannelPipeline pipeline = Channels.pipeline();
// If you wish to support HyBi V00, you need to use
// WebSocketHttpResponseDecoder instead for
// HttpResponseDecoder.
pipeline.addLast("decoder", new HttpResponseDecoder());
pipeline.addLast("encoder", new HttpRequestEncoder());
pipeline.addLast("ws-handler", new WebSocketClientHandler(handshaker));
return pipeline;
}
});
// Connect
System.out.println("WebSocket Client connecting");
ChannelFuture future =
bootstrap.connect(
new InetSocketAddress(uri.getHost(), uri.getPort()));
future.awaitUninterruptibly().rethrowIfFailed();
ch = future.getChannel();
handshaker.handshake(ch).awaitUninterruptibly().rethrowIfFailed();
// Send 10 messages and wait for responses
System.out.println("WebSocket Client sending message");
for (int i = 0; i < 10; i++) {
ch.write(new TextWebSocketFrame("Message #" + i));
}
// Ping
System.out.println("WebSocket Client sending ping");
ch.write(new PingWebSocketFrame(ChannelBuffers.copiedBuffer(new byte[]{1, 2, 3, 4, 5, 6})));
// Close
System.out.println("WebSocket Client sending close");
ch.write(new CloseWebSocketFrame());
// WebSocketClientHandler will close the connection when the server
// responds to the CloseWebSocketFrame.
ch.getCloseFuture().awaitUninterruptibly();
} finally {
if (ch != null) {
ch.close();
}
bootstrap.releaseExternalResources();
}
Thread.sleep(1000);
// Ping
System.out.println("WebSocket Client sending ping");
ch.write(new PingWebSocketFrame(ChannelBuffers.copiedBuffer(new byte[] { 1, 2, 3, 4, 5, 6 })));
Thread.sleep(1000);
// Close
System.out.println("WebSocket Client sending close");
ch.write(new CloseWebSocketFrame());
// WebSocketClientHandler will close the connection when the server
// responds to the CloseWebSocketFrame.
ch.getCloseFuture().awaitUninterruptibly();
bootstrap.releaseExternalResources();
}
public static void main(String[] args) throws Exception {

4
src/test/java/org/jboss/netty/channel/CompleteChannelFutureTest.java
View File

@ -96,6 +96,10 @@ public class CompleteChannelFutureTest {
public boolean isSuccess() {
throw new Error();
}
public ChannelFuture rethrowIfFailed() throws Exception {
throw new Error();
}
}
private static class ExpectedError extends Error {