WebSocket API: Undertow implementation

Author: jknack

examples/pom.xml

@@ -36,7 +36,7 @@
     </dependency>
     <dependency>
       <groupId>io.jooby</groupId>
-      <artifactId>jooby-netty</artifactId>
+      <artifactId>jooby-utow</artifactId>
       <version>${jooby.version}</version>
     </dependency>
     <dependency>

examples/src/main/java/examples/WebSocketApp.java

@@ -15,9 +15,11 @@ public class WebSocketApp extends Jooby {
     assets("/?*", Paths.get(System.getProperty("user.dir"), "examples", "www", "websocket"));
 
     ws("/ws", (ctx, initializer) -> {
+      System.out.println(Thread.currentThread());
+      System.out.println("Response Started: " + ctx.isResponseStarted());
       initializer.onConnect(ws -> {
+        System.out.println("Connected: " + Thread.currentThread());
         ws.send("Welcome");
-        ws.close();
       });
       initializer.onMessage((ws, msg) -> {
         ws.send("Got: " + msg.value(), true);
@@ -26,11 +28,14 @@ public class WebSocketApp extends Jooby {
         System.out.println("Closed " + closeStatus);
       });
 
+      initializer.onError((ws, cause) -> {
+        ws.getContext().getRouter().getLog().error("error ", cause);
+      });
+
     });
   }
 
   public static void main(String[] args) {
     runApp(args, ExecutionMode.DEFAULT, WebSocketApp::new);
-    //    runApp(args, ExecutionMode.EVENT_LOOP, WebSocketApp::new);
   }
 }

jooby/src/main/java/io/jooby/WebSocket.java

@@ -1,6 +1,7 @@
 package io.jooby;
 
 import javax.annotation.Nonnull;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -76,12 +77,12 @@ interface OnMessage {
 
   /**
    * On close callback. Generated when client close the connection or when explicit calls to
-   * {@link #close(WebSocketCloseStatus)} or {@link #disconnect()}.
+   * {@link #close(WebSocketCloseStatus)}.
    */
   interface OnClose {
     /**
      * Generated when client close the connection or when explicit calls to
-     * {@link #close(WebSocketCloseStatus)} or {@link #disconnect()}.
+     * {@link #close(WebSocketCloseStatus)}.
      *
      * @param ws WebSocket.
      * @param closeStatus Close status.
@@ -155,6 +156,21 @@ interface OnError {
     return send(message, false);
   }
 
+  /**
+   * Web sockets connected to the same path. This method doesn't include the current
+   * websocket.
+   *
+   * @return Web sockets or empty list.
+   */
+  @Nonnull List<WebSocket> getSessions();
+
+  /**
+   * True if websocket is open.
+   *
+   * @return True when open.
+   */
+  boolean isOpen();
+
   /**
    * Send a text message to current client (broadcast = false) or to ALL connected clients under the
    * websocket path (broadcast = true).

jooby/src/main/java/io/jooby/WebSocketMessage.java

@@ -4,6 +4,7 @@
 
 import javax.annotation.Nonnull;
 import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
 
 /**
  * Websocket message generated from a {@link WebSocket.OnMessage} callback. Message is a subclass.
@@ -32,4 +33,15 @@ public interface WebSocketMessage extends Value {
   static @Nonnull WebSocketMessage create(@Nonnull Context ctx, @Nonnull byte[] bytes) {
     return new WebSocketMessageImpl(ctx, bytes);
   }
+
+  /**
+   * Creates a websocket message.
+   *
+   * @param ctx HTTP context.
+   * @param message Text message.
+   * @return A websocket message.
+   */
+  static @Nonnull WebSocketMessage create(@Nonnull Context ctx, @Nonnull String message) {
+    return new WebSocketMessageImpl(ctx, message.getBytes(StandardCharsets.UTF_8));
+  }
 }

jooby/src/main/java/io/jooby/internal/ReadOnlyContext.java

@@ -34,6 +34,10 @@ public ReadOnlyContext(@Nonnull Context context) {
     super(context);
   }
 
+  @Override public boolean isResponseStarted() {
+    return true;
+  }
+
   @Nonnull @Override public Context send(@Nonnull Path file) {
     throw new IllegalStateException(MESSAGE);
   }

modules/jooby-jackson/src/main/java/io/jooby/json/JacksonModule.java

@@ -5,6 +5,7 @@
  */
 package io.jooby.json;
 
+import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParseException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.Module;
@@ -25,8 +26,10 @@
 import io.jooby.StatusCode;
 
 import javax.annotation.Nonnull;
+import java.io.DataOutput;
 import java.io.InputStream;
 import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
 import java.util.HashSet;
 import java.util.Set;
 
@@ -129,7 +132,7 @@ public JacksonModule module(Class<? extends Module> module) {
 
   @Override public byte[] encode(@Nonnull Context ctx, @Nonnull Object value) throws Exception {
     ctx.setDefaultResponseType(MediaType.json);
-    return mapper.writeValueAsBytes(value);
+    return mapper.writer().writeValueAsBytes(value);
   }
 
   @Override public <T> T decode(Context ctx, Type type) throws Exception {

modules/jooby-netty/src/main/java/io/jooby/internal/netty/NettyWebSocket.java

@@ -17,6 +17,8 @@
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 
+import javax.annotation.Nonnull;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
@@ -38,8 +40,8 @@ public class NettyWebSocket implements WebSocketConfigurer, WebSocket {
 
   public NettyWebSocket(NettyContext ctx) {
     this.netty = ctx;
-    this.key = ctx.pathString();
-    dispatch = !ctx.isInIoThread();
+    this.key = ctx.getRoute().getPattern();
+    this.dispatch = !ctx.isInIoThread();
   }
 
   public WebSocket send(String text, boolean broadcast) {
@@ -88,6 +90,16 @@ private WebSocket send(TextWebSocketFrame frame) {
     return Context.readOnly(netty);
   }
 
+  @Nonnull @Override public List<WebSocket> getSessions() {
+    List<WebSocket> sessions = all.get(key);
+    if (sessions == null) {
+      return Collections.emptyList();
+    }
+    List<WebSocket> result = new ArrayList<>(sessions);
+    result.remove(this);
+    return result;
+  }
+
   public boolean isOpen() {
     return netty.ctx.channel().isOpen();
   }

modules/jooby-utow/src/main/java/io/jooby/internal/utow/UtowContext.java

@@ -23,6 +23,8 @@
 import io.jooby.StatusCode;
 import io.jooby.Value;
 import io.jooby.ValueNode;
+import io.jooby.WebSocket;
+import io.undertow.Handlers;
 import io.undertow.io.IoCallback;
 import io.undertow.io.Sender;
 import io.undertow.server.HttpServerExchange;
@@ -32,6 +34,9 @@
 import io.undertow.util.Headers;
 import io.undertow.util.HttpString;
 import io.undertow.util.SameThreadExecutor;
+import io.undertow.websockets.WebSocketConnectionCallback;
+import io.undertow.websockets.core.WebSocketChannel;
+import io.undertow.websockets.spi.WebSocketHttpExchange;
 import org.slf4j.Logger;
 
 import javax.annotation.Nonnull;
@@ -222,6 +227,19 @@ public UtowContext(HttpServerExchange exchange, Router router) {
     return this;
   }
 
+  @Nonnull @Override public Context upgrade(@Nonnull WebSocket.Initializer handler) {
+    try {
+      Handlers.websocket((exchange, channel) -> {
+        UtowWebSocket ws = new UtowWebSocket(this, channel);
+        handler.init(Context.readOnly(this), ws);
+        ws.fireConnect();
+      }).handleRequest(exchange);
+      return this;
+    } catch (Exception x) {
+      throw SneakyThrows.propagate(x);
+    }
+  }
+
   @Nonnull @Override public StatusCode getResponseCode() {
     return StatusCode.valueOf(exchange.getStatusCode());
   }