ryu/ryu/app/wsgi.py

   1 # Copyright (C) 2012 Nippon Telegraph and Telephone Corporation.
   2 # Copyright (C) 2012 Isaku Yamahata <yamahata at private email ne jp>
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  13 # implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 import inspect
  18 from types import MethodType
  19
  20 from routes import Mapper
  21 from routes.util import URLGenerator
  22 import six
  23 from tinyrpc.server import RPCServer
  24 from tinyrpc.dispatch import RPCDispatcher
  25 from tinyrpc.dispatch import public as rpc_public
  26 from tinyrpc.protocols.jsonrpc import JSONRPCProtocol
  27 from tinyrpc.transports import ServerTransport, ClientTransport
  28 from tinyrpc.client import RPCClient
  29 import webob.dec
  30 import webob.exc
  31 from webob.request import Request as webob_Request
  32 from webob.response import Response as webob_Response
  33
  34 from ryu import cfg
  35 from ryu.lib import hub
  36
  37 DEFAULT_WSGI_HOST = '127.0.0.1'
  38 DEFAULT_WSGI_PORT = 8080
  39
  40 CONF = cfg.CONF
  41 CONF.register_cli_opts([
  42     cfg.StrOpt(
  43         'wsapi-host', default=DEFAULT_WSGI_HOST,
  44         help='webapp listen host (default %s)' % DEFAULT_WSGI_HOST),
  45     cfg.IntOpt(
  46         'wsapi-port', default=DEFAULT_WSGI_PORT,
  47         help='webapp listen port (default %s)' % DEFAULT_WSGI_PORT),
  48 ])
  49
  50 HEX_PATTERN = r'0x[0-9a-z]+'
  51 DIGIT_PATTERN = r'[1-9][0-9]*'
  52
  53
  54 def route(name, path, methods=None, requirements=None):
  55     def _route(controller_method):
  56         controller_method.routing_info = {
  57             'name': name,
  58             'path': path,
  59             'methods': methods,
  60             'requirements': requirements,
  61         }
  62         return controller_method
  63     return _route
  64
  65
  66 class Request(webob_Request):
  67     """
  68     Wrapper class for webob.request.Request.
  69
  70     The behavior of this class is the same as webob.request.Request
  71     except for setting "charset" to "UTF-8" automatically.
  72     """
  73     DEFAULT_CHARSET = "UTF-8"
  74
  75     def __init__(self, environ, charset=DEFAULT_CHARSET, *args, **kwargs):
  76         super(Request, self).__init__(
  77             environ, charset=charset, *args, **kwargs)
  78
  79
  80 class Response(webob_Response):
  81     """
  82     Wrapper class for webob.response.Response.
  83
  84     The behavior of this class is the same as webob.response.Response
  85     except for setting "charset" to "UTF-8" automatically.
  86     """
  87     DEFAULT_CHARSET = "UTF-8"
  88
  89     def __init__(self, charset=DEFAULT_CHARSET, *args, **kwargs):
  90         super(Response, self).__init__(charset=charset, *args, **kwargs)
  91
  92
  93 class WebSocketRegistrationWrapper(object):
  94
  95     def __init__(self, func, controller):
  96         self._controller = controller
  97         self._controller_method = MethodType(func, controller)
  98
  99     def __call__(self, ws):
 100         wsgi_application = self._controller.parent
 101         ws_manager = wsgi_application.websocketmanager
 102         ws_manager.add_connection(ws)
 103         try:
 104             self._controller_method(ws)
 105         finally:
 106             ws_manager.delete_connection(ws)
 107
 108
 109 class _AlreadyHandledResponse(Response):
 110     # XXX: Eventlet API should not be used directly.
 111     from eventlet.wsgi import ALREADY_HANDLED
 112     _ALREADY_HANDLED = ALREADY_HANDLED
 113
 114     def __call__(self, environ, start_response):
 115         return self._ALREADY_HANDLED
 116
 117
 118 def websocket(name, path):
 119     def _websocket(controller_func):
 120         def __websocket(self, req, **_):
 121             wrapper = WebSocketRegistrationWrapper(controller_func, self)
 122             ws_wsgi = hub.WebSocketWSGI(wrapper)
 123             ws_wsgi(req.environ, req.start_response)
 124             # XXX: In order to prevent the writing to a already closed socket.
 125             #      This issue is caused by combined use:
 126             #       - webob.dec.wsgify()
 127             #       - eventlet.wsgi.HttpProtocol.handle_one_response()
 128             return _AlreadyHandledResponse()
 129         __websocket.routing_info = {
 130             'name': name,
 131             'path': path,
 132             'methods': None,
 133             'requirements': None,
 134         }
 135         return __websocket
 136     return _websocket
 137
 138
 139 class ControllerBase(object):
 140     special_vars = ['action', 'controller']
 141
 142     def __init__(self, req, link, data, **config):
 143         self.req = req
 144         self.link = link
 145         self.data = data
 146         self.parent = None
 147         for name, value in config.items():
 148             setattr(self, name, value)
 149
 150     def __call__(self, req):
 151         action = self.req.urlvars.get('action', 'index')
 152         if hasattr(self, '__before__'):
 153             self.__before__()
 154
 155         kwargs = self.req.urlvars.copy()
 156         for attr in self.special_vars:
 157             if attr in kwargs:
 158                 del kwargs[attr]
 159
 160         return getattr(self, action)(req, **kwargs)
 161
 162
 163 class WebSocketDisconnectedError(Exception):
 164     pass
 165
 166
 167 class WebSocketServerTransport(ServerTransport):
 168     def __init__(self, ws):
 169         self.ws = ws
 170
 171     def receive_message(self):
 172         message = self.ws.wait()
 173         if message is None:
 174             raise WebSocketDisconnectedError()
 175         context = None
 176         return context, message
 177
 178     def send_reply(self, context, reply):
 179         self.ws.send(six.text_type(reply))
 180
 181
 182 class WebSocketRPCServer(RPCServer):
 183     def __init__(self, ws, rpc_callback):
 184         dispatcher = RPCDispatcher()
 185         dispatcher.register_instance(rpc_callback)
 186         super(WebSocketRPCServer, self).__init__(
 187             WebSocketServerTransport(ws),
 188             JSONRPCProtocol(),
 189             dispatcher,
 190         )
 191
 192     def serve_forever(self):
 193         try:
 194             super(WebSocketRPCServer, self).serve_forever()
 195         except WebSocketDisconnectedError:
 196             return
 197
 198     def _spawn(self, func, *args, **kwargs):
 199         hub.spawn(func, *args, **kwargs)
 200
 201
 202 class WebSocketClientTransport(ClientTransport):
 203
 204     def __init__(self, ws, queue):
 205         self.ws = ws
 206         self.queue = queue
 207
 208     def send_message(self, message, expect_reply=True):
 209         self.ws.send(six.text_type(message))
 210
 211         if expect_reply:
 212             return self.queue.get()
 213
 214
 215 class WebSocketRPCClient(RPCClient):
 216
 217     def __init__(self, ws):
 218         self.ws = ws
 219         self.queue = hub.Queue()
 220         super(WebSocketRPCClient, self).__init__(
 221             JSONRPCProtocol(),
 222             WebSocketClientTransport(ws, self.queue),
 223         )
 224
 225     def serve_forever(self):
 226         while True:
 227             msg = self.ws.wait()
 228             if msg is None:
 229                 break
 230             self.queue.put(msg)
 231
 232
 233 class wsgify_hack(webob.dec.wsgify):
 234     def __call__(self, environ, start_response):
 235         self.kwargs['start_response'] = start_response
 236         return super(wsgify_hack, self).__call__(environ, start_response)
 237
 238
 239 class WebSocketManager(object):
 240
 241     def __init__(self):
 242         self._connections = []
 243
 244     def add_connection(self, ws):
 245         self._connections.append(ws)
 246
 247     def delete_connection(self, ws):
 248         self._connections.remove(ws)
 249
 250     def broadcast(self, msg):
 251         for connection in self._connections:
 252             connection.send(msg)
 253
 254
 255 class WSGIApplication(object):
 256     def __init__(self, **config):
 257         self.config = config
 258         self.mapper = Mapper()
 259         self.registory = {}
 260         self._wsmanager = WebSocketManager()
 261         super(WSGIApplication, self).__init__()
 262
 263     def _match(self, req):
 264         # Note: Invoke the new API, first. If the arguments unmatched,
 265         # invoke the old API.
 266         try:
 267             return self.mapper.match(environ=req.environ)
 268         except TypeError:
 269             self.mapper.environ = req.environ
 270             return self.mapper.match(req.path_info)
 271
 272     @wsgify_hack
 273     def __call__(self, req, start_response):
 274         match = self._match(req)
 275
 276         if not match:
 277             return webob.exc.HTTPNotFound()
 278
 279         req.start_response = start_response
 280         req.urlvars = match
 281         link = URLGenerator(self.mapper, req.environ)
 282
 283         data = None
 284         name = match['controller'].__name__
 285         if name in self.registory:
 286             data = self.registory[name]
 287
 288         controller = match['controller'](req, link, data, **self.config)
 289         controller.parent = self
 290         return controller(req)
 291
 292     def register(self, controller, data=None):
 293         def _target_filter(attr):
 294             if not inspect.ismethod(attr) and not inspect.isfunction(attr):
 295                 return False
 296             if not hasattr(attr, 'routing_info'):
 297                 return False
 298             return True
 299         methods = inspect.getmembers(controller, _target_filter)
 300         for method_name, method in methods:
 301             routing_info = getattr(method, 'routing_info')
 302             name = routing_info['name']
 303             path = routing_info['path']
 304             conditions = {}
 305             if routing_info.get('methods'):
 306                 conditions['method'] = routing_info['methods']
 307             requirements = routing_info.get('requirements') or {}
 308             self.mapper.connect(name,
 309                                 path,
 310                                 controller=controller,
 311                                 requirements=requirements,
 312                                 action=method_name,
 313                                 conditions=conditions)
 314         if data:
 315             self.registory[controller.__name__] = data
 316
 317     @property
 318     def websocketmanager(self):
 319         return self._wsmanager
 320
 321
 322 class WSGIServer(hub.WSGIServer):
 323     def __init__(self, application, **config):
 324         super(WSGIServer, self).__init__((CONF.wsapi_host, CONF.wsapi_port),
 325                                          application, **config)
 326
 327     def __call__(self):
 328         self.serve_forever()
 329
 330
 331 def start_service(app_mgr):
 332     for instance in app_mgr.contexts.values():
 333         if instance.__class__ == WSGIApplication:
 334             return WSGIServer(instance)
 335
 336     return None