WebSockets: #isOpen fix #574

* synchronized ws access

Author: jknack

coverage-report/src/test/java/org/jooby/issues/Issue636.java

@@ -0,0 +1,63 @@
+package org.jooby.issues;
+
+import java.nio.channels.ClosedChannelException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.jooby.Err;
+import org.jooby.test.ServerFeature;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.ning.http.client.AsyncHttpClient;
+import com.ning.http.client.AsyncHttpClientConfig;
+import com.ning.http.client.ws.WebSocket;
+import com.ning.http.client.ws.WebSocketUpgradeHandler;
+
+public class Issue636 extends ServerFeature {
+
+  private static CountDownLatch closeLatch;
+
+  {
+    ws("/636", ws -> {
+
+      ws.onClose(status -> {
+        try {
+          ws.send("636");
+        } catch (Err x) {
+          closeLatch.countDown();
+        }
+      });
+    });
+
+  }
+
+  private AsyncHttpClient client;
+
+  @Before
+  public void before() {
+    client = new AsyncHttpClient(new AsyncHttpClientConfig.Builder().build());
+  }
+
+  @After
+  public void after() {
+    client.close();
+  }
+
+  @Test
+  public void sendClose() throws Exception {
+    closeLatch = new CountDownLatch(1);
+
+    WebSocket ws = client.prepareGet(ws("636").toString())
+        .execute(new WebSocketUpgradeHandler.Builder().build())
+        .get();
+
+    ws.sendMessage("foo");
+    Thread.sleep(100L);
+    ws.close();
+    Thread.sleep(100L);
+    closeLatch.await(1L, TimeUnit.SECONDS);
+  }
+
+}

jooby-netty/src/main/java/org/jooby/internal/netty/NettyHandler.java

@@ -21,8 +21,7 @@
 import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
 import static java.util.Objects.requireNonNull;
 
-import java.io.IOException;
-
+import org.jooby.internal.ConnectionResetByPeer;
 import org.jooby.spi.HttpHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -114,7 +113,7 @@ public void channelReadComplete(final ChannelHandlerContext ctx) throws Exceptio
   @Override
   public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
     try {
-      if (connectionResetByPeer(cause)) {
+      if (ConnectionResetByPeer.test(cause)) {
         log.trace("execution of: " + ctx.channel().attr(PATH).get() + " resulted in error", cause);
       } else {
         Attribute<NettyWebSocket> ws = ctx.channel().attr(NettyWebSocket.KEY);
@@ -131,11 +130,6 @@ public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cau
 
   }
 
-  private boolean connectionResetByPeer(final Throwable cause) {
-    return cause instanceof IOException
-        && cause.getMessage().toLowerCase().contains("connection reset by peer");
-  }
-
   @Override
   public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt)
       throws Exception {

jooby/src/main/java/org/jooby/Err.java

@@ -144,7 +144,19 @@ public Err(final Status status, final String message, final Throwable cause) {
    * @param cause The cause of the problem.
    */
   public Err(final int status, final String message, final Throwable cause) {
-    this(Status.valueOf(status), message, cause);
+    super(message("", status, message), cause);
+    this.status = status;
+  }
+
+  /**
+   * Creates a new {@link Err}.
+   *
+   * @param status A web socket close status. Required.
+   * @param message Close message.
+   */
+  public Err(final WebSocket.CloseStatus status, String message) {
+    super(message(status.reason(), status.code(), message));
+    this.status = status.code();
   }
 
   /**
@@ -251,7 +263,19 @@ public Map<String, Object> toMap() {
    * @return An error message.
    */
   private static String message(final Status status, final String tail) {
-    requireNonNull(status, "A HTTP Status is required.");
-    return status.reason() + "(" + status.value() + ")" + (tail == null ? "" : ": " + tail);
+    return message(status.reason(), status.value(), tail);
   }
+
+  /**
+   * Build an error message using the HTTP status.
+   *
+   * @param reason Reason.
+   * @param status The Status.
+   * @param tail A message to append.
+   * @return An error message.
+   */
+  private static String message(final String reason, int status, final String tail) {
+    return reason + "(" + status + ")" + (tail == null ? "" : ": " + tail);
+  }
+
 }

jooby/src/main/java/org/jooby/WebSocket.java

@@ -21,6 +21,7 @@
 import static java.util.Objects.requireNonNull;
 
 import java.io.Closeable;
+import java.nio.channels.ClosedChannelException;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -585,6 +586,13 @@ default void close() {
     close(NORMAL);
   }
 
+  /**
+   * True if the websocket is still open.
+   *
+   * @return True if the websocket is still open.
+   */
+  boolean isOpen();
+
   /**
    * Gracefully closes the connection, after sending a description message
    *
@@ -612,6 +620,8 @@ default void close() {
   /**
    * Send data through the connection.
    *
+   * If the web socket is closed this method throw an {@link Err} with {@link #NORMAL} close status.
+   *
    * @param data Data to send.
    * @throws Exception If something goes wrong.
    */
@@ -622,6 +632,8 @@ default void send(final Object data) throws Exception {
   /**
    * Send data through the connection.
    *
+   * If the web socket is closed this method throw an {@link Err} with {@link #NORMAL} close status.
+   *
    * @param data Data to send.
    * @param success A success callback.
    * @throws Exception If something goes wrong.
@@ -633,24 +645,26 @@ default void send(final Object data, final SuccessCallback success) throws Excep
   /**
    * Send data through the connection.
    *
+   * If the web socket is closed this method throw an {@link Err} with {@link #NORMAL} close status.
+   *
    * @param data Data to send.
    * @param err An err callback.
    * @throws Exception If something goes wrong.
    */
-  default void send(final Object data, final ErrCallback err)
-      throws Exception {
+  default void send(final Object data, final ErrCallback err) throws Exception {
     send(data, SUCCESS, err);
   }
 
   /**
    * Send data through the connection.
    *
+   * If the web socket is closed this method throw an {@link Err} with {@link #NORMAL} close status.
+   *
    * @param data Data to send.
    * @param success A success callback.
    * @param err An err callback.
    * @throws Exception If something goes wrong.
    */
-  void send(Object data, SuccessCallback success, ErrCallback err)
-      throws Exception;
+  void send(Object data, SuccessCallback success, ErrCallback err) throws Exception;
 
 }

jooby/src/main/java/org/jooby/internal/ConnectionResetByPeer.java

@@ -0,0 +1,19 @@
+package org.jooby.internal;
+
+import static javaslang.Predicates.instanceOf;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.Optional;
+
+public class ConnectionResetByPeer {
+
+  public static boolean test(final Throwable cause) {
+    return Optional.ofNullable(cause)
+        .filter(instanceOf(IOException.class))
+        .map(x -> x.getMessage())
+        .filter(Objects::nonNull)
+        .map(message -> message.toLowerCase().contains("connection reset by peer"))
+        .orElse(false);
+  }
+}