| 203 | self.server_options.update(kwargs) |
| 204 | |
| 205 | def init_app(self, app, **kwargs): |
| 206 | if app is not None: |
| 207 | if not hasattr(app, 'extensions'): |
| 208 | app.extensions = {} # pragma: no cover |
| 209 | app.extensions['socketio'] = self |
| 210 | self.server_options.update(kwargs) |
| 211 | self.manage_session = self.server_options.pop('manage_session', |
| 212 | self.manage_session) |
| 213 | |
| 214 | if 'client_manager' not in kwargs: |
| 215 | url = self.server_options.get('message_queue', None) |
| 216 | channel = self.server_options.pop('channel', 'flask-socketio') |
| 217 | write_only = app is None |
| 218 | if url: |
| 219 | if url.startswith(('redis://', "rediss://")): |
| 220 | queue_class = socketio.RedisManager |
| 221 | elif url.startswith('kafka://'): |
| 222 | queue_class = socketio.KafkaManager |
| 223 | elif url.startswith('zmq'): |
| 224 | queue_class = socketio.ZmqManager |
| 225 | else: |
| 226 | queue_class = socketio.KombuManager |
| 227 | queue = queue_class(url, channel=channel, |
| 228 | write_only=write_only) |
| 229 | self.server_options['client_manager'] = queue |
| 230 | |
| 231 | if 'json' in self.server_options and \ |
| 232 | self.server_options['json'] == flask_json: |
| 233 | # flask's json module is tricky to use because its output |
| 234 | # changes when it is invoked inside or outside the app context |
| 235 | # so here to prevent any ambiguities we replace it with wrappers |
| 236 | # that ensure that the app context is always present |
| 237 | class FlaskSafeJSON: |
| 238 | @staticmethod |
| 239 | def dumps(*args, **kwargs): |
| 240 | with app.app_context(): |
| 241 | return flask_json.dumps(*args, **kwargs) |
| 242 | |
| 243 | @staticmethod |
| 244 | def loads(*args, **kwargs): |
| 245 | with app.app_context(): |
| 246 | return flask_json.loads(*args, **kwargs) |
| 247 | |
| 248 | self.server_options['json'] = FlaskSafeJSON |
| 249 | |
| 250 | resource = self.server_options.pop('path', None) or \ |
| 251 | self.server_options.pop('resource', None) or 'socket.io' |
| 252 | if resource.startswith('/'): |
| 253 | resource = resource[1:] |
| 254 | if os.environ.get('FLASK_RUN_FROM_CLI'): |
| 255 | if self.server_options.get('async_mode') is None: |
| 256 | self.server_options['async_mode'] = 'threading' |
| 257 | self.server = socketio.Server(**self.server_options) |
| 258 | self.async_mode = self.server.async_mode |
| 259 | for handler in self.handlers: |
| 260 | self.server.on(handler[0], handler[1], namespace=handler[2]) |
| 261 | for namespace_handler in self.namespace_handlers: |
| 262 | self.server.register_namespace(namespace_handler) |