Skip to content

Instantly share code, notes, and snippets.

@ahopkins
Last active May 30, 2024 15:03
Show Gist options
  • Save ahopkins/9cdf98147da863a0c1c5b0b8581d6980 to your computer and use it in GitHub Desktop.
Save ahopkins/9cdf98147da863a0c1c5b0b8581d6980 to your computer and use it in GitHub Desktop.
Server Sent Events
<!DOCTYPE html>
<script>
let eventSource
let in_focus = true
let should_run = false
window.onblur = function () {
in_focus = false
}
window.onfocus = function () {
in_focus = true
if (should_run) {
start()
}
}
function start() {
should_run = true
if (!window.EventSource) {
// IE or an old browser
alert("The browser doesn't support EventSource.")
return
}
eventSource = new EventSource('/sse')
eventSource.onopen = function (e) {
log("Event: open")
}
eventSource.onerror = function (e) {
log("Event: error")
if (this.readyState == EventSource.CONNECTING) {
log(`Reconnecting (readyState=${this.readyState})...`)
} else {
log("Error has occured.")
}
}
eventSource.addEventListener('bye', function (e) {
log("Event: bye, data: " + e.data)
if (e.data == "close") {
eventSource.close()
}
})
eventSource.addEventListener('first_only', function (e) {
log("Event: first_only, data: " + e.data)
})
eventSource.onmessage = function (e) {
log("Event: message, data: " + e.data)
if (!in_focus) {
do_stop()
}
}
}
function stop(manual) {
should_run = false
do_stop()
}
function do_stop() {
eventSource.close()
log("eventSource.close()")
}
function log(msg) {
console.log(msg)
logElem.innerHTML += msg + "<br>"
document.documentElement.scrollTop = 99999999
}
function reset() {
document.querySelector("#logElem").innerHTML = ""
}
</script>
<button onclick="start()">Start</button>
<button onclick="stop()">Stop</button>
<button onclick="reset()">Clear</button>
<pre id="logElem" style="margin: 6px 0"></pre>
import asyncio
from itertools import count
from sanic import Sanic
from sanic.request import Request
from sanic.response import redirect
c = count()
app = Sanic("__BASE__")
asyncio.set_event_loop_policy(None)
@app.get("/sse")
async def sse(request: Request):
print("Incoming SSE")
headers = {"Cache-Control": "no-cache"}
response = await request.respond(
headers=headers, content_type="text/event-stream"
)
await response.send("event: first_only\n")
while True:
i = next(c)
await response.send(f"data: {i}\n\n")
await asyncio.sleep(1)
if i and i % 4 == 0:
await response.send("event: bye\n")
await response.send("data: close\n\n")
break
await response.send("event: message\n")
print("Done")
app.static("/index.html", "./index.html")
@app.route("/")
def index(request):
return redirect("/index.html")
app.run(port=9999, dev=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment