Source

gevent-websocket / geventwebsocket / resource.py

from .protocols.base import BaseProtocol
from .exceptions import WebSocketError


class WebSocketApplication(object):
    def __init__(self, ws):
        self.protocol = self.build_protocol()
        self.ws = ws

    def handle(self):
        self.protocol.on_open()

        while True:
            try:
                message = self.ws.receive()
            except WebSocketError:
                break

            if message is None:
                self.protocol.on_close()
                break
            else:
                self.protocol.on_message(message)

    def on_open(self, *args, **kwargs):
        pass

    def on_close(self, *args, **kwargs):
        pass

    def on_message(self, message, *args, **kwargs):
        self.ws.send(message, **kwargs)

    def build_protocol(self):
        return BaseProtocol(self)

    @classmethod
    def protocol(self):
        return ''


class Resource(object):
    def __init__(self, apps=None, environ=None):
        self.environ = environ
        self.ws = None
        self.apps = apps if apps else []
        self.current_app = None

    def app_protocol(self, path):
        if path in self.apps:
            return self.apps[path].protocol()
        else:
            return ''

    def listen(self):
        self.ws = self.environ['wsgi.websocket']

        if self.ws.path in self.apps:
            self.current_app = self.apps[self.ws.path](self.ws)

        if self.current_app:
            self.current_app.ws = self.ws
            self.current_app.handle()
        else:
            raise Exception("No apps defined")

    def run_app(self, environ, start_response):
        if self.environ['PATH_INFO'] in self.apps:
            return self.apps[self.environ['PATH_INFO']](environ, start_response)
        else:
            raise Exception("No apps defined")

    def __call__(self, environ, start_response):
        self.environ = environ

        if 'wsgi.websocket' in self.environ:
            self.listen()

            return None
        else:
            return self.run_app(environ, start_response)
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.