Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- @cherrypy.expose
- def getUpdate(self):
- #Set the expected headers...
- cherrypy.response.headers["Content-Type"] = "text/event-stream;charset=utf-8"
- def content():
- yield "Hello,"
- yield "world"
- return content()
- var sseEvent = new EventSource('http://localhost:8090/getUpdate');
- sseEvent.onmessage = function (event) {
- console.log(event);
- };
- sseEvent.onopen = function (event) {
- //console.log("I have started...");
- };
- @cherrypy.expose
- def getUpdate(self, _=None):
- cherrypy.response.headers["Content-Type"] = "text/event-stream;charset=utf-8"
- if _:
- data = 'retry: 200ndata: ' + str( self.prog_output) + 'nn'
- return data
- else:
- def content():
- data = 'retry: 200ndata: ' + str( self.prog_output) + 'nn'
- return data
- return content()
- getUpdate._cp_config = {'response.stream': True, 'tools.encode.encoding':'utf-8'}
- <!DOCTYPE html>
- <html lang="en">
- <head>
- <meta charset="utf-8">
- <title>title</title>
- <link rel="stylesheet" href="style.css">
- </head>
- <body>
- <div>server-sent events</div>
- <h3>sse_handler.onmessage()</h3>
- <ul id="sse_message"></ul>
- <script>
- // crée l'objet qui permettra la connexion avec le serveur
- var sse_handler = new EventSource('http://localhost:8080/getUpdate');
- sse_handler.onmessage = function (event) {
- console.log("-- sse_handler.onmessage()", event);
- /* https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Fields
- the onmessage handler catch all generic message, with no event field
- */
- var h_ul = document.getElementById('sse_message');
- var h_li = document.createElement("li");
- h_li.innerHTML = event.data;
- h_ul.appendChild(h_li);
- };
- sse_handler.onopen = function (event) {
- console.log("-- sse_handler.onopen()", event);
- };
- </script>
- </body>
- </html>
- #!/usr/bin/env python3
- import datetime
- import cherrypy
- from pathlib import Path
- class TestServerSentEvent(object):
- @cherrypy.expose
- def index(self):
- return Path("index.html").read_text()
- @cherrypy.expose
- def getUpdate(self, * pos, ** nam):
- cherrypy.response.headers["Content-Type"] = "text/event-stream;charset=utf-8"
- return 'retry: 1200ndata: {0}nn'.format(self.output())
- def output(self) :
- d = datetime.datetime.now()
- return f"TEST - {d}"
- getUpdate._cp_config = {'response.stream': True, 'tools.encode.encoding':'utf-8'}
- if __name__ == '__main__':
- cherrypy.quickstart(TestServerSentEvent())
Add Comment
Please, Sign In to add comment