Created
February 14, 2020 21:39
-
-
Save jacopofar/b328948c018dc360da8471a930140c06 to your computer and use it in GitHub Desktop.
Starlette SSE with demo frontend
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from asyncio.queues import Queue | |
| import uvicorn | |
| from starlette.applications import Starlette | |
| from starlette.requests import Request | |
| from starlette.responses import JSONResponse, StreamingResponse, HTMLResponse | |
| class SSE: | |
| def __init__(self, data, event=None, event_id=None, retry=None): | |
| self.data = data | |
| self.event = event | |
| self.id = event_id | |
| self.retry = retry | |
| def encode(self): | |
| message = f"data: {self.data}" | |
| if self.event is not None: | |
| message += f"\nevent: {self.event}" | |
| if self.id is not None: | |
| message += f"\nid: {self.id}" | |
| if self.retry is not None: | |
| message += f"\nretry: {self.retry}" | |
| message += "\r\n\r\n" | |
| return message.encode("utf-8") | |
| app = Starlette(debug=True) | |
| app.queues = [] | |
| @app.route("/subscribe", methods=["GET"]) | |
| async def subscribe(request: Request): | |
| async def event_publisher(): | |
| while True: | |
| event = await queue.get() | |
| yield event.encode() | |
| queue = Queue() | |
| app.queues.append(queue) | |
| headers = { | |
| "Content-Type": "text/event-stream", | |
| "Cache-Control": "no-cache", | |
| "Connection": "keep-alive", | |
| "X-Accel-Buffering": "no", | |
| } | |
| return StreamingResponse(content=event_publisher(), headers=headers) | |
| @app.route("/publish", methods=["POST"]) | |
| async def publish(request: Request): | |
| payload = await request.json() | |
| data = payload["data"] | |
| for queue in app.queues: | |
| event = SSE(data) | |
| await queue.put(event) | |
| return JSONResponse({"message": "ok"}) | |
| @app.route("/", methods=["GET"]) | |
| async def index(request: Request): | |
| return HTMLResponse(''' | |
| <html> | |
| <header><title>SSE test</title></header> | |
| <body> | |
| <ul id="list"></ul> | |
| <script> | |
| const evtSource = new EventSource("/subscribe"); | |
| evtSource.onmessage = function(event) { | |
| const newElement = document.createElement("li"); | |
| const eventList = document.getElementById("list"); | |
| newElement.innerHTML = "message: " + event.data; | |
| eventList.appendChild(newElement); | |
| } | |
| </script> | |
| </body> | |
| </html> | |
| ''') | |
| if __name__ == "__main__": | |
| uvicorn.run("__main__:app", host="0.0.0.0", port=4321, reload=True) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment