WebSocket пример с репозитория Netty

LaunchServer/src/main/java/ru/gravit/launchserver/socket/NettyServerSocketHandler.java

@@ -8,8 +8,10 @@
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
 import io.netty.handler.logging.LogLevel;
 import io.netty.handler.logging.LoggingHandler;
 import ru.gravit.launcher.LauncherAPI;
@@ -20,6 +22,8 @@
 import ru.gravit.launcher.ssl.LauncherTrustManager;
 import ru.gravit.launchserver.LaunchServer;
 import ru.gravit.launchserver.response.Response;
+import ru.gravit.launchserver.socket.websocket.WebSocketFrameHandler;
+import ru.gravit.launchserver.socket.websocket.WebSocketIndexPageHandler;
 import ru.gravit.utils.helper.CommonHelper;
 import ru.gravit.utils.helper.LogHelper;
 import ru.gravit.utils.helper.VerifyHelper;
@@ -44,6 +48,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 public final class NettyServerSocketHandler implements Runnable, AutoCloseable {
+    private static final String WEBSOCKET_PATH = "/api";
     private static SSLServerSocketFactory ssf;
     private static final ThreadFactory THREAD_FACTORY = r -> CommonHelper.newThread("Network Thread", true, r);
     @LauncherAPI
@@ -133,11 +138,15 @@ public void run() {
                     .childHandler(new ChannelInitializer<NioSocketChannel>() {
                         @Override
                         public void initChannel(NioSocketChannel ch) throws Exception {
-                            ChannelPipeline p = ch.pipeline();
+                            ChannelPipeline pipeline = ch.pipeline();
                             //p.addLast(new LoggingHandler(LogLevel.INFO));
                             System.out.println("P!");
-                            p.addLast("httpServerCodec", new HttpServerCodec());
+                            pipeline.addLast(new HttpServerCodec());
-                            p.addLast(new WebSocketServerProtocolHandler("/chat"));
+                            pipeline.addLast(new HttpObjectAggregator(65536));
+                            pipeline.addLast(new WebSocketServerCompressionHandler());
+                            pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
+                            pipeline.addLast(new WebSocketIndexPageHandler(WEBSOCKET_PATH));
+                            pipeline.addLast(new WebSocketFrameHandler());
                         }
                     });
             ChannelFuture f = b.bind(server.config.getSocketAddress()).sync();

LaunchServer/src/main/java/ru/gravit/launchserver/socket/websocket/WebSocketFrameHandler.java

@@ -0,0 +1,28 @@
+package ru.gravit.launchserver.socket.websocket;
+
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketFrame;
+import io.netty.util.concurrent.EventExecutorGroup;
+
+import java.util.Locale;
+
+public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
+
+    @Override
+
+    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
+        // ping and pong frames already handled
+
+        if (frame instanceof TextWebSocketFrame) {
+            // Send the uppercase string back.
+            String request = ((TextWebSocketFrame) frame).text();
+            ctx.channel().writeAndFlush(new TextWebSocketFrame(request.toUpperCase(Locale.US)));
+        } else {
+            String message = "unsupported frame type: " + frame.getClass().getName();
+            throw new UnsupportedOperationException(message);
+        }
+    }
+}

LaunchServer/src/main/java/ru/gravit/launchserver/socket/websocket/WebSocketIndexPageHandler.java

@@ -0,0 +1,84 @@
+package ru.gravit.launchserver.socket.websocket;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.*;
+import io.netty.handler.codec.http.*;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.CharsetUtil;
+
+import static io.netty.handler.codec.http.HttpMethod.GET;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static io.netty.handler.codec.rtsp.RtspResponseStatuses.NOT_FOUND;
+
+public class WebSocketIndexPageHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+
+    private final String websocketPath;
+
+    public WebSocketIndexPageHandler(String websocketPath) {
+        this.websocketPath = websocketPath;
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
+        // Handle a bad request.
+        if (!req.decoderResult().isSuccess()) {
+            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
+            return;
+        }
+
+        // Allow only GET methods.
+        if (req.method() != GET) {
+            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
+            return;
+        }
+
+        // Send the index page
+        if ("/".equals(req.uri()) || "/index.html".equals(req.uri())) {
+            String webSocketLocation = getWebSocketLocation(ctx.pipeline(), req, websocketPath);
+            ByteBuf content = WebSocketServerIndexPage.getContent(webSocketLocation);
+            FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, OK, content);
+
+            res.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");
+            HttpUtil.setContentLength(res, content.readableBytes());
+
+            sendHttpResponse(ctx, req, res);
+        } else {
+            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        cause.printStackTrace();
+        ctx.close();
+    }
+
+    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
+        // Generate an error page if response getStatus code is not OK (200).
+        if (res.status().code() != 200) {
+            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
+            res.content().writeBytes(buf);
+            buf.release();
+            HttpUtil.setContentLength(res, res.content().readableBytes());
+        }
+
+        // Send the response and close the connection if necessary.
+        ChannelFuture f = ctx.channel().writeAndFlush(res);
+        if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
+            f.addListener(ChannelFutureListener.CLOSE);
+        }
+    }
+
+    private static String getWebSocketLocation(ChannelPipeline cp, HttpRequest req, String path) {
+        String protocol = "ws";
+        if (cp.get(SslHandler.class) != null) {
+            // SSL in use so use Secure WebSockets
+            protocol = "wss";
+        }
+        return protocol + "://" + req.headers().get(HttpHeaderNames.HOST) + path;
+    }
+}

LaunchServer/src/main/java/ru/gravit/launchserver/socket/websocket/WebSocketServerIndexPage.java

@@ -0,0 +1,61 @@
+package ru.gravit.launchserver.socket.websocket;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.util.CharsetUtil;
+
+public final class WebSocketServerIndexPage {
+
+    private static final String NEWLINE = "\r\n";
+
+    public static ByteBuf getContent(String webSocketLocation) {
+        return Unpooled.copiedBuffer(
+                "<html><head><title>Web Socket Test</title></head>" + NEWLINE +
+                        "<body>" + NEWLINE +
+                        "<script type=\"text/javascript\">" + NEWLINE +
+                        "var socket;" + NEWLINE +
+                        "if (!window.WebSocket) {" + NEWLINE +
+                        "  window.WebSocket = window.MozWebSocket;" + NEWLINE +
+                        '}' + NEWLINE +
+                        "if (window.WebSocket) {" + NEWLINE +
+                        "  socket = new WebSocket(\"" + webSocketLocation + "\");" + NEWLINE +
+                        "  socket.onmessage = function(event) {" + NEWLINE +
+                        "    var ta = document.getElementById('responseText');" + NEWLINE +
+                        "    ta.value = ta.value + '\\n' + event.data" + NEWLINE +
+                        "  };" + NEWLINE +
+                        "  socket.onopen = function(event) {" + NEWLINE +
+                        "    var ta = document.getElementById('responseText');" + NEWLINE +
+                        "    ta.value = \"Web Socket opened!\";" + NEWLINE +
+                        "  };" + NEWLINE +
+                        "  socket.onclose = function(event) {" + NEWLINE +
+                        "    var ta = document.getElementById('responseText');" + NEWLINE +
+                        "    ta.value = ta.value + \"Web Socket closed\"; " + NEWLINE +
+                        "  };" + NEWLINE +
+                        "} else {" + NEWLINE +
+                        "  alert(\"Your browser does not support Web Socket.\");" + NEWLINE +
+                        '}' + NEWLINE +
+                        NEWLINE +
+                        "function send(message) {" + NEWLINE +
+                        "  if (!window.WebSocket) { return; }" + NEWLINE +
+                        "  if (socket.readyState == WebSocket.OPEN) {" + NEWLINE +
+                        "    socket.send(message);" + NEWLINE +
+                        "  } else {" + NEWLINE +
+                        "    alert(\"The socket is not open.\");" + NEWLINE +
+                        "  }" + NEWLINE +
+                        '}' + NEWLINE +
+                        "</script>" + NEWLINE +
+                        "<form onsubmit=\"return false;\">" + NEWLINE +
+                        "<input type=\"text\" name=\"message\" value=\"Hello, World!\"/>" +
+                        "<input type=\"button\" value=\"Send Web Socket Data\"" + NEWLINE +
+                        "       onclick=\"send(this.form.message.value)\" />" + NEWLINE +
+                        "<h3>Output</h3>" + NEWLINE +
+                        "<textarea id=\"responseText\" style=\"width:500px;height:300px;\"></textarea>" + NEWLINE +
+                        "</form>" + NEWLINE +
+                        "</body>" + NEWLINE +
+                        "</html>" + NEWLINE, CharsetUtil.US_ASCII);
+    }
+
+    private WebSocketServerIndexPage() {
+        // Unused
+    }
+}