Skip to content

Instantly share code, notes, and snippets.

@dketov
Created July 25, 2019 22:01
Show Gist options
  • Save dketov/1f5dfd3c0a80b2ee866f73a26d88a330 to your computer and use it in GitHub Desktop.
Save dketov/1f5dfd3c0a80b2ee866f73a26d88a330 to your computer and use it in GitHub Desktop.
import sys
import json
from wsgiref.simple_server import make_server
from ws4py.websocket import WebSocket
from ws4py.server.wsgirefserver import WSGIServer, WebSocketWSGIRequestHandler
from ws4py.server.wsgiutils import WebSocketWSGIApplication
class MyWebSocket(WebSocket):
def __init__(self, sock, ps, es, e):
self.e = e
super(MyWebSocket, self).__init__(sock, ps, es, e)
def received_message(self, message):
self.send(json.dumps(dict((k, str(v)) for k, v in self.e.iteritems())), False)
class MyApplication(WebSocketWSGIApplication):
def __call__(self, e, sr):
path_info = e["PATH_INFO"]
if path_info == "/ws":
return super(MyApplication, self).__call__(e, sr)
sr("200 Ok", [("Content-Type", "text/html")])
return "\n".join("<pre>%s=%s</pre>" % (k, v) for k, v in e.iteritems())
server = make_server('', 9000, server_class=WSGIServer,
handler_class=WebSocketWSGIRequestHandler,
app=MyApplication(handler_cls=MyWebSocket))
server.initialize_websockets_manager()
server.serve_forever()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment