server.py

"""
Server that sends zstd-compressed HTTP responses where each chunk
is a complete, independent zstd frame.

This is valid HTTP but exposes a bug in httpx's ZStandardDecoder.
"""

import zstandard as zstd
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import time


class ZstdChunkedHandler(BaseHTTPRequestHandler):
    """Handler that sends multiple zstd frames as separate HTTP chunks."""

    def do_GET(self):
        if self.path == "/zstd-chunks":
            self.send_zstd_chunked_response()
        else:
            self.send_error(404)

    def send_zstd_chunked_response(self):
        # Send headers
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Transfer-Encoding", "chunked")
        self.send_header("Content-Encoding", "zstd")
        self.end_headers()

        # Create chunks of data, each compressed as a COMPLETE zstd frame
        messages = [
            b"First chunk of data - this is a complete zstd frame",
            b"Second chunk of data - another independent zstd frame",
            b"Third chunk of data - yet another independent zstd frame",
        ]

        cctx = zstd.ZstdCompressor()

        for msg in messages:
            # Each message is compressed as its own complete zstd frame
            # This is the key: each chunk is INDEPENDENT, not a continuous stream
            compressed = cctx.compress(msg)

            # Send as HTTP chunk: size in hex, \r\n, data, \r\n
            chunk_size = len(compressed)
            self.wfile.write(f"{chunk_size:x}\r\n".encode())
            self.wfile.write(compressed)
            self.wfile.write(b"\r\n")
            self.wfile.flush()
            print(f"Sent chunk: {chunk_size} bytes compressed, {len(msg)} bytes original")

            # Small delay to ensure chunks are received separately
            time.sleep(0.1)

        # Send final empty chunk to signal end of response
        self.wfile.write(b"0\r\n\r\n")
        self.wfile.flush()
        print("Response complete")

    def log_message(self, format, *args):
        # Suppress default logging
        pass


def run_server(port=8765):
    server = HTTPServer(("127.0.0.1", port), ZstdChunkedHandler)
    print(f"Server running on http://127.0.0.1:{port}")
    print("Endpoints:")
    print(f"  GET http://127.0.0.1:{port}/zstd-chunks - Sends zstd chunked response")
    print()
    server.serve_forever()


if __name__ == "__main__":
    run_server()