Function app

python/vercel-runtime/tests/fixtures/wsgi_websocket_app.py:78–120

(environ, start_response)

def app(environ, start_response):
    is_upgrade = environ.get("HTTP_UPGRADE", "").lower() == "websocket"
    if not is_upgrade:
        # A normal HTTP request must still work through the regular WSGI path.
        body = b"not a websocket"
        start_response(
            "200 OK",
            [
                ("Content-Type", "text/plain"),
                ("Content-Length", str(len(body))),
            ],
        )
        return [body]

    sock = environ.get("werkzeug.socket")
    if sock is None:
        body = b"no socket in environ"
        start_response(
            "500 Internal Server Error",
            [
                ("Content-Type", "text/plain"),
                ("Content-Length", str(len(body))),
            ],
        )
        return [body]

    key = environ["HTTP_SEC_WEBSOCKET_KEY"]
    _handshake(sock, key)

    # Echo text frames back with an "echo:" prefix until the peer closes.
    while True:
        try:
            opcode, payload = _read_frame(sock)
        except (ConnectionError, OSError):
            break
        if opcode == 0x8:  # close
            _write_frame(sock, 0x8, struct.pack("!H", 1000))
            break
        if opcode == 0x1:  # text
            text = payload.decode("utf-8")
            _write_frame(sock, 0x1, f"echo:{text}".encode())

    return []