jacopofar/sse_fe.py

## sse_fe.py
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)
	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)