Work to adapt routers to new Session message protocol.

IPython/frontend/html/notebook/handlers.py

@@ -1,163 +1,116 b''
+"""Tornado handlers for the notebook."""
+
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
+
 import datetime
 import json
 import logging
 import os
 import urllib
-import uuid
-from Queue import Queue
 
 from tornado import web
 from tornado import websocket
 
+#-----------------------------------------------------------------------------
+# Handlers
+#-----------------------------------------------------------------------------
 
-_kernel_id_regex = r"(?P<kernel_id>\w+-\w+-\w+-\w+-\w+)"
-_kernel_action_regex = r"(?P<action>restart|interrupt)"
 
 class MainHandler(web.RequestHandler):
     def get(self):
         self.render('notebook.html')
 
 
 class KernelHandler(web.RequestHandler):
 
     def get(self):
         self.write(json.dumps(self.application.kernel_ids))
 
     def post(self):
         kernel_id = self.application.start_kernel()
         self.write(json.dumps(kernel_id))
 
 
 class KernelActionHandler(web.RequestHandler):
 
     def post(self, kernel_id, action):
         # TODO: figure out a better way of handling RPC style calls.
         if action == 'interrupt':
             self.application.interrupt_kernel(kernel_id)
         if action == 'restart':
             new_kernel_id = self.application.restart_kernel(kernel_id)
             self.write(json.dumps(new_kernel_id))
 
 
-class ZMQStreamRouter(object):
-
-    def __init__(self, zmq_stream):
-        self.zmq_stream = zmq_stream
-        self._clients = {}
-        self.zmq_stream.on_recv(self._on_zmq_reply)
-
-    def register_client(self, client):
-        client_id = uuid.uuid4()
-        self._clients[client_id] = client
-        return client_id
-
-    def unregister_client(self, client_id):
-        del self._clients[client_id]
-
-    def copy_clients(self, router):
-        # Copy the clients of another router.
-        for client_id, client in router._clients.items():
-            client.router = self
-            self._clients[client_id] = client
-
-
-class IOPubStreamRouter(ZMQStreamRouter):
-
-    def _on_zmq_reply(self, msg_list):
-        for client_id, client in self._clients.items():
-            for msg in msg_list:
-                client.write_message(msg)
-
-    def forward_unicode(self, client_id, msg):
-        # This is a SUB stream that we should never write to.
-        pass
-
-
-class ShellStreamRouter(ZMQStreamRouter):
-
-    def __init__(self, zmq_stream):
-        ZMQStreamRouter.__init__(self, zmq_stream)
-        self._request_queue = Queue()
-
-    def _on_zmq_reply(self, msg_list):
-        client_id = self._request_queue.get(block=False)
-        client = self._clients.get(client_id)
-        if client is not None:
-            for msg in msg_list:
-                client.write_message(msg)
-
-    def forward_unicode(self, client_id, msg):
-        self._request_queue.put(client_id)
-        self.zmq_stream.send_unicode(msg)
-
-
 class ZMQStreamHandler(websocket.WebSocketHandler):
 
     def initialize(self, stream_name):
         self.stream_name = stream_name
 
     def open(self, kernel_id):
         self.router = self.application.get_router(kernel_id, self.stream_name)
         self.client_id = self.router.register_client(self)
         logging.info("Connection open: %s, %s" % (kernel_id, self.client_id))
 
     def on_message(self, msg):
         self.router.forward_unicode(self.client_id, msg)
 
     def on_close(self):
         self.router.unregister_client(self.client_id)
         logging.info("Connection closed: %s" % self.client_id)
 
 
 class NotebookRootHandler(web.RequestHandler):
 
     def get(self):
         files = os.listdir(os.getcwd())
         files = [file for file in files if file.endswith(".ipynb")]
         self.write(json.dumps(files))
 
 
 class NotebookHandler(web.RequestHandler):
 
     SUPPORTED_METHODS = ("GET", "DELETE", "PUT")
 
     def find_path(self, filename):
         filename = urllib.unquote(filename)
         if not filename.endswith('.ipynb'):
             raise web.HTTPError(400)
         path = os.path.join(os.getcwd(), filename)
         return path
 
     def get(self, filename):
         path = self.find_path(filename)
         if not os.path.isfile(path):
             raise web.HTTPError(404)
         info = os.stat(path)
         self.set_header("Content-Type", "application/unknown")
         self.set_header("Last-Modified", datetime.datetime.utcfromtimestamp(
             info.st_mtime))
         f = open(path, "r")
         try:
             self.finish(f.read())
         finally:
             f.close()
 
     def put(self, filename):
         path = self.find_path(filename)
         f = open(path, "w")
         f.write(self.request.body)
         f.close()
         self.finish()
 
     def delete(self, filename):
         path = self.find_path(filename)
         if not os.path.isfile(path):
             raise web.HTTPError(404)
         os.unlink(path)
         self.set_status(204)
         self.finish()
 
 
 
 
 

IPython/frontend/html/notebook/kernelmanager.py

@@ -1,182 +1,181 b''
 """A kernel manager for multiple kernels."""
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
-import logging
 import signal
 import sys
 import uuid
 
 import zmq
 
 from IPython.config.configurable import LoggingConfigurable
 from IPython.zmq.ipkernel import launch_kernel
-from IPython.utils.traitlets import Instance, Dict, Unicode
+from IPython.utils.traitlets import Instance, Dict
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class DuplicateKernelError(Exception):
     pass
 
 
 class KernelManager(LoggingConfigurable):
     """A class for managing multiple kernels."""
 
     context = Instance('zmq.Context')
     def _context_default(self):
         return zmq.Context.instance()
 
     _kernels = Dict()
 
     @property
     def kernel_ids(self):
         """Return a list of the kernel ids of the active kernels."""
         return self._kernels.keys()
 
     def __len__(self):
         """Return the number of running kernels."""
         return len(self.kernel_ids)
 
     def __contains__(self, kernel_id):
         if kernel_id in self.kernel_ids:
             return True
         else:
             return False
 
     def start_kernel(self, **kwargs):
         """Start a new kernel."""
         kernel_id = str(uuid.uuid4())
         (process, shell_port, iopub_port, stdin_port, hb_port) = launch_kernel(**kwargs)
         # Store the information for contacting the kernel. This assumes the kernel is
         # running on localhost.
         d = dict(
             process = process,
             stdin_port = stdin_port,
             iopub_port = iopub_port,
             shell_port = shell_port,
             hb_port = hb_port,
             ip = '127.0.0.1'
         )
         self._kernels[kernel_id] = d
         return kernel_id
 
     def kill_kernel(self, kernel_id):
         """Kill a kernel by its kernel uuid.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel to kill.
         """
         kernel_process = self.get_kernel_process(kernel_id)
         if kernel_process is not None:
             # Attempt to kill the kernel.
             try:
                 kernel_process.kill()
             except OSError, e:
                 # In Windows, we will get an Access Denied error if the process
                 # has already terminated. Ignore it.
                 if not (sys.platform == 'win32' and e.winerror == 5):
                     raise
             del self._kernels[kernel_id]
 
     def interrupt_kernel(self, kernel_id):
         """Interrupt (SIGINT) the kernel by its uuid.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel to interrupt.
         """
         kernel_process = self.get_kernel_process(kernel_id)
         if kernel_process is not None:
             if sys.platform == 'win32':
                 from parentpoller import ParentPollerWindows as Poller
                 Poller.send_interrupt(kernel_process.win32_interrupt_event)
             else:
                 kernel_process.send_signal(signal.SIGINT)
 
     def signal_kernel(self, kernel_id, signum):
         """ Sends a signal to the kernel by its uuid.
 
         Note that since only SIGTERM is supported on Windows, this function
         is only useful on Unix systems.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel to signal.
         """
         kernel_process = self.get_kernel_process(kernel_id)
         if kernel_process is not None:
             kernel_process.send_signal(signum)
 
     def get_kernel_process(self, kernel_id):
         """Get the process object for a kernel by its uuid.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel.
         """
         d = self._kernels.get(kernel_id)
         if d is not None:
             return d['process']
         else:
             raise KeyError("Kernel with id not found: %s" % kernel_id)
 
     def get_kernel_ports(self, kernel_id):
         """Return a dictionary of ports for a kernel.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel.
 
         Returns
         =======
         port_dict : dict
             A dict of key, value pairs where the keys are the names
             (stdin_port,iopub_port,shell_port) and the values are the
             integer port numbers for those channels.
         """
         d = self._kernels.get(kernel_id)
         if d is not None:
             dcopy = d.copy()
             dcopy.pop('process')
             dcopy.pop('ip')
             return dcopy
         else:
             raise KeyError("Kernel with id not found: %s" % kernel_id)
 
     def get_kernel_ip(self, kernel_id):
         """Return ip address for a kernel.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel.
 
         Returns
         =======
         ip : str
             The ip address of the kernel.
         """
         d = self._kernels.get(kernel_id)
         if d is not None:
             return d['ip']
         else:
             raise KeyError("Kernel with id not found: %s" % kernel_id)
 
     def create_session_manager(self, kernel_id):
         """Create a new session manager for a kernel by its uuid."""
         from sessionmanager import SessionManager
         return SessionManager(
             kernel_id=kernel_id, kernel_manager=self, 
             config=self.config, context=self.context, log=self.log
         )
 

IPython/frontend/html/notebook/notebookapp.py

@@ -1,247 +1,249 b''
 """A tornado based IPython notebook server."""
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import logging
 import os
 import signal
 import sys
 
 import zmq
 
 # Install the pyzmq ioloop. This has to be done before anything else from
 # tornado is imported.
 from zmq.eventloop import ioloop
 import tornado.ioloop
 tornado.ioloop = ioloop
 
 from tornado import httpserver
 from tornado import web
 
 from kernelmanager import KernelManager
 from sessionmanager import SessionManager
 from handlers import (
     MainHandler, KernelHandler, KernelActionHandler, ZMQStreamHandler,
     NotebookRootHandler, NotebookHandler
 )
 from routers import IOPubStreamRouter, ShellStreamRouter
 
 from IPython.core.application import BaseIPythonApplication
 from IPython.core.profiledir import ProfileDir
 from IPython.frontend.qt.console.rich_ipython_widget import RichIPythonWidget
 from IPython.zmq.session import Session
 from IPython.zmq.zmqshell import ZMQInteractiveShell
 from IPython.zmq.ipkernel import (
     flags as ipkernel_flags,
     aliases as ipkernel_aliases,
     IPKernelApp
 )
 from IPython.utils.traitlets import Dict, Unicode, Int, Any, List, Enum
 
 #-----------------------------------------------------------------------------
 # Module globals
 #-----------------------------------------------------------------------------
 
 _kernel_id_regex = r"(?P<kernel_id>\w+-\w+-\w+-\w+-\w+)"
 _kernel_action_regex = r"(?P<action>restart|interrupt)"
 
 LOCALHOST = '127.0.0.1'
 
 #-----------------------------------------------------------------------------
 # The Tornado web application
 #-----------------------------------------------------------------------------
 
 class NotebookWebApplication(web.Application):
 
     def __init__(self, kernel_manager, log, kernel_argv):
         handlers = [
             (r"/", MainHandler),
             (r"/kernels", KernelHandler),
             (r"/kernels/%s/%s" % (_kernel_id_regex, _kernel_action_regex), KernelActionHandler),
             (r"/kernels/%s/iopub" % _kernel_id_regex, ZMQStreamHandler, dict(stream_name='iopub')),
             (r"/kernels/%s/shell" % _kernel_id_regex, ZMQStreamHandler, dict(stream_name='shell')),
             (r"/notebooks", NotebookRootHandler),
             (r"/notebooks/([^/]+)", NotebookHandler)
         ]
         settings = dict(
             template_path=os.path.join(os.path.dirname(__file__), "templates"),
             static_path=os.path.join(os.path.dirname(__file__), "static"),
         )
         web.Application.__init__(self, handlers, **settings)
 
         self.kernel_manager = kernel_manager
         self.log = log
         self.kernel_argv = kernel_argv
         self._routers = {}
         self._session_dict = {}
 
     #-------------------------------------------------------------------------
     # Methods for managing kernels and sessions
     #-------------------------------------------------------------------------
 
     @property
     def kernel_ids(self):
         return self.kernel_manager.kernel_ids
 
     def start_kernel(self):
         kwargs = dict()
         kwargs['extra_arguments'] = self.kernel_argv
         kernel_id = self.kernel_manager.start_kernel(**kwargs)
         self.log.info("Kernel started: %s" % kernel_id)
         self.log.debug("Kernel args: %r" % kwargs)
         self.start_session_manager(kernel_id)
         return kernel_id
 
     def start_session_manager(self, kernel_id):
         sm = self.kernel_manager.create_session_manager(kernel_id)
         self._session_dict[kernel_id] = sm
         iopub_stream = sm.get_iopub_stream()
         shell_stream = sm.get_shell_stream()
-        iopub_router = IOPubStreamRouter(iopub_stream)
-        shell_router = ShellStreamRouter(shell_stream)
+        iopub_router = IOPubStreamRouter(iopub_stream, sm.session)
+        shell_router = ShellStreamRouter(shell_stream, sm.session)
         self._routers[(kernel_id, 'iopub')] = iopub_router
         self._routers[(kernel_id, 'shell')] = shell_router
 
     def kill_kernel(self, kernel_id):
         sm = self._session_dict.pop(kernel_id)
         sm.stop()
         self.kernel_manager.kill_kernel(kernel_id)
         self.log.info("Kernel killed: %s" % kernel_id)
 
     def interrupt_kernel(self, kernel_id):
         self.kernel_manager.interrupt_kernel(kernel_id)
         self.log.debug("Kernel interrupted: %s" % kernel_id)
 
     def restart_kernel(self, kernel_id):
         # Create the new kernel first so we can move the clients over.
         new_kernel_id = self.start_kernel()
 
         # Copy the clients over to the new routers.
         old_iopub_router = self.get_router(kernel_id, 'iopub')
         old_shell_router = self.get_router(kernel_id, 'shell')
         new_iopub_router = self.get_router(new_kernel_id, 'iopub')
         new_shell_router = self.get_router(new_kernel_id, 'shell')
         new_iopub_router.copy_clients(old_iopub_router)
         new_shell_router.copy_clients(old_shell_router)
 
         # Now shutdown the old session and the kernel.
         # TODO: This causes a hard crash in ZMQStream.close, which sets
         # self.socket to None to hastily. We will need to fix this in PyZMQ
         # itself. For now, we just leave the old kernel running :(
         # self.kill_kernel(kernel_id)
 
         self.log.debug("Kernel restarted: %s -> %s" % (kernel_id, new_kernel_id))
         return new_kernel_id
 
     def get_router(self, kernel_id, stream_name):
         router = self._routers[(kernel_id, stream_name)]
         return router
 
+    
+
 #-----------------------------------------------------------------------------
 # Aliases and Flags
 #-----------------------------------------------------------------------------
 
 flags = dict(ipkernel_flags)
 
 # the flags that are specific to the frontend
 # these must be scrubbed before being passed to the kernel,
 # or it will raise an error on unrecognized flags
 notebook_flags = []
 
 aliases = dict(ipkernel_aliases)
 
 aliases.update(dict(
     ip = 'IPythonNotebookApp.ip',
     port = 'IPythonNotebookApp.port',
     colors = 'ZMQInteractiveShell.colors',
     editor = 'RichIPythonWidget.editor',
 ))
 
 #-----------------------------------------------------------------------------
 # IPythonNotebookApp
 #-----------------------------------------------------------------------------
 
 class IPythonNotebookApp(BaseIPythonApplication):
     name = 'ipython-notebook'
     default_config_file_name='ipython_notebook_config.py'
     
     description = """
         The IPython HTML Notebook.
         
         This launches a Tornado based HTML Notebook Server that serves up an
         HTML5/Javascript Notebook client.
     """
     
     classes = [IPKernelApp, ZMQInteractiveShell, ProfileDir, Session,
                KernelManager, SessionManager, RichIPythonWidget]
     flags = Dict(flags)
     aliases = Dict(aliases)
 
     kernel_argv = List(Unicode)
 
     log_level = Enum((0,10,20,30,40,50,'DEBUG','INFO','WARN','ERROR','CRITICAL'),
                     default_value=logging.INFO,
                     config=True,
                     help="Set the log level by value or name.")
 
     # connection info:
     ip = Unicode(LOCALHOST, config=True,
         help="The IP address the notebook server will listen on."
     )
 
     port = Int(8888, config=True,
         help="The port the notebook server will listen on."
     )
 
     # the factory for creating a widget
     widget_factory = Any(RichIPythonWidget)
 
     def parse_command_line(self, argv=None):
         super(IPythonNotebookApp, self).parse_command_line(argv)
         if argv is None:
             argv = sys.argv[1:]
 
         self.kernel_argv = list(argv) # copy
         # kernel should inherit default config file from frontend
         self.kernel_argv.append("--KernelApp.parent_appname='%s'"%self.name)
         # scrub frontend-specific flags
         for a in argv:
             if a.startswith('-') and a.lstrip('-') in notebook_flags:
                 self.kernel_argv.remove(a)
 
     def init_kernel_manager(self):
         # Don't let Qt or ZMQ swallow KeyboardInterupts.
         signal.signal(signal.SIGINT, signal.SIG_DFL)
 
         # Create a KernelManager and start a kernel.
         self.kernel_manager = KernelManager(config=self.config, log=self.log)
 
     def init_logging(self):
         super(IPythonNotebookApp, self).init_logging()
         # This prevents double log messages because tornado use a root logger that
         # self.log is a child of. The logging module dipatches log messages to a log
         # and all of its ancenstors until propagate is set to False.
         self.log.propagate = False
 
     def initialize(self, argv=None):
         super(IPythonNotebookApp, self).initialize(argv)
         self.init_kernel_manager()
         self.web_app = NotebookWebApplication(self.kernel_manager, self.log, self.kernel_argv)
         self.http_server = httpserver.HTTPServer(self.web_app)
         self.http_server.listen(self.port)
 
     def start(self):
         self.log.info("The IPython Notebook is running at: http://%s:%i" % (self.ip, self.port))
         ioloop.IOLoop.instance().start()
 
 #-----------------------------------------------------------------------------
 # Main entry point
 #-----------------------------------------------------------------------------
 
 def launch_new_instance():
     app = IPythonNotebookApp()
     app.initialize()
     app.start()
 

IPython/frontend/html/notebook/routers.py

@@ -1,57 +1,58 b''
 import uuid
 from Queue import Queue
-
+import json
 
 class ZMQStreamRouter(object):
 
-    def __init__(self, zmq_stream):
+    def __init__(self, zmq_stream, session):
         self.zmq_stream = zmq_stream
+        self.session = session
         self._clients = {}
         self.zmq_stream.on_recv(self._on_zmq_reply)
 
     def register_client(self, client):
         client_id = uuid.uuid4()
         self._clients[client_id] = client
         return client_id
 
     def unregister_client(self, client_id):
         del self._clients[client_id]
 
     def copy_clients(self, router):
         # Copy the clients of another router.
         for client_id, client in router._clients.items():
             client.router = self
             self._clients[client_id] = client
 
 
 class IOPubStreamRouter(ZMQStreamRouter):
 
     def _on_zmq_reply(self, msg_list):
         for client_id, client in self._clients.items():
             for msg in msg_list:
+                print "Got message: ", msg
                 client.write_message(msg)
 
     def forward_unicode(self, client_id, msg):
         # This is a SUB stream that we should never write to.
         pass
 
 
 class ShellStreamRouter(ZMQStreamRouter):
 
-    def __init__(self, zmq_stream):
-        ZMQStreamRouter.__init__(self, zmq_stream)
+    def __init__(self, zmq_stream, session):
+        ZMQStreamRouter.__init__(self, zmq_stream, session)
         self._request_queue = Queue()
 
     def _on_zmq_reply(self, msg_list):
         client_id = self._request_queue.get(block=False)
         client = self._clients.get(client_id)
         if client is not None:
             for msg in msg_list:
                 client.write_message(msg)
 
     def forward_unicode(self, client_id, msg):
+        print "Inbound message: ", msg
         self._request_queue.put(client_id)
-        self.zmq_stream.send_unicode(msg)
-
-
+        self.session.send(self.zmq_stream, msg)