Fix websocket_closed not always being queued on failure

- Reorganise the HTTP test code to make it a bit more extensible. Add support for sending messages to connected websockets. - Provide a friendlier message for too-large-payload errors. - Return failure reason from Websocket.receive Fixes #2149.
2026-09-22 08:58:46 +00:00 · 2025-12-19 21:12:37 +00:00
parent 1520bebb6c
commit 90e7307fb4
5 changed files with 94 additions and 46 deletions
@@ -17,6 +17,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.DecoderException;
 import io.netty.handler.codec.TooLongFrameException;
+import io.netty.handler.codec.http.websocketx.CorruptedWebSocketFrameException;
+import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus;
 import io.netty.handler.codec.http.websocketx.WebSocketHandshakeException;
 import io.netty.handler.proxy.HttpProxyHandler;
 import io.netty.handler.proxy.Socks4ProxyHandler;
@@ -245,6 +247,10 @@ public final class NetworkUtils {
            return "Timed out";
        } else if (cause instanceof SSLHandshakeException || (cause instanceof DecoderException && cause.getCause() instanceof SSLHandshakeException)) {
            return "Could not create a secure connection";
+        } else if (cause instanceof CorruptedWebSocketFrameException e) {
+            return e.closeStatus() == WebSocketCloseStatus.MESSAGE_TOO_BIG
+                ? "Received a too-large message"
+                : "Corrupted websocket message";
        } else {
            return "Could not connect";
        }
@@ -57,8 +57,11 @@ public class WebsocketHandle {
     * @cc.treturn [1] string The received message.
     * @cc.treturn boolean If this was a binary message.
     * @cc.treturn [2] nil If the websocket was closed while waiting, or if we timed out.
+     * @cc.treturn [2] string The reason we failed to receive a message. Either the reason the websocket was closed
+     * (as returned by [`websocket_closed`], or the string {@code "Timed out"}.
     * @cc.changed 1.80pr1.13 Added return value indicating whether the message was binary.
     * @cc.changed 1.87.0 Added timeout argument.
+     * @cc.changed 1.117.0 Added return value indicating why receiving the message failed.
     */
    @LuaFunction
    public final MethodResult receive(Optional<Double> timeout) throws LuaException {
@@ -155,11 +158,11 @@ public class WebsocketHandle {
            } else if (event.length >= 2 && Objects.equals(event[0], CLOSE_EVENT) && Objects.equals(event[1], address) && websocket.isClosed()) {
                // If the socket is closed abort.
                environment.cancelTimer(timeoutId);
-                return MethodResult.of();
+                return MethodResult.of(null, event.length > 2 ? event[2] : "Connection closed");
            } else if (event.length >= 2 && timeoutId != -1 && Objects.equals(event[0], TIMER_EVENT)
                && event[1] instanceof Number id && id.intValue() == timeoutId) {
                // If we received a matching timer event then abort.
-                return MethodResult.of();
+                return MethodResult.of(null, "Timed out");
            }

            return pull;
@@ -71,9 +71,8 @@ class WebsocketHandler extends SimpleChannelInboundHandler<Object> {

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-        ctx.close();
-
        fail(NetworkUtils.toFriendlyError(cause));
+        ctx.close();
    }

    private void fail(String message) {
@@ -20,38 +20,49 @@ import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketSe
 import java.net.InetSocketAddress
 import java.nio.charset.StandardCharsets

-/**
- * Runs a small HTTP server to run alongside [TestHttpApi]
- */
-object HttpServer {
-    fun runServer(run: (port: Int, stop: () -> Unit) -> Unit) {
-        val workerGroup: EventLoopGroup = NioEventLoopGroup(2)
-        try {
-            val ch = ServerBootstrap()
-                .group(workerGroup)
-                .channel(NioServerSocketChannel::class.java)
-                .childHandler(
-                    object : ChannelInitializer<SocketChannel>() {
-                        override fun initChannel(ch: SocketChannel) {
-                            val p: ChannelPipeline = ch.pipeline()
-                            p.addLast(HttpServerCodec())
-                            p.addLast(HttpContentCompressor())
-                            p.addLast(HttpObjectAggregator(8192))
-                            p.addLast(HttpServerHandler())
-                            p.addLast(WebSocketServerCompressionHandler())
-                            p.addLast(WebSocketServerProtocolHandler("/ws", null, true))
-                            p.addLast(WebSocketFrameHandler())
-                        }
-                    },
-                ).bind(0).sync().channel()
-            val port = (ch.localAddress() as InetSocketAddress).port
+class HttpServer(val port: Int, private val workerGroup: EventLoopGroup, private val activeConnections: Set<Channel>) {
+    /** Stop the server from running */
+    fun stop() {
+        workerGroup.shutdownGracefully()
+    }
+
+    /** Broadcast this message to every connected websocket */
+    fun broadcast(message: WebSocketFrame) {
+        for (chan in activeConnections) chan.writeAndFlush(message)
+    }
+
+    companion object {
+        /** Runs a small HTTP server to run alongside [TestHttpApi] */
+        fun runServer(run: (server: HttpServer) -> Unit) {
+            val workerGroup: EventLoopGroup = NioEventLoopGroup(2)
+            val activeConnections = mutableSetOf<Channel>()
            try {
-                run(port) { workerGroup.shutdownGracefully() }
+                val ch = ServerBootstrap()
+                    .group(workerGroup)
+                    .channel(NioServerSocketChannel::class.java)
+                    .childHandler(
+                        object : ChannelInitializer<SocketChannel>() {
+                            override fun initChannel(ch: SocketChannel) {
+                                val p: ChannelPipeline = ch.pipeline()
+                                p.addLast(HttpServerCodec())
+                                p.addLast(HttpContentCompressor())
+                                p.addLast(HttpObjectAggregator(8192))
+                                p.addLast(HttpServerHandler())
+                                p.addLast(WebSocketServerCompressionHandler())
+                                p.addLast(WebSocketServerProtocolHandler("/ws", null, true))
+                                p.addLast(WebSocketFrameHandler(activeConnections))
+                            }
+                        },
+                    ).bind(0).sync().channel()
+                val port = (ch.localAddress() as InetSocketAddress).port
+                try {
+                    run(HttpServer(port, workerGroup, activeConnections))
+                } finally {
+                    ch.close().sync()
+                }
            } finally {
-                ch.close().sync()
+                workerGroup.shutdownGracefully().get()
            }
-        } finally {
-            workerGroup.shutdownGracefully().get()
        }
    }
 }
@@ -111,7 +122,7 @@ private class HttpServerHandler : SimpleChannelInboundHandler<FullHttpRequest>()
 /**
 * A basic WS server which just sends back the original message.
 */
-private class WebSocketFrameHandler : SimpleChannelInboundHandler<WebSocketFrame>() {
+private class WebSocketFrameHandler(private val activeConnections: MutableSet<Channel>) : SimpleChannelInboundHandler<WebSocketFrame>() {
    override fun channelRead0(ctx: ChannelHandlerContext, frame: WebSocketFrame) {
        if (frame is TextWebSocketFrame) {
            // Send the uppercase string back.
@@ -124,10 +135,16 @@ private class WebSocketFrameHandler : SimpleChannelInboundHandler<WebSocketFrame

    override fun userEventTriggered(ctx: ChannelHandlerContext, evt: Any) {
        if (evt is HandshakeComplete) {
-            // Channel upgrade to websocket, remove WebSocketIndexPageHandler.
+            // Channel upgrade to websocket, remove HttpServerHandler.
            ctx.pipeline().remove(HttpServerHandler::class.java)
+            activeConnections.add(ctx.channel())
        } else {
            super.userEventTriggered(ctx, evt)
        }
    }
+
+    override fun channelInactive(ctx: ChannelHandlerContext) {
+        super.channelInactive(ctx)
+        activeConnections.remove(ctx.channel())
+    }
 }
@@ -11,12 +11,14 @@ import dan200.computercraft.api.lua.ObjectArguments
 import dan200.computercraft.core.CoreConfig
 import dan200.computercraft.core.apis.HTTPAPI
 import dan200.computercraft.core.apis.handles.ReadHandle
-import dan200.computercraft.core.apis.http.HttpServer.runServer
+import dan200.computercraft.core.apis.http.HttpServer.Companion.runServer
 import dan200.computercraft.core.apis.http.options.Action
 import dan200.computercraft.core.apis.http.options.AddressRule
 import dan200.computercraft.core.apis.http.request.HttpResponseHandle
 import dan200.computercraft.core.apis.http.websocket.WebsocketHandle
 import dan200.computercraft.test.core.computer.LuaTaskRunner
+import io.netty.buffer.Unpooled
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame
 import org.hamcrest.MatcherAssert.assertThat
 import org.hamcrest.Matchers.*
 import org.junit.jupiter.api.AfterAll
@@ -48,9 +50,9 @@ class TestHttpApi {

    @Test
    fun `Connects to a HTTP server`() {
-        runServer { port, _ ->
+        runServer { server ->
            LuaTaskRunner.runTest {
-                val url = "http://127.0.0.1:$port"
+                val url = "http://127.0.0.1:${server.port}"
                val httpApi = addApi(HTTPAPI(environment))
                assertThat("http.request succeeded", httpApi.request(ObjectArguments(url)), array(equalTo(true)))

@@ -66,9 +68,9 @@ class TestHttpApi {

    @Test
    fun `Connects to websocket`() {
-        runServer { port, _ ->
+        runServer { server ->
            LuaTaskRunner.runTest {
-                val url = "ws://127.0.0.1:$port/ws"
+                val url = "ws://127.0.0.1:${server.port}/ws"
                val httpApi = addApi(HTTPAPI(environment))
                assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true)))

@@ -91,9 +93,9 @@ class TestHttpApi {

    @Test
    fun `Errors if too many websocket messages are sent`() {
-        runServer { port, _ ->
+        runServer { server ->
            LuaTaskRunner.runTest {
-                val url = "ws://127.0.0.1:$port/ws"
+                val url = "ws://127.0.0.1:${server.port}/ws"
                val httpApi = addApi(HTTPAPI(environment))
                assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true)))

@@ -115,10 +117,31 @@ class TestHttpApi {
    }

    @Test
-    fun `Queues an event when the socket is externally closed`() {
-        runServer { port, stop ->
+    fun `Closes if a websocket message is too large`() {
+        runServer { server ->
            LuaTaskRunner.runTest {
-                val url = "ws://127.0.0.1:$port/ws"
+                val url = "ws://127.0.0.1:${server.port}/ws"
+                val httpApi = addApi(HTTPAPI(environment))
+                assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true)))
+
+                val connectEvent = pullEvent()
+                assertThat(connectEvent, array(equalTo("websocket_success"), equalTo(url), isA(WebsocketHandle::class.java)))
+
+                val out = ByteArray(AddressRule.WEBSOCKET_MESSAGE + 1)
+                Random(0xDEADBEEF).nextBytes(out)
+                server.broadcast(BinaryWebSocketFrame(Unpooled.wrappedBuffer(out)))
+
+                val closeEvent = pullEvent()
+                assertThat(closeEvent, array(equalTo("websocket_closed"), equalTo(url), equalTo("Received a too-large message"), nullValue()))
+            }
+        }
+    }
+
+    @Test
+    fun `Queues an event when the socket is externally closed`() {
+        runServer { server ->
+            LuaTaskRunner.runTest {
+                val url = "ws://127.0.0.1:${server.port}/ws"
                val httpApi = addApi(HTTPAPI(environment))
                assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true)))

@@ -127,7 +150,7 @@ class TestHttpApi {

                val websocket = connectEvent[2] as WebsocketHandle

-                stop()
+                server.stop()

                val closeEvent = pullEvent("websocket_closed")
                assertThat(