Websockets are broken

- remove manual demand, use `AutoDemaing`
- fix  #3814

Author: jknack

modules/jooby-jetty/src/main/java/io/jooby/internal/jetty/JettyWebSocket.java

@@ -8,7 +8,6 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import java.nio.ByteBuffer;
-import java.nio.channels.ReadPendingException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -36,7 +35,8 @@
 import io.jooby.WebSocketMessage;
 import io.jooby.output.Output;
 
-public class JettyWebSocket implements Session.Listener, WebSocketConfigurer, WebSocket {
+public class JettyWebSocket
+    implements Session.Listener.AutoDemanding, WebSocketConfigurer, WebSocket {
 
   private static class WriteCallbackAdaptor implements org.eclipse.jetty.websocket.api.Callback {
 
@@ -75,11 +75,7 @@ public void fail(Throwable cause) {
 
     @Override
     public void succeed() {
-      try {
-        callback.operationComplete(ws, null);
-      } finally {
-        ws.demand();
-      }
+      callback.operationComplete(ws, null);
     }
   }
 
@@ -138,14 +134,11 @@ public void onWebSocketOpen(Session session) {
       open.set(true);
       this.session = session;
       addSession(this);
-      demand();
       if (onConnectCallback != null) {
         onConnectCallback.onConnect(this);
       }
     } catch (Throwable x) {
       onWebSocketError(x);
-    } finally {
-      demand();
     }
   }
 
@@ -217,14 +210,6 @@ public Context getContext() {
     return Context.readOnly(ctx);
   }
 
-  private void demand() {
-    try {
-      session.demand();
-    } catch (ReadPendingException cause) {
-      ctx.getRouter().getLog().debug("Websocket resulted in exception: {}", path, cause);
-    }
-  }
-
   @NonNull @Override
   public List<WebSocket> getSessions() {
     List<JettyWebSocket> sessions = all.get(key);

tests/src/test/java/io/jooby/i3814/Issue3814.java

@@ -0,0 +1,83 @@
+/*
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.i3814;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import io.jooby.junit.ServerTest;
+import io.jooby.junit.ServerTestRunner;
+import okhttp3.Response;
+import okhttp3.WebSocket;
+import okhttp3.WebSocketListener;
+import okio.ByteString;
+
+public class Issue3814 {
+
+  @ServerTest
+  public void shouldJettyWebSocketWorks(ServerTestRunner runner) {
+    int messageCount = 10;
+    var latch = new CountDownLatch(messageCount);
+    runner
+        .define(
+            app -> {
+              app.ws(
+                  "/3814",
+                  (ctx, initializer) -> {
+                    initializer.onMessage(
+                        (ws, message) -> {
+                          latch.countDown();
+                        });
+                  });
+            })
+        .ready(
+            client -> {
+              client.webSocket(
+                  "/3814",
+                  new WebSocketListener() {
+                    @Override
+                    public void onOpen(@NotNull WebSocket ws, @NotNull Response response) {
+                      for (int i = 0; i < messageCount; i++) {
+                        ws.send(">" + i);
+                      }
+                    }
+
+                    @Override
+                    public void onMessage(@NotNull WebSocket webSocket, @NotNull String text) {
+                      super.onMessage(webSocket, text);
+                    }
+
+                    @Override
+                    public void onMessage(@NotNull WebSocket webSocket, @NotNull ByteString bytes) {
+                      super.onMessage(webSocket, bytes);
+                    }
+
+                    @Override
+                    public void onClosing(
+                        @NotNull WebSocket webSocket, int code, @NotNull String reason) {
+                      super.onClosing(webSocket, code, reason);
+                    }
+
+                    @Override
+                    public void onClosed(
+                        @NotNull WebSocket webSocket, int code, @NotNull String reason) {
+                      super.onClosed(webSocket, code, reason);
+                    }
+
+                    @Override
+                    public void onFailure(
+                        @NotNull WebSocket webSocket,
+                        @NotNull Throwable t,
+                        @Nullable Response response) {
+                      super.onFailure(webSocket, t, response);
+                    }
+                  });
+              latch.await();
+            });
+  }
+}

tests/src/test/java/io/jooby/test/WebClient.java

@@ -313,6 +313,14 @@ public void syncWebSocket(String path, SneakyThrows.Consumer<BlockingWebSocket>
     blockingWebSocket.close();
   }
 
+  public WebSocket webSocket(String path, WebSocketListener listener) {
+    okhttp3.Request.Builder req = new okhttp3.Request.Builder();
+    req.url("ws://localhost:" + port + path);
+    setRequestHeaders(req);
+    okhttp3.Request r = req.build();
+    return client.newWebSocket(r, listener);
+  }
+
   public Request options(String path) {
     return invoke("OPTIONS", path, null);
   }

tests/src/test/java/io/jooby/test/WebSocketTest.java

@@ -47,7 +47,7 @@ public void webSocket(ServerTestRunner runner) {
   }
 
   @ServerTest
-  public void webSocketRecieveByteMessage(ServerTestRunner runner) {
+  public void webSocketReceiveByteMessage(ServerTestRunner runner) {
     runner
         .define(
             app -> {