|
|
"""Routers that connect WebSockets to ZMQ sockets."""
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Copyright (C) 2011 The IPython Development Team
|
|
|
#
|
|
|
# Distributed under the terms of the BSD License. The full license is in
|
|
|
# the file COPYING.txt, distributed as part of this software.
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Imports
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
import uuid
|
|
|
from Queue import Queue
|
|
|
import json
|
|
|
|
|
|
from IPython.config.configurable import Configurable
|
|
|
from IPython.utils.traitlets import Instance, Int, Dict
|
|
|
|
|
|
#-----------------------------------------------------------------------------
|
|
|
# Classes
|
|
|
#-----------------------------------------------------------------------------
|
|
|
|
|
|
class ZMQStreamRouter(Configurable):
|
|
|
|
|
|
zmq_stream = Instance('zmq.eventloop.zmqstream.ZMQStream')
|
|
|
session = Instance('IPython.zmq.session.Session')
|
|
|
max_msg_size = Int(2048, config=True, help="""
|
|
|
The max raw message size accepted from the browser
|
|
|
over a WebSocket connection.
|
|
|
""")
|
|
|
|
|
|
_clients = Dict()
|
|
|
|
|
|
def __init__(self, **kwargs):
|
|
|
super(ZMQStreamRouter,self).__init__(**kwargs)
|
|
|
self.zmq_stream.on_recv(self._on_zmq_reply)
|
|
|
|
|
|
def register_client(self, client):
|
|
|
"""Register a client, returning a client uuid."""
|
|
|
client_id = uuid.uuid4()
|
|
|
self._clients[client_id] = client
|
|
|
return client_id
|
|
|
|
|
|
def unregister_client(self, client_id):
|
|
|
"""Unregister a client by its client uuid."""
|
|
|
del self._clients[client_id]
|
|
|
|
|
|
def copy_clients(self, router):
|
|
|
"""Copy the clients of another router to this one.
|
|
|
|
|
|
This is used to enable the backend zeromq stream to disconnect
|
|
|
and reconnect while the WebSocket connections to browsers
|
|
|
remain, such as when a kernel is restarted.
|
|
|
"""
|
|
|
for client_id, client in router._clients.items():
|
|
|
client.router = self
|
|
|
self._clients[client_id] = client
|
|
|
|
|
|
def forward_msg(self, client_id, msg):
|
|
|
"""Forward a msg to a client by its id.
|
|
|
|
|
|
The default implementation of this will fail silently if a message
|
|
|
arrives on a socket that doesn't support it. This method should
|
|
|
use max_msg_size to check and silently discard message that are too
|
|
|
long."""
|
|
|
pass
|
|
|
|
|
|
def _on_zmq_reply(self, msg_list):
|
|
|
"""Handle a message the ZMQ stream sends to the router.
|
|
|
|
|
|
Usually, this is where the return message will be written to
|
|
|
clients that need it using client.write_message().
|
|
|
"""
|
|
|
pass
|
|
|
|
|
|
def _reserialize_reply(self, msg_list):
|
|
|
"""Reserialize a reply message using JSON.
|
|
|
|
|
|
This takes the msg list from the ZMQ socket, unserializes it using
|
|
|
self.session and then serializes the result using JSON. This method
|
|
|
should be used by self._on_zmq_reply to build messages that can
|
|
|
be sent back to the browser.
|
|
|
"""
|
|
|
idents, msg_list = self.session.feed_identities(msg_list)
|
|
|
msg = self.session.unserialize(msg_list)
|
|
|
msg['header'].pop('date')
|
|
|
msg.pop('buffers')
|
|
|
return json.dumps(msg)
|
|
|
|
|
|
|
|
|
class IOPubStreamRouter(ZMQStreamRouter):
|
|
|
|
|
|
def _on_zmq_reply(self, msg_list):
|
|
|
msg = self._reserialize_reply(msg_list)
|
|
|
for client_id, client in self._clients.items():
|
|
|
client.write_message(msg)
|
|
|
|
|
|
|
|
|
class ShellStreamRouter(ZMQStreamRouter):
|
|
|
|
|
|
_request_queue = Instance(Queue,(),{})
|
|
|
|
|
|
def _on_zmq_reply(self, msg_list):
|
|
|
msg = self._reserialize_reply(msg_list)
|
|
|
client_id = self._request_queue.get(block=False)
|
|
|
client = self._clients.get(client_id)
|
|
|
if client is not None:
|
|
|
client.write_message(msg)
|
|
|
|
|
|
def forward_msg(self, client_id, msg):
|
|
|
if len(msg) < self.max_msg_size:
|
|
|
msg = json.loads(msg)
|
|
|
to_send = self.session.serialize(msg)
|
|
|
self._request_queue.put(client_id)
|
|
|
self.session.send(self.zmq_stream, msg)
|
|
|
|
|
|
|
