import asyncio
import json
from typing import Set
from aiohttp import web
from aiohttp_sse import EventSourceResponse, sse_response
channels = web.AppKey("channels", Set[asyncio.Queue[str]])
async def chat(_request: web.Request) -> web.Response:
html = """
Tiny Chat
Anonymous
:
"""
return web.Response(text=html, content_type="text/html")
async def message(request: web.Request) -> web.Response:
app = request.app
data = await request.post()
for queue in app[channels]:
payload = json.dumps(dict(data))
await queue.put(payload)
return web.Response()
async def subscribe(request: web.Request) -> EventSourceResponse:
async with sse_response(request) as response:
app = request.app
queue: asyncio.Queue[str] = asyncio.Queue()
print("Someone joined.")
app[channels].add(queue)
try:
while response.is_connected():
payload = await queue.get()
await response.send(payload)
queue.task_done()
finally:
app[channels].remove(queue)
print("Someone left.")
return response
if __name__ == "__main__":
app = web.Application()
app[channels] = set() # type: ignore[misc]
app.router.add_route("GET", "/", chat)
app.router.add_route("POST", "/everyone", message)
app.router.add_route("GET", "/subscribe", subscribe)
web.run_app(app, host="127.0.0.1", port=8080)