use Promise instead of CallbackNotifier

stephenyik · Aug 14, 2013 · 65fd9c0 · 65fd9c0
1 parent 48eb73f
commit 65fd9c0
Show file tree

Hide file tree

Showing 4 changed files with 49 additions and 53 deletions.
diff --git a/example/src/main/java/io/netty/example/socksproxy/CallbackNotifier.java b/example/src/main/java/io/netty/example/socksproxy/CallbackNotifier.java
diff --git a/example/src/main/java/io/netty/example/socksproxy/DirectClientHandler.java b/example/src/main/java/io/netty/example/socksproxy/DirectClientHandler.java
@@ -17,6 +17,7 @@
 
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.util.concurrent.Promise;
 
 
 public final class DirectClientHandler extends ChannelInboundHandlerAdapter {
@@ -25,20 +26,20 @@ public final class DirectClientHandler extends ChannelInboundHandlerAdapter {
     public static String getName() {
         return name;
     }
-    private final CallbackNotifier cb;
+    private final Promise promise;
 
-    public DirectClientHandler(CallbackNotifier cb) {
-        this.cb = cb;
+    public DirectClientHandler(Promise promise) {
+        this.promise = promise;
     }
 
     @Override
     public void channelActive(ChannelHandlerContext ctx) {
         ctx.pipeline().remove(this);
-        cb.onSuccess(ctx);
+        promise.setSuccess(ctx.channel());
     }
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable throwable) throws Exception {
-        cb.onFailure(ctx, throwable);
+        promise.setFailure(throwable);
     }
 }
diff --git a/example/src/main/java/io/netty/example/socksproxy/DirectClientInitializer.java b/example/src/main/java/io/netty/example/socksproxy/DirectClientInitializer.java
@@ -18,19 +18,20 @@
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.util.concurrent.Promise;
 
 
 public final class DirectClientInitializer extends ChannelInitializer<SocketChannel> {
 
-    private final CallbackNotifier callbackNotifier;
+    private final Promise promise;
 
-    public DirectClientInitializer(CallbackNotifier callbackNotifier) {
-        this.callbackNotifier = callbackNotifier;
+    public DirectClientInitializer(Promise promise) {
+        this.promise = promise;
     }
 
     @Override
     public void initChannel(SocketChannel socketChannel) throws Exception {
         ChannelPipeline channelPipeline = socketChannel.pipeline();
-        channelPipeline.addLast(DirectClientHandler.getName(), new DirectClientHandler(callbackNotifier));
+        channelPipeline.addLast(DirectClientHandler.getName(), new DirectClientHandler(promise));
     }
 }
diff --git a/example/src/main/java/io/netty/example/socksproxy/SocksServerConnectHandler.java b/example/src/main/java/io/netty/example/socksproxy/SocksServerConnectHandler.java
@@ -27,6 +27,9 @@
 import io.netty.handler.codec.socks.SocksCmdRequest;
 import io.netty.handler.codec.socks.SocksCmdResponse;
 import io.netty.handler.codec.socks.SocksCmdStatus;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
+import io.netty.util.concurrent.Promise;
 
 @ChannelHandler.Sharable
 public final class SocksServerConnectHandler extends SimpleChannelInboundHandler<SocksCmdRequest> {
@@ -40,35 +43,49 @@ public static String getName() {
 
     @Override
     public void channelRead0(final ChannelHandlerContext ctx, final SocksCmdRequest request) throws Exception {
-        CallbackNotifier cb = new CallbackNotifier() {
+        Promise promise = ctx.executor().newPromise();
+        promise.addListener(
+            new GenericFutureListener<Future<Channel>>() {
             @Override
-            public void onSuccess(final ChannelHandlerContext outboundCtx) {
-                ctx.channel().writeAndFlush(new SocksCmdResponse(SocksCmdStatus.SUCCESS, request.addressType()))
-                             .addListener(new ChannelFutureListener() {
-                    @Override
-                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
-                        ctx.pipeline().remove(getName());
-                        outboundCtx.channel().pipeline().addLast(new RelayHandler(ctx.channel()));
-                        ctx.channel().pipeline().addLast(new RelayHandler(outboundCtx.channel()));
-                    }
-                });
-            }
-
-            @Override
-            public void onFailure(ChannelHandlerContext outboundCtx, Throwable cause) {
-                ctx.channel().writeAndFlush(new SocksCmdResponse(SocksCmdStatus.FAILURE, request.addressType()));
-                SocksServerUtils.closeOnFlush(ctx.channel());
+            public void operationComplete(final Future<Channel> future) throws Exception {
+                final Channel outboundChannel = future.getNow();
+                if (future.isSuccess()) {
+                    ctx.channel().writeAndFlush(new SocksCmdResponse(SocksCmdStatus.SUCCESS, request.addressType()))
+                            .addListener(new ChannelFutureListener() {
+                                @Override
+                                public void operationComplete(ChannelFuture channelFuture) throws Exception {
+                                    ctx.pipeline().remove(getName());
+                                    outboundChannel.pipeline().addLast(new RelayHandler(ctx.channel()));
+                                    ctx.channel().pipeline().addLast(new RelayHandler(outboundChannel));
+                                }
+                            });
+                } else {
+                    ctx.channel().writeAndFlush(new SocksCmdResponse(SocksCmdStatus.FAILURE, request.addressType()));
+                    SocksServerUtils.closeOnFlush(ctx.channel());
+                }
             }
-        };
+        });
 
         final Channel inboundChannel = ctx.channel();
         b.group(inboundChannel.eventLoop())
                 .channel(NioSocketChannel.class)
                 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
                 .option(ChannelOption.SO_KEEPALIVE, true)
-                .handler(new DirectClientInitializer(cb));
-
-        b.connect(request.host(), request.port());
+                .handler(new DirectClientInitializer(promise));
+        b.connect(request.host(), request.port())
+          .addListener(new ChannelFutureListener() {
+                @Override
+                public void operationComplete(ChannelFuture future) throws Exception {
+                    if (future.isSuccess()) {
+                        // Connection established use handler provided results
+                    } else {
+                        // Close the connection if the connection attempt has failed.
+                        ctx.channel().writeAndFlush(
+                                new SocksCmdResponse(SocksCmdStatus.FAILURE, request.addressType()));
+                        SocksServerUtils.closeOnFlush(ctx.channel());
+                    }
+                }
+           });
     }
 
     @Override