asmhatreandClaude Sonnet 5.5
fadfd05670
Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01KqJyCidz3ZgRyHABv2GVJh
35 lines
1.7 KiB
Python
35 lines
1.7 KiB
Python
"""The test application. respond() is shared by the Hypercorn ASGI wrapper (HTTP/1.1, HTTP/2) and the aioquic
|
|
HTTP/3 server. It reports which HTTP version the request arrived on, returns N bytes for /bytes/N, and can redirect."""
|
|
import os
|
|
|
|
ALT_SVC = os.environ.get("ALT_SVC_PORT") # set by serve.py in h3 mode: advertise h3 on that UDP port
|
|
|
|
def respond(path, version, host, port):
|
|
hdrs = []
|
|
if ALT_SVC:
|
|
hdrs.append((b"alt-svc", ('h3=":%s"; ma=3600' % ALT_SVC).encode()))
|
|
if path.startswith("/bytes/"):
|
|
body = b"x" * int(path.rsplit("/", 1)[1]); ctype = b"application/octet-stream"; status = 200
|
|
elif path == "/redirect":
|
|
hdrs.append((b"location", b"/hello")); return 302, hdrs, b""
|
|
else:
|
|
body = ("hello from %s:%s via HTTP/%s\n" % (host, port, version)).encode(); ctype = b"text/plain"; status = 200
|
|
hdrs += [(b"content-type", ctype), (b"content-length", str(len(body)).encode())]
|
|
return status, hdrs, body
|
|
|
|
async def app(scope, receive, send):
|
|
if scope["type"] == "lifespan":
|
|
while True:
|
|
m = await receive()
|
|
if m["type"] == "lifespan.startup":
|
|
await send({"type": "lifespan.startup.complete"})
|
|
elif m["type"] == "lifespan.shutdown":
|
|
await send({"type": "lifespan.shutdown.complete"}); return
|
|
if scope["type"] != "http":
|
|
return
|
|
while (await receive()).get("more_body"): # drain the request body
|
|
pass
|
|
status, headers, body = respond(scope["path"], scope["http_version"], scope["server"][0], scope["server"][1])
|
|
await send({"type": "http.response.start", "status": status, "headers": headers})
|
|
await send({"type": "http.response.body", "body": body})
|