Skip to content

Instantly share code, notes, and snippets.

@driftregion
Created August 6, 2021 11:52
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save driftregion/f731d36891df699f796f7313d3f94508 to your computer and use it in GitHub Desktop.
Save driftregion/f731d36891df699f796f7313d3f94508 to your computer and use it in GitHub Desktop.
Flask app that pushes updates to the browser with a websocket
#!/usr/bin/env python3
import threading
from flask import Flask
from websocket_server import WebsocketServer
import logging
import time
app = Flask(__name__)
@app.route("/")
def hello_world():
return """
<p id="counter">wating for server...</p>
<script>
var ws = new WebSocket("ws://" + document.domain + ":" + "13254" + "/json")
ws.onopen = function(event) {
ws.send("Webpage opened at " + Date.now() );
}
var counter_txt = document.getElementById("counter");
ws.onmessage = function(event) {
counter_txt.textContent = event.data;
}
</script>
"""
should_exit = threading.Event()
def onReceived(client, srv, msg):
print(f"server received {msg}")
server = WebsocketServer(13254, host='127.0.0.1', loglevel=logging.INFO)
server.set_fn_message_received(onReceived)
def websocketPeriodicTask():
i = 0
while not should_exit.is_set():
server.send_message_to_all(f"this is update number {i}")
i += 1
time.sleep(1)
def websocketServerTask():
while not should_exit.is_set():
server.run_forever()
threads = [threading.Thread(target=t) for t in [websocketServerTask, websocketPeriodicTask]]
[t.start() for t in threads]
try:
app.run(use_reloader=False)
except KeyboardInterrupt:
should_exit.set()
[t.join() for t in threads]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment