patx/mrhttp-asgi

import mrhttp


startup_seen = False
shutdown_seen = False


async def app(scope, receive, send):
  global startup_seen, shutdown_seen

  if scope["type"] == "lifespan":
    while True:
      message = await receive()
      if message["type"] == "lifespan.startup":
        startup_seen = True
        await send({"type": "lifespan.startup.complete"})
      elif message["type"] == "lifespan.shutdown":
        shutdown_seen = True
        await send({"type": "lifespan.shutdown.complete"})
        return

  assert scope["type"] == "http"
  message = await receive()
  body = message.get("body", b"")

  if scope["path"] == "/":
    content = b"hello asgi"
  elif scope["path"] == "/echo":
    content = body
  elif scope["path"] == "/scope":
    headers = scope["headers"]
    dup_count = len([h for h in headers if h[0] == b"x-dup"])
    content = "|".join([
      scope["method"],
      scope["path"],
      scope["raw_path"].decode("ascii"),
      scope["query_string"].decode("ascii"),
      str(dup_count),
      str(startup_seen),
    ]).encode("utf-8")
  else:
    await send({"type": "http.response.start", "status": 404, "headers": []})
    await send({"type": "http.response.body", "body": b"missing"})
    return

  await send({
    "type": "http.response.start",
    "status": 200,
    "headers": [(b"content-type", b"text/plain")],
  })
  await send({"type": "http.response.body", "body": content[:3], "more_body": True})
  await send({"type": "http.response.body", "body": content[3:]})


mrhttp.run(app, host="127.0.0.1", port=8080, workers=1, lifespan="on")