Provide listener for send/stream methods, close AsyncHttpClient#613

Author: dguggemos

client/src/main/java/org/asynchttpclient/netty/ws/NettyWebSocket.java

@@ -17,6 +17,7 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.asynchttpclient.netty.util.ByteBufUtils.byteBuf2Bytes;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
@@ -37,6 +38,7 @@
 import org.asynchttpclient.ws.WebSocketPingListener;
 import org.asynchttpclient.ws.WebSocketPongListener;
 import org.asynchttpclient.ws.WebSocketTextListener;
+import org.asynchttpclient.ws.WebSocketWriteCompleteListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -81,42 +83,98 @@ public WebSocket sendMessage(byte[] message) {
         return this;
     }
 
+    @Override
+    public WebSocket sendMessage(byte[] message, WebSocketWriteCompleteListener listener) {
+        final ChannelPromise channelPromise = channel.newPromise();
+        channelPromise.addListener(listener);
+        channel.writeAndFlush(new BinaryWebSocketFrame(wrappedBuffer(message)), channelPromise);
+        return this;
+    }
+
     @Override
     public WebSocket stream(byte[] fragment, boolean last) {
         channel.writeAndFlush(new BinaryWebSocketFrame(last, 0, wrappedBuffer(fragment)), channel.voidPromise());
         return this;
     }
 
+    @Override
+    public WebSocket stream(final byte[] fragment, final boolean last, final WebSocketWriteCompleteListener listener) {
+        final ChannelPromise channelPromise = channel.newPromise();
+        channelPromise.addListener(listener);
+        channel.writeAndFlush(new BinaryWebSocketFrame(last, 0, wrappedBuffer(fragment)), channelPromise);
+        return this;
+    }
+
     @Override
     public WebSocket stream(byte[] fragment, int offset, int len, boolean last) {
         channel.writeAndFlush(new BinaryWebSocketFrame(last, 0, wrappedBuffer(fragment, offset, len)), channel.voidPromise());
         return this;
     }
 
+    @Override
+    public WebSocket stream(final byte[] fragment, final int offset, final int len, final boolean last, final WebSocketWriteCompleteListener listener) {
+        final ChannelPromise channelPromise = channel.newPromise();
+        channelPromise.addListener(listener);
+        channel.writeAndFlush(new BinaryWebSocketFrame(last, 0, wrappedBuffer(fragment, offset, len)), channelPromise);
+        return this;
+    }
+
     @Override
     public WebSocket sendMessage(String message) {
         channel.writeAndFlush(new TextWebSocketFrame(message), channel.voidPromise());
         return this;
     }
 
+    @Override
+    public WebSocket sendMessage(String message, WebSocketWriteCompleteListener listener) {
+        final ChannelPromise channelPromise = channel.newPromise();
+        channelPromise.addListener(listener);
+        channel.writeAndFlush(new TextWebSocketFrame(message), channelPromise);
+        return this;
+    }
+
     @Override
     public WebSocket stream(String fragment, boolean last) {
         channel.writeAndFlush(new TextWebSocketFrame(last, 0, fragment), channel.voidPromise());
         return this;
     }
 
+    @Override
+    public WebSocket stream(final String fragment, final boolean last, final WebSocketWriteCompleteListener listener) {
+        final ChannelPromise channelPromise = channel.newPromise();
+        channelPromise.addListener(listener);
+        channel.writeAndFlush(new TextWebSocketFrame(last, 0, fragment), channelPromise);
+        return this;
+    }
+
     @Override
     public WebSocket sendPing(byte[] payload) {
         channel.writeAndFlush(new PingWebSocketFrame(wrappedBuffer(payload)), channel.voidPromise());
         return this;
     }
 
+    @Override
+    public WebSocket sendPing(final byte[] payload, final WebSocketWriteCompleteListener listener) {
+        final ChannelPromise channelPromise = channel.newPromise();
+        channelPromise.addListener(listener);
+        channel.writeAndFlush(new PingWebSocketFrame(wrappedBuffer(payload)), channelPromise);
+        return this;
+    }
+
     @Override
     public WebSocket sendPong(byte[] payload) {
         channel.writeAndFlush(new PongWebSocketFrame(wrappedBuffer(payload)), channel.voidPromise());
         return this;
     }
 
+    @Override
+    public WebSocket sendPong(final byte[] payload, final WebSocketWriteCompleteListener listener) {
+        final ChannelPromise channelPromise = channel.newPromise();
+        channelPromise.addListener(listener);
+        channel.writeAndFlush(new PongWebSocketFrame(wrappedBuffer(payload)), channelPromise);
+        return this;
+    }
+
     @Override
     public boolean isOpen() {
         return channel.isOpen();

client/src/main/java/org/asynchttpclient/ws/WebSocket.java

@@ -12,11 +12,11 @@
  */
 package org.asynchttpclient.ws;
 
-import io.netty.handler.codec.http.HttpHeaders;
-
 import java.io.Closeable;
 import java.net.SocketAddress;
 
+import io.netty.handler.codec.http.HttpHeaders;
+
 /**
  * A WebSocket client
  */
@@ -49,6 +49,15 @@ public interface WebSocket extends Closeable {
      */
     WebSocket sendMessage(byte[] message);
 
+    /**
+     * Send a byte message.
+     *
+     * @param message a byte message
+     * @param listener is notified when a message was successfully processed by the channel or in case of failure
+     * @return this
+     */
+    WebSocket sendMessage(byte[] message, WebSocketWriteCompleteListener listener);
+
     /**
      * Allows streaming of multiple binary fragments.
      * 
@@ -59,6 +68,17 @@ public interface WebSocket extends Closeable {
      */
     WebSocket stream(byte[] fragment, boolean last);
 
+    /**
+     * Allows streaming of multiple binary fragments.
+     *
+     * @param fragment binary fragment.
+     * @param last flag indicating whether or not this is the last fragment.
+     * @param listener is notified when a fragment was successfully processed by the channel or in case of failure
+     *
+     * @return this
+     */
+    WebSocket stream(byte[] fragment, boolean last, WebSocketWriteCompleteListener listener);
+
     /**
      * Allows streaming of multiple binary fragments.
      * 
@@ -70,6 +90,18 @@ public interface WebSocket extends Closeable {
      */
     WebSocket stream(byte[] fragment, int offset, int len, boolean last);
 
+    /**
+     * Allows streaming of multiple binary fragments.
+     *
+     * @param fragment binary fragment.
+     * @param offset starting offset.
+     * @param len length.
+     * @param last flag indicating whether or not this is the last fragment.
+     * @param listener is notified when a fragment was successfully processed by the channel or in case of failure
+     * @return this
+     */
+    WebSocket stream(byte[] fragment, int offset, int len, boolean last, WebSocketWriteCompleteListener listener);
+
     /**
      * Send a text message
      * 
@@ -78,6 +110,15 @@ public interface WebSocket extends Closeable {
      */
     WebSocket sendMessage(String message);
 
+    /**
+     * Send a text message
+     *
+     * @param message a text message
+     * @param listener is notified when a message was successfully processed by the channel or in case of failure
+     * @return this
+     */
+    WebSocket sendMessage(String message, WebSocketWriteCompleteListener listener);
+
     /**
      * Allows streaming of multiple text fragments.
      * 
@@ -87,6 +128,16 @@ public interface WebSocket extends Closeable {
      */
     WebSocket stream(String fragment, boolean last);
 
+    /**
+     * Allows streaming of multiple text fragments.
+     *
+     * @param fragment text fragment.
+     * @param last flag indicating whether or not this is the last fragment.
+     * @param listener is notified when a fragment was successfully processed by the channel or in case of failure
+     * @return this
+     */
+    WebSocket stream(String fragment, boolean last, WebSocketWriteCompleteListener listener);
+
     /**
      * Send a <code>ping</code> with an optional payload (limited to 125 bytes or less).
      * 
@@ -97,12 +148,31 @@ public interface WebSocket extends Closeable {
 
     /**
      * Send a <code>ping</code> with an optional payload (limited to 125 bytes or less).
-     * 
+     *
+     * @param payload the ping payload.
+     * @param listener is notified when the ping was successfully processed by the channel or in case of failure
+     * @return this
+     */
+    WebSocket sendPing(byte[] payload, WebSocketWriteCompleteListener listener);
+
+    /**
+     * Send a <code>ping</code> with an optional payload (limited to 125 bytes or less).
+     *
      * @param payload the pong payload.
      * @return this
      */
     WebSocket sendPong(byte[] payload);
 
+    /**
+     * Send a <code>ping</code> with an optional payload (limited to 125 bytes or less).
+     *
+     * @param payload the pong payload.
+     * @param listener is notified when the pong was successfully processed by the channel or in case of failure
+
+     * @return this
+     */
+    WebSocket sendPong(byte[] payload, WebSocketWriteCompleteListener listener);
+
     /**
      * Add a {@link WebSocketListener}
      * 

client/src/main/java/org/asynchttpclient/ws/WebSocketWriteCompleteListener.java

@@ -0,0 +1,84 @@
+package org.asynchttpclient.ws;
+
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
+
+/**
+ * A listener for result of WebSocket write operations.
+ */
+public interface WebSocketWriteCompleteListener extends FutureListener<Void> {
+
+    /**
+     * Is called when a write operation completes, either successful or failing with an exception.
+     * @param result contains the result of the write operation
+     */
+    void onComplete(WriteCompleteResult result);
+
+    @Override
+    default void operationComplete(Future<Void> future) throws Exception {
+        if (future.isSuccess()) {
+            onComplete(WriteCompleteResult.SUCCEEDED);
+        } else {
+            onComplete(WriteCompleteResult.failed(future.cause()));
+        }
+    }
+
+    /**
+     * The result of a write operation.
+     */
+    interface WriteCompleteResult {
+
+        /**
+         * Constant for succeeded result.
+         */
+        WriteCompleteResult SUCCEEDED = new WriteCompleteResult() {
+            @Override public Throwable getFailure() {
+                return null;
+            }
+
+            @Override public boolean isSuccess() {
+                return true;
+            }
+
+            @Override public boolean isFailed() {
+                return false;
+            }
+        };
+
+        /**
+         * @param t the exception that caused the failure.
+         * @return a failed result
+         */
+        static WriteCompleteResult failed(Throwable t)
+        {
+            return new WriteCompleteResult() {
+                @Override public Throwable getFailure() {
+                    return t;
+                }
+
+                @Override public boolean isSuccess() {
+                    return false;
+                }
+
+                @Override public boolean isFailed() {
+                    return true;
+                }
+            };
+        }
+
+        /**
+         * @return the exception in case the write operation failed, @{@code null} otherwise.
+         */
+        Throwable getFailure();
+
+        /**
+         * @return @{@code true} if the operation succeeded, {@code false} otherwise.
+         */
+        boolean isSuccess();
+
+        /**
+         * @return @{@code true} if the operation failed, {@code false} otherwise.
+         */
+        boolean isFailed();
+    }
+}