Merge pull request #7389 from minrk/one-websocket

use single WebSocket connection for all channels

docs/source/whatsnew/pr/incompat-single-websocket.rst

@@ -0,0 +1,3 b''
+* The notebook now uses a single websocket at `/kernels/<kernel-id>/channels` instead of separate
+  `/kernels/<kernel-id>/{shell|iopub|stdin}` channels. Messages on each channel are identified by a
+  `channel` key in the message dict, for both send and recv.

IPython/html/base/zmqhandlers.py

@@ -1,260 +1,263 b''
 # coding: utf-8
 """Tornado handlers for WebSocket <-> ZMQ sockets."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import os
 import json
 import struct
 import warnings
 
 try:
     from urllib.parse import urlparse # Py 3
 except ImportError:
     from urlparse import urlparse # Py 2
 
 import tornado
 from tornado import gen, ioloop, web
 from tornado.websocket import WebSocketHandler
 
 from IPython.kernel.zmq.session import Session
 from IPython.utils.jsonutil import date_default, extract_dates
 from IPython.utils.py3compat import cast_unicode
 
 from .handlers import IPythonHandler
 
 def serialize_binary_message(msg):
     """serialize a message as a binary blob
 
     Header:
 
     4 bytes: number of msg parts (nbufs) as 32b int
     4 * nbufs bytes: offset for each buffer as integer as 32b int
 
     Offsets are from the start of the buffer, including the header.
 
     Returns
     -------
 
     The message serialized to bytes.
 
     """
     # don't modify msg or buffer list in-place
     msg = msg.copy()
     buffers = list(msg.pop('buffers'))
     bmsg = json.dumps(msg, default=date_default).encode('utf8')
     buffers.insert(0, bmsg)
     nbufs = len(buffers)
     offsets = [4 * (nbufs + 1)]
     for buf in buffers[:-1]:
         offsets.append(offsets[-1] + len(buf))
     offsets_buf = struct.pack('!' + 'I' * (nbufs + 1), nbufs, *offsets)
     buffers.insert(0, offsets_buf)
     return b''.join(buffers)
 
 
 def deserialize_binary_message(bmsg):
     """deserialize a message from a binary blog
 
     Header:
 
     4 bytes: number of msg parts (nbufs) as 32b int
     4 * nbufs bytes: offset for each buffer as integer as 32b int
 
     Offsets are from the start of the buffer, including the header.
 
     Returns
     -------
 
     message dictionary
     """
     nbufs = struct.unpack('!i', bmsg[:4])[0]
     offsets = list(struct.unpack('!' + 'I' * nbufs, bmsg[4:4*(nbufs+1)]))
     offsets.append(None)
     bufs = []
     for start, stop in zip(offsets[:-1], offsets[1:]):
         bufs.append(bmsg[start:stop])
     msg = json.loads(bufs[0].decode('utf8'))
     msg['header'] = extract_dates(msg['header'])
     msg['parent_header'] = extract_dates(msg['parent_header'])
     msg['buffers'] = bufs[1:]
     return msg
 
 # ping interval for keeping websockets alive (30 seconds)
 WS_PING_INTERVAL = 30000
 
 if os.environ.get('IPYTHON_ALLOW_DRAFT_WEBSOCKETS_FOR_PHANTOMJS', False):
     warnings.warn("""Allowing draft76 websocket connections!
     This should only be done for testing with phantomjs!""")
     from IPython.html import allow76
     WebSocketHandler = allow76.AllowDraftWebSocketHandler
     # draft 76 doesn't support ping
     WS_PING_INTERVAL = 0
 
 class ZMQStreamHandler(WebSocketHandler):
     
     def check_origin(self, origin):
         """Check Origin == Host or Access-Control-Allow-Origin.
         
         Tornado >= 4 calls this method automatically, raising 403 if it returns False.
         We call it explicitly in `open` on Tornado < 4.
         """
         if self.allow_origin == '*':
             return True
 
         host = self.request.headers.get("Host")
 
         # If no header is provided, assume we can't verify origin
         if origin is None:
             self.log.warn("Missing Origin header, rejecting WebSocket connection.")
             return False
         if host is None:
             self.log.warn("Missing Host header, rejecting WebSocket connection.")
             return False
         
         origin = origin.lower()
         origin_host = urlparse(origin).netloc
         
         # OK if origin matches host
         if origin_host == host:
             return True
         
         # Check CORS headers
         if self.allow_origin:
             allow = self.allow_origin == origin
         elif self.allow_origin_pat:
             allow = bool(self.allow_origin_pat.match(origin))
         else:
             # No CORS headers deny the request
             allow = False
         if not allow:
             self.log.warn("Blocking Cross Origin WebSocket Attempt.  Origin: %s, Host: %s",
                 origin, host,
             )
         return allow
 
     def clear_cookie(self, *args, **kwargs):
         """meaningless for websockets"""
         pass
 
-    def _reserialize_reply(self, msg_list):
+    def _reserialize_reply(self, msg_list, channel=None):
         """Reserialize a reply message using JSON.
 
         This takes the msg list from the ZMQ socket, deserializes 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.deserialize(msg_list)
+        if channel:
+            msg['channel'] = channel
         if msg['buffers']:
             buf = serialize_binary_message(msg)
             return buf
         else:
             smsg = json.dumps(msg, default=date_default)
             return cast_unicode(smsg)
 
-    def _on_zmq_reply(self, msg_list):
+    def _on_zmq_reply(self, stream, msg_list):
         # Sometimes this gets triggered when the on_close method is scheduled in the
         # eventloop but hasn't been called.
-        if self.stream.closed(): return
+        if stream.closed(): return
+        channel = getattr(stream, 'channel', None)
         try:
-            msg = self._reserialize_reply(msg_list)
+            msg = self._reserialize_reply(msg_list, channel=channel)
         except Exception:
             self.log.critical("Malformed message: %r" % msg_list, exc_info=True)
         else:
             self.write_message(msg, binary=isinstance(msg, bytes))
 
 class AuthenticatedZMQStreamHandler(ZMQStreamHandler, IPythonHandler):
     ping_callback = None
     last_ping = 0
     last_pong = 0
     
     @property
     def ping_interval(self):
         """The interval for websocket keep-alive pings.
         
         Set ws_ping_interval = 0 to disable pings.
         """
         return self.settings.get('ws_ping_interval', WS_PING_INTERVAL)
     
     @property
     def ping_timeout(self):
         """If no ping is received in this many milliseconds,
         close the websocket connection (VPNs, etc. can fail to cleanly close ws connections).
         Default is max of 3 pings or 30 seconds.
         """
         return self.settings.get('ws_ping_timeout',
             max(3 * self.ping_interval, WS_PING_INTERVAL)
         )
 
     def set_default_headers(self):
         """Undo the set_default_headers in IPythonHandler
         
         which doesn't make sense for websockets
         """
         pass
     
     def pre_get(self):
         """Run before finishing the GET request
         
         Extend this method to add logic that should fire before
         the websocket finishes completing.
         """
         # authenticate the request before opening the websocket
         if self.get_current_user() is None:
             self.log.warn("Couldn't authenticate WebSocket connection")
             raise web.HTTPError(403)
         
         if self.get_argument('session_id', False):
             self.session.session = cast_unicode(self.get_argument('session_id'))
         else:
             self.log.warn("No session ID specified")
     
     @gen.coroutine
     def get(self, *args, **kwargs):
         # pre_get can be a coroutine in subclasses
         # assign and yield in two step to avoid tornado 3 issues
         res = self.pre_get()
         yield gen.maybe_future(res)
         super(AuthenticatedZMQStreamHandler, self).get(*args, **kwargs)
     
     def initialize(self):
         self.log.debug("Initializing websocket connection %s", self.request.path)
         self.session = Session(config=self.config)
     
     def open(self, *args, **kwargs):
         self.log.debug("Opening websocket %s", self.request.path)
         
         # start the pinging
         if self.ping_interval > 0:
             loop = ioloop.IOLoop.current()
             self.last_ping = loop.time()  # Remember time of last ping
             self.last_pong = self.last_ping
             self.ping_callback = ioloop.PeriodicCallback(
                 self.send_ping, self.ping_interval, io_loop=loop,
             )
             self.ping_callback.start()
 
     def send_ping(self):
         """send a ping to keep the websocket alive"""
         if self.stream.closed() and self.ping_callback is not None:
             self.ping_callback.stop()
             return
         
         # check for timeout on pong.  Make sure that we really have sent a recent ping in
         # case the machine with both server and client has been suspended since the last ping.
         now = ioloop.IOLoop.current().time()
         since_last_pong = 1e3 * (now - self.last_pong)
         since_last_ping = 1e3 * (now - self.last_ping)
         if since_last_ping < 2*self.ping_interval and since_last_pong > self.ping_timeout:
             self.log.warn("WebSocket ping timeout after %i ms.", since_last_pong)
             self.close()
             return
 
         self.ping(b'')
         self.last_ping = now
 
     def on_pong(self, data):
         self.last_pong = ioloop.IOLoop.current().time()

IPython/html/services/kernels/handlers.py

@@ -1,295 +1,279 b''
 """Tornado handlers for kernels."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import json
 import logging
 from tornado import gen, web
 from tornado.concurrent import Future
 from tornado.ioloop import IOLoop
 
 from IPython.utils.jsonutil import date_default
 from IPython.utils.py3compat import cast_unicode
 from IPython.html.utils import url_path_join, url_escape
 
 from ...base.handlers import IPythonHandler, json_errors
 from ...base.zmqhandlers import AuthenticatedZMQStreamHandler, deserialize_binary_message
 
 from IPython.core.release import kernel_protocol_version
 
 class MainKernelHandler(IPythonHandler):
 
     @web.authenticated
     @json_errors
     def get(self):
         km = self.kernel_manager
         self.finish(json.dumps(km.list_kernels()))
 
     @web.authenticated
     @json_errors
     def post(self):
         km = self.kernel_manager
         model = self.get_json_body()
         if model is None:
             model = {
                 'name': km.default_kernel_name
             }
         else:
             model.setdefault('name', km.default_kernel_name)
 
         kernel_id = km.start_kernel(kernel_name=model['name'])
         model = km.kernel_model(kernel_id)
         location = url_path_join(self.base_url, 'api', 'kernels', kernel_id)
         self.set_header('Location', url_escape(location))
         self.set_status(201)
         self.finish(json.dumps(model))
 
 
 class KernelHandler(IPythonHandler):
 
     SUPPORTED_METHODS = ('DELETE', 'GET')
 
     @web.authenticated
     @json_errors
     def get(self, kernel_id):
         km = self.kernel_manager
         km._check_kernel_id(kernel_id)
         model = km.kernel_model(kernel_id)
         self.finish(json.dumps(model))
 
     @web.authenticated
     @json_errors
     def delete(self, kernel_id):
         km = self.kernel_manager
         km.shutdown_kernel(kernel_id)
         self.set_status(204)
         self.finish()
 
 
 class KernelActionHandler(IPythonHandler):
 
     @web.authenticated
     @json_errors
     def post(self, kernel_id, action):
         km = self.kernel_manager
         if action == 'interrupt':
             km.interrupt_kernel(kernel_id)
             self.set_status(204)
         if action == 'restart':
             km.restart_kernel(kernel_id)
             model = km.kernel_model(kernel_id)
             self.set_header('Location', '{0}api/kernels/{1}'.format(self.base_url, kernel_id))
             self.write(json.dumps(model))
         self.finish()
 
 
-class ZMQChannelHandler(AuthenticatedZMQStreamHandler):
+class ZMQChannelsHandler(AuthenticatedZMQStreamHandler):
     
     @property
     def kernel_info_timeout(self):
         return self.settings.get('kernel_info_timeout', 10)
     
     def __repr__(self):
         return "%s(%s)" % (self.__class__.__name__, getattr(self, 'kernel_id', 'uninitialized'))
     
     def create_stream(self):
         km = self.kernel_manager
-        meth = getattr(km, 'connect_%s' % self.channel)
-        self.zmq_stream = meth(self.kernel_id, identity=self.session.bsession)
+        identity = self.session.bsession
+        for channel in ('shell', 'iopub', 'stdin'):
+            meth = getattr(km, 'connect_' + channel)
+            self.channels[channel] = stream = meth(self.kernel_id, identity=identity)
+            stream.channel = channel
+        km.add_restart_callback(self.kernel_id, self.on_kernel_restarted)
+        km.add_restart_callback(self.kernel_id, self.on_restart_failed, 'dead')
     
     def request_kernel_info(self):
         """send a request for kernel_info"""
         km = self.kernel_manager
         kernel = km.get_kernel(self.kernel_id)
         try:
             # check for previous request
             future = kernel._kernel_info_future
         except AttributeError:
             self.log.debug("Requesting kernel info from %s", self.kernel_id)
             # Create a kernel_info channel to query the kernel protocol version.
             # This channel will be closed after the kernel_info reply is received.
             if self.kernel_info_channel is None:
                 self.kernel_info_channel = km.connect_shell(self.kernel_id)
             self.kernel_info_channel.on_recv(self._handle_kernel_info_reply)
             self.session.send(self.kernel_info_channel, "kernel_info_request")
             # store the future on the kernel, so only one request is sent
             kernel._kernel_info_future = self._kernel_info_future
         else:
             if not future.done():
                 self.log.debug("Waiting for pending kernel_info request")
             future.add_done_callback(lambda f: self._finish_kernel_info(f.result()))
         return self._kernel_info_future
     
     def _handle_kernel_info_reply(self, msg):
         """process the kernel_info_reply
         
         enabling msg spec adaptation, if necessary
         """
         idents,msg = self.session.feed_identities(msg)
         try:
             msg = self.session.deserialize(msg)
         except:
             self.log.error("Bad kernel_info reply", exc_info=True)
             self._kernel_info_future.set_result({})
             return
         else:
             info = msg['content']
             self.log.debug("Received kernel info: %s", info)
             if msg['msg_type'] != 'kernel_info_reply' or 'protocol_version' not in info:
                 self.log.error("Kernel info request failed, assuming current %s", info)
                 info = {}
             self._finish_kernel_info(info)
         
         # close the kernel_info channel, we don't need it anymore
         if self.kernel_info_channel:
             self.kernel_info_channel.close()
         self.kernel_info_channel = None
     
     def _finish_kernel_info(self, info):
         """Finish handling kernel_info reply
         
         Set up protocol adaptation, if needed,
         and signal that connection can continue.
         """
         protocol_version = info.get('protocol_version', kernel_protocol_version)
         if protocol_version != kernel_protocol_version:
             self.session.adapt_version = int(protocol_version.split('.')[0])
             self.log.info("Kernel %s speaks protocol %s", self.kernel_id, protocol_version)
         if not self._kernel_info_future.done():
             self._kernel_info_future.set_result(info)
     
     def initialize(self):
-        super(ZMQChannelHandler, self).initialize()
+        super(ZMQChannelsHandler, self).initialize()
         self.zmq_stream = None
+        self.channels = {}
         self.kernel_id = None
         self.kernel_info_channel = None
         self._kernel_info_future = Future()
     
     @gen.coroutine
     def pre_get(self):
         # authenticate first
-        super(ZMQChannelHandler, self).pre_get()
+        super(ZMQChannelsHandler, self).pre_get()
         # then request kernel info, waiting up to a certain time before giving up.
         # We don't want to wait forever, because browsers don't take it well when
         # servers never respond to websocket connection requests.
         future = self.request_kernel_info()
         
         def give_up():
             """Don't wait forever for the kernel to reply"""
             if future.done():
                 return
             self.log.warn("Timeout waiting for kernel_info reply from %s", self.kernel_id)
             future.set_result({})
         loop = IOLoop.current()
         loop.add_timeout(loop.time() + self.kernel_info_timeout, give_up)
         # actually wait for it
         yield future
     
     @gen.coroutine
     def get(self, kernel_id):
         self.kernel_id = cast_unicode(kernel_id, 'ascii')
-        yield super(ZMQChannelHandler, self).get(kernel_id=kernel_id)
+        yield super(ZMQChannelsHandler, self).get(kernel_id=kernel_id)
     
     def open(self, kernel_id):
-        super(ZMQChannelHandler, self).open()
+        super(ZMQChannelsHandler, self).open()
         try:
             self.create_stream()
         except web.HTTPError as e:
             self.log.error("Error opening stream: %s", e)
             # WebSockets don't response to traditional error codes so we
             # close the connection.
-            if not self.stream.closed():
-                self.stream.close()
+            for channel, stream in self.channels.items():
+                if not stream.closed():
+                    stream.close()
             self.close()
         else:
-            self.zmq_stream.on_recv(self._on_zmq_reply)
+            for channel, stream in self.channels.items():
+                stream.on_recv_stream(self._on_zmq_reply)
 
     def on_message(self, msg):
-        if self.zmq_stream is None:
-            return
-        elif self.zmq_stream.closed():
-            self.log.info("%s closed, closing websocket.", self)
-            self.close()
-            return
         if isinstance(msg, bytes):
             msg = deserialize_binary_message(msg)
         else:
             msg = json.loads(msg)
-        self.session.send(self.zmq_stream, msg)
-
-    def on_close(self):
-        # This method can be called twice, once by self.kernel_died and once
-        # from the WebSocket close event. If the WebSocket connection is
-        # closed before the ZMQ streams are setup, they could be None.
-        if self.zmq_stream is not None and not self.zmq_stream.closed():
-            self.zmq_stream.on_recv(None)
-            # close the socket directly, don't wait for the stream
-            socket = self.zmq_stream.socket
-            self.zmq_stream.close()
-            socket.close()
-
-
-class IOPubHandler(ZMQChannelHandler):
-    channel = 'iopub'
-    
-    def create_stream(self):
-        super(IOPubHandler, self).create_stream()
-        km = self.kernel_manager
-        km.add_restart_callback(self.kernel_id, self.on_kernel_restarted)
-        km.add_restart_callback(self.kernel_id, self.on_restart_failed, 'dead')
+        channel = msg.pop('channel', None)
+        if channel is None:
+            self.log.warn("No channel specified, assuming shell: %s", msg)
+            channel = 'shell'
+        stream = self.channels[channel]
+        self.session.send(stream, msg)
     
     def on_close(self):
         km = self.kernel_manager
         if self.kernel_id in km:
             km.remove_restart_callback(
                 self.kernel_id, self.on_kernel_restarted,
             )
             km.remove_restart_callback(
                 self.kernel_id, self.on_restart_failed, 'dead',
             )
-        super(IOPubHandler, self).on_close()
+        # This method can be called twice, once by self.kernel_died and once
+        # from the WebSocket close event. If the WebSocket connection is
+        # closed before the ZMQ streams are setup, they could be None.
+        for channel, stream in self.channels.items():
+            if stream is not None and not stream.closed():
+                stream.on_recv(None)
+                # close the socket directly, don't wait for the stream
+                socket = stream.socket
+                stream.close()
+                socket.close()
+        
+        self.channels = {}
     
     def _send_status_message(self, status):
         msg = self.session.msg("status",
             {'execution_state': status}
         )
+        msg['channel'] = 'iopub'
         self.write_message(json.dumps(msg, default=date_default))
 
     def on_kernel_restarted(self):
         logging.warn("kernel %s restarted", self.kernel_id)
         self._send_status_message('restarting')
 
     def on_restart_failed(self):
         logging.error("kernel %s restarted failed!", self.kernel_id)
         self._send_status_message('dead')
     
-    def on_message(self, msg):
-        """IOPub messages make no sense"""
-        pass
-
-
-class ShellHandler(ZMQChannelHandler):
-    channel = 'shell'
-
-
-class StdinHandler(ZMQChannelHandler):
-    channel = 'stdin'
-
 
 #-----------------------------------------------------------------------------
 # URL to handler mappings
 #-----------------------------------------------------------------------------
 
 
 _kernel_id_regex = r"(?P<kernel_id>\w+-\w+-\w+-\w+-\w+)"
 _kernel_action_regex = r"(?P<action>restart|interrupt)"
 
 default_handlers = [
     (r"/api/kernels", MainKernelHandler),
     (r"/api/kernels/%s" % _kernel_id_regex, KernelHandler),
     (r"/api/kernels/%s/%s" % (_kernel_id_regex, _kernel_action_regex), KernelActionHandler),
-    (r"/api/kernels/%s/iopub" % _kernel_id_regex, IOPubHandler),
-    (r"/api/kernels/%s/shell" % _kernel_id_regex, ShellHandler),
-    (r"/api/kernels/%s/stdin" % _kernel_id_regex, StdinHandler)
+    (r"/api/kernels/%s/channels" % _kernel_id_regex, ZMQChannelsHandler),
 ]

IPython/html/static/services/kernels/kernel.js

@@ -1,1074 +1,1052 b''
 // Copyright (c) IPython Development Team.
 // Distributed under the terms of the Modified BSD License.
 
 define([
     'base/js/namespace',
     'jquery',
     'base/js/utils',
     './comm',
     './serialize',
     'widgets/js/init'
 ], function(IPython, $, utils, comm, serialize, widgetmanager) {
     "use strict";
 
     /**
      * A Kernel class to communicate with the Python kernel. This
      * should generally not be constructed directly, but be created
      * by.  the `Session` object. Once created, this object should be
      * used to communicate with the kernel.
      * 
      * @class Kernel
      * @param {string} kernel_service_url - the URL to access the kernel REST api
      * @param {string} ws_url - the websockets URL
      * @param {Notebook} notebook - notebook object
      * @param {string} name - the kernel type (e.g. python3)
      */
     var Kernel = function (kernel_service_url, ws_url, notebook, name) {
         this.events = notebook.events;
 
         this.id = null;
         this.name = name;
-
-        this.channels = {
-            'shell': null,
-            'iopub': null,
-            'stdin': null
-        };
+        this.ws = null;
 
         this.kernel_service_url = kernel_service_url;
         this.kernel_url = null;
         this.ws_url = ws_url || IPython.utils.get_body_data("wsUrl");
         if (!this.ws_url) {
             // trailing 's' in https will become wss for secure web sockets
             this.ws_url = location.protocol.replace('http', 'ws') + "//" + location.host;
         }
 
         this.username = "username";
         this.session_id = utils.uuid();
         this._msg_callbacks = {};
         this.info_reply = {}; // kernel_info_reply stored here after starting
 
         if (typeof(WebSocket) !== 'undefined') {
             this.WebSocket = WebSocket;
         } else if (typeof(MozWebSocket) !== 'undefined') {
             this.WebSocket = MozWebSocket;
         } else {
             alert('Your browser does not have WebSocket support, please try Chrome, Safari or Firefox ≥ 6. Firefox 4 and 5 are also supported by you have to enable WebSockets in about:config.');
         }
         
         this.bind_events();
         this.init_iopub_handlers();
         this.comm_manager = new comm.CommManager(this);
         this.widget_manager = new widgetmanager.WidgetManager(this.comm_manager, notebook);
         
         this.last_msg_id = null;
         this.last_msg_callbacks = {};
 
         this._autorestart_attempt = 0;
         this._reconnect_attempt = 0;
         this.reconnect_limit = 7;
     };
 
     /**
      * @function _get_msg
      */
     Kernel.prototype._get_msg = function (msg_type, content, metadata, buffers) {
         var msg = {
             header : {
                 msg_id : utils.uuid(),
                 username : this.username,
                 session : this.session_id,
                 msg_type : msg_type,
                 version : "5.0"
             },
             metadata : metadata || {},
             content : content,
             buffers : buffers || [],
             parent_header : {}
         };
         return msg;
     };
 
     /**
      * @function bind_events
      */
     Kernel.prototype.bind_events = function () {
         var that = this;
         this.events.on('send_input_reply.Kernel', function(evt, data) { 
             that.send_input_reply(data);
         });
 
         var record_status = function (evt, info) {
             console.log('Kernel: ' + evt.type + ' (' + info.kernel.id + ')');
         };
 
         this.events.on('kernel_created.Kernel', record_status);
         this.events.on('kernel_reconnecting.Kernel', record_status);
         this.events.on('kernel_connected.Kernel', record_status);
         this.events.on('kernel_starting.Kernel', record_status);
         this.events.on('kernel_restarting.Kernel', record_status);
         this.events.on('kernel_autorestarting.Kernel', record_status);
         this.events.on('kernel_interrupting.Kernel', record_status);
         this.events.on('kernel_disconnected.Kernel', record_status);
         // these are commented out because they are triggered a lot, but can
         // be uncommented for debugging purposes
         //this.events.on('kernel_idle.Kernel', record_status);
         //this.events.on('kernel_busy.Kernel', record_status);
         this.events.on('kernel_ready.Kernel', record_status);
         this.events.on('kernel_killed.Kernel', record_status);
         this.events.on('kernel_dead.Kernel', record_status);
 
         this.events.on('kernel_ready.Kernel', function () {
             that._autorestart_attempt = 0;
         });
         this.events.on('kernel_connected.Kernel', function () {
             that._reconnect_attempt = 0;
         });
     };
 
     /**
      * Initialize the iopub handlers.
      *
      * @function init_iopub_handlers
      */
     Kernel.prototype.init_iopub_handlers = function () {
         var output_msg_types = ['stream', 'display_data', 'execute_result', 'error'];
         this._iopub_handlers = {};
         this.register_iopub_handler('status', $.proxy(this._handle_status_message, this));
         this.register_iopub_handler('clear_output', $.proxy(this._handle_clear_output, this));
         this.register_iopub_handler('execute_input', $.proxy(this._handle_input_message, this));
         
         for (var i=0; i < output_msg_types.length; i++) {
             this.register_iopub_handler(output_msg_types[i], $.proxy(this._handle_output_message, this));
         }
     };
 
     /**
      * GET /api/kernels
      *
      * Get the list of running kernels.
      *
      * @function list
      * @param {function} [success] - function executed on ajax success
      * @param {function} [error] - functon executed on ajax error
      */
     Kernel.prototype.list = function (success, error) {
         $.ajax(this.kernel_service_url, {
             processData: false,
             cache: false,
             type: "GET",
             dataType: "json",
             success: success,
             error: this._on_error(error)
         });
     };
 
     /**
      * POST /api/kernels
      *
      * Start a new kernel.
      *
      * In general this shouldn't be used -- the kernel should be
      * started through the session API. If you use this function and
      * are also using the session API then your session and kernel
      * WILL be out of sync!
      *
      * @function start
      * @param {params} [Object] - parameters to include in the query string
      * @param {function} [success] - function executed on ajax success
      * @param {function} [error] - functon executed on ajax error
      */
     Kernel.prototype.start = function (params, success, error) {
         var url = this.kernel_service_url;
         var qs = $.param(params || {}); // query string for sage math stuff
         if (qs !== "") {
             url = url + "?" + qs;
         }
 
         var that = this;
         var on_success = function (data, status, xhr) {
             that.events.trigger('kernel_created.Kernel', {kernel: that});
             that._kernel_created(data);
             if (success) {
                 success(data, status, xhr);
             }
         };
 
         $.ajax(url, {
             processData: false,
             cache: false,
             type: "POST",
             data: JSON.stringify({name: this.name}),
             dataType: "json",
             success: this._on_success(on_success),
             error: this._on_error(error)
         });
 
         return url;
     };
 
     /**
      * GET /api/kernels/[:kernel_id]
      *
      * Get information about the kernel.
      *
      * @function get_info
      * @param {function} [success] - function executed on ajax success
      * @param {function} [error] - functon executed on ajax error
      */
     Kernel.prototype.get_info = function (success, error) {
         $.ajax(this.kernel_url, {
             processData: false,
             cache: false,
             type: "GET",
             dataType: "json",
             success: this._on_success(success),
             error: this._on_error(error)
         });
     };
 
     /**
      * DELETE /api/kernels/[:kernel_id]
      *
      * Shutdown the kernel.
      *
      * If you are also using sessions, then this function shoul NOT be
      * used. Instead, use Session.delete. Otherwise, the session and
      * kernel WILL be out of sync.
      *
      * @function kill
      * @param {function} [success] - function executed on ajax success
      * @param {function} [error] - functon executed on ajax error
      */
     Kernel.prototype.kill = function (success, error) {
         this.events.trigger('kernel_killed.Kernel', {kernel: this});
         this._kernel_dead();
         $.ajax(this.kernel_url, {
             processData: false,
             cache: false,
             type: "DELETE",
             dataType: "json",
             success: this._on_success(success),
             error: this._on_error(error)
         });
     };
 
     /**
      * POST /api/kernels/[:kernel_id]/interrupt
      *
      * Interrupt the kernel.
      *
      * @function interrupt
      * @param {function} [success] - function executed on ajax success
      * @param {function} [error] - functon executed on ajax error
      */
     Kernel.prototype.interrupt = function (success, error) {
         this.events.trigger('kernel_interrupting.Kernel', {kernel: this});
 
         var that = this;
         var on_success = function (data, status, xhr) {
             /**
              * get kernel info so we know what state the kernel is in
              */
             that.kernel_info();
             if (success) {
                 success(data, status, xhr);
             }
         };
 
         var url = utils.url_join_encode(this.kernel_url, 'interrupt');
         $.ajax(url, {
             processData: false,
             cache: false,
             type: "POST",
             dataType: "json",
             success: this._on_success(on_success),
             error: this._on_error(error)
         });
     };
 
     Kernel.prototype.restart = function (success, error) {
         /**
          * POST /api/kernels/[:kernel_id]/restart
          *
          * Restart the kernel.
          *
          * @function interrupt
          * @param {function} [success] - function executed on ajax success
          * @param {function} [error] - functon executed on ajax error
          */
         this.events.trigger('kernel_restarting.Kernel', {kernel: this});
         this.stop_channels();
 
         var that = this;
         var on_success = function (data, status, xhr) {
             that.events.trigger('kernel_created.Kernel', {kernel: that});
             that._kernel_created(data);
             if (success) {
                 success(data, status, xhr);
             }
         };
 
         var on_error = function (xhr, status, err) {
             that.events.trigger('kernel_dead.Kernel', {kernel: that});
             that._kernel_dead();
             if (error) {
                 error(xhr, status, err);
             }
         };
 
         var url = utils.url_join_encode(this.kernel_url, 'restart');
         $.ajax(url, {
             processData: false,
             cache: false,
             type: "POST",
             dataType: "json",
             success: this._on_success(on_success),
             error: this._on_error(on_error)
         });
     };
 
     Kernel.prototype.reconnect = function () {
         /**
          * Reconnect to a disconnected kernel. This is not actually a
          * standard HTTP request, but useful function nonetheless for
          * reconnecting to the kernel if the connection is somehow lost.
          *
          * @function reconnect
          */
         if (this.is_connected()) {
             return;
         }
         this._reconnect_attempt = this._reconnect_attempt + 1;
         this.events.trigger('kernel_reconnecting.Kernel', {
             kernel: this,
             attempt: this._reconnect_attempt,
         });
         this.start_channels();
     };
 
     Kernel.prototype._on_success = function (success) {
         /**
          * Handle a successful AJAX request by updating the kernel id and
          * name from the response, and then optionally calling a provided
          * callback.
          *
          * @function _on_success
          * @param {function} success - callback
          */
         var that = this;
         return function (data, status, xhr) {
             if (data) {
                 that.id = data.id;
                 that.name = data.name;
             }
             that.kernel_url = utils.url_join_encode(that.kernel_service_url, that.id);
             if (success) {
                 success(data, status, xhr);
             }
         };
     };
 
     Kernel.prototype._on_error = function (error) {
         /**
          * Handle a failed AJAX request by logging the error message, and
          * then optionally calling a provided callback.
          *
          * @function _on_error
          * @param {function} error - callback
          */
         return function (xhr, status, err) {
             utils.log_ajax_error(xhr, status, err);
             if (error) {
                 error(xhr, status, err);
             }
         };
     };
 
     Kernel.prototype._kernel_created = function (data) {
         /**
          * Perform necessary tasks once the kernel has been started,
          * including actually connecting to the kernel.
          *
          * @function _kernel_created
          * @param {Object} data - information about the kernel including id
          */
         this.id = data.id;
         this.kernel_url = utils.url_join_encode(this.kernel_service_url, this.id);
         this.start_channels();
     };
 
     Kernel.prototype._kernel_connected = function () {
         /**
          * Perform necessary tasks once the connection to the kernel has
          * been established. This includes requesting information about
          * the kernel.
          *
          * @function _kernel_connected
          */
         this.events.trigger('kernel_connected.Kernel', {kernel: this});
         this.events.trigger('kernel_starting.Kernel', {kernel: this});
         // get kernel info so we know what state the kernel is in
         var that = this;
         this.kernel_info(function (reply) {
             that.info_reply = reply.content;
             that.events.trigger('kernel_ready.Kernel', {kernel: that});
         });
     };
 
     Kernel.prototype._kernel_dead = function () {
         /**
          * Perform necessary tasks after the kernel has died. This closing
          * communication channels to the kernel if they are still somehow
          * open.
          *
          * @function _kernel_dead
          */
         this.stop_channels();
     };
 
     Kernel.prototype.start_channels = function () {
         /**
-         * Start the `shell`and `iopub` channels.
+         * Start the websocket channels.
          * Will stop and restart them if they already exist.
          *
          * @function start_channels
          */
         var that = this;
         this.stop_channels();
         var ws_host_url = this.ws_url + this.kernel_url;
 
         console.log("Starting WebSockets:", ws_host_url);
         
-        var channel_url = function(channel) {
-            return [
+        this.ws = new this.WebSocket([
                 that.ws_url,
-                utils.url_join_encode(that.kernel_url, channel),
+                utils.url_join_encode(that.kernel_url, 'channels'),
                 "?session_id=" + that.session_id
-            ].join('');
-        };
-        this.channels.shell = new this.WebSocket(channel_url("shell"));
-        this.channels.stdin = new this.WebSocket(channel_url("stdin"));
-        this.channels.iopub = new this.WebSocket(channel_url("iopub"));
+            ].join('')
+        );
         
         var already_called_onclose = false; // only alert once
         var ws_closed_early = function(evt){
             if (already_called_onclose){
                 return;
             }
             already_called_onclose = true;
             if ( ! evt.wasClean ){
                 // If the websocket was closed early, that could mean
                 // that the kernel is actually dead. Try getting
                 // information about the kernel from the API call --
                 // if that fails, then assume the kernel is dead,
                 // otherwise just follow the typical websocket closed
                 // protocol.
                 that.get_info(function () {
                     that._ws_closed(ws_host_url, false);
                 }, function () {
                     that.events.trigger('kernel_dead.Kernel', {kernel: that});
                     that._kernel_dead();
                 });
             }
         };
         var ws_closed_late = function(evt){
             if (already_called_onclose){
                 return;
             }
             already_called_onclose = true;
             if ( ! evt.wasClean ){
                 that._ws_closed(ws_host_url, false);
             }
         };
         var ws_error = function(evt){
             if (already_called_onclose){
                 return;
             }
             already_called_onclose = true;
             that._ws_closed(ws_host_url, true);
         };
 
-        for (var c in this.channels) {
-            this.channels[c].onopen = $.proxy(this._ws_opened, this);
-            this.channels[c].onclose = ws_closed_early;
-            this.channels[c].onerror = ws_error;
-        }
+        this.ws.onopen = $.proxy(this._ws_opened, this);
+        this.ws.onclose = ws_closed_early;
+        this.ws.onerror = ws_error;
         // switch from early-close to late-close message after 1s
         setTimeout(function() {
-            for (var c in that.channels) {
-                if (that.channels[c] !== null) {
-                    that.channels[c].onclose = ws_closed_late;
-                }
+            if (that.ws !== null) {
+                that.ws.onclose = ws_closed_late;
             }
         }, 1000);
-        this.channels.shell.onmessage = $.proxy(this._handle_shell_reply, this);
-        this.channels.iopub.onmessage = $.proxy(this._handle_iopub_message, this);
-        this.channels.stdin.onmessage = $.proxy(this._handle_input_request, this);
+        this.ws.onmessage = $.proxy(this._handle_ws_message, this);
     };
 
     Kernel.prototype._ws_opened = function (evt) {
         /**
          * Handle a websocket entering the open state,
-         * signaling that the kernel is connected when all channels are open.
+         * signaling that the kernel is connected when websocket is open.
          *
          * @function _ws_opened
          */
         if (this.is_connected()) {
             // all events ready, trigger started event.
             this._kernel_connected();
         }
     };
 
     Kernel.prototype._ws_closed = function(ws_url, error) {
         /**
-         * Handle a websocket entering the closed state. This closes the
-         * other communication channels if they are open. If the websocket
+         * Handle a websocket entering the closed state.  If the websocket
          * was not closed due to an error, try to reconnect to the kernel.
          *
          * @function _ws_closed
          * @param {string} ws_url - the websocket url
          * @param {bool} error - whether the connection was closed due to an error
          */
         this.stop_channels();
 
         this.events.trigger('kernel_disconnected.Kernel', {kernel: this});
         if (error) {
             console.log('WebSocket connection failed: ', ws_url);
             this.events.trigger('kernel_connection_failed.Kernel', {kernel: this, ws_url: ws_url, attempt: this._reconnect_attempt});
         }
         this._schedule_reconnect();
     };
     
     Kernel.prototype._schedule_reconnect = function () {
         /**
          * function to call when kernel connection is lost
          * schedules reconnect, or fires 'connection_dead' if reconnect limit is hit
          */
         if (this._reconnect_attempt < this.reconnect_limit) {
             var timeout = Math.pow(2, this._reconnect_attempt);
             console.log("Connection lost, reconnecting in " + timeout + " seconds.");
             setTimeout($.proxy(this.reconnect, this), 1e3 * timeout);
         } else {
             this.events.trigger('kernel_connection_dead.Kernel', {
                 kernel: this,
                 reconnect_attempt: this._reconnect_attempt,
             });
             console.log("Failed to reconnect, giving up.");
         }
     };
     
     Kernel.prototype.stop_channels = function () {
         /**
-         * Close the websocket channels. After successful close, the value
-         * in `this.channels[channel_name]` will be null.
+         * Close the websocket. After successful close, the value
+         * in `this.ws` will be null.
          *
          * @function stop_channels
          */
         var that = this;
-        var close = function (c) {
-            return function () {
-                if (that.channels[c] && that.channels[c].readyState === WebSocket.CLOSED) {
-                    that.channels[c] = null;
+        var close = function () {
+            if (that.ws && that.ws.readyState === WebSocket.CLOSED) {
+                that.ws = null;
                 }
             };
-        };
-        for (var c in this.channels) {
-            if ( this.channels[c] !== null ) {
-                if (this.channels[c].readyState === WebSocket.OPEN) {
-                    this.channels[c].onclose = close(c);
-                    this.channels[c].close();
+        if (this.ws !== null) {
+            if (this.ws.readyState === WebSocket.OPEN) {
+                this.ws.onclose = close;
+                this.ws.close();
                 } else {
-                    close(c)();
-                }
+                close();
             }
         }
     };
 
     Kernel.prototype.is_connected = function () {
         /**
          * Check whether there is a connection to the kernel. This
-         * function only returns true if all channel objects have been
-         * created and have a state of WebSocket.OPEN.
+         * function only returns true if websocket has been
+         * created and has a state of WebSocket.OPEN.
          *
          * @function is_connected
          * @returns {bool} - whether there is a connection
          */
-        for (var c in this.channels) {
             // if any channel is not ready, then we're not connected
-            if (this.channels[c] === null) {
+        if (this.ws === null) {
                 return false;
             }
-            if (this.channels[c].readyState !== WebSocket.OPEN) {
+        if (this.ws.readyState !== WebSocket.OPEN) {
                 return false;
             }
-        }
         return true;
     };
 
     Kernel.prototype.is_fully_disconnected = function () {
         /**
          * Check whether the connection to the kernel has been completely
          * severed. This function only returns true if all channel objects
          * are null.
          *
          * @function is_fully_disconnected
          * @returns {bool} - whether the kernel is fully disconnected
          */
-        for (var c in this.channels) {
-            if (this.channels[c] === null) {
-                return true;
-            }
-        }
-        return false;
+        return (this.ws === null);
     };
     
     Kernel.prototype.send_shell_message = function (msg_type, content, callbacks, metadata, buffers) {
         /**
          * Send a message on the Kernel's shell channel
          *
          * @function send_shell_message
          */
         if (!this.is_connected()) {
             throw new Error("kernel is not connected");
         }
         var msg = this._get_msg(msg_type, content, metadata, buffers);
-        this.channels.shell.send(serialize.serialize(msg));
+        msg.channel = 'shell';
+        this.ws.send(serialize.serialize(msg));
         this.set_callbacks_for_msg(msg.header.msg_id, callbacks);
         return msg.header.msg_id;
     };
 
     Kernel.prototype.kernel_info = function (callback) {
         /**
          * Get kernel info
          *
          * @function kernel_info
          * @param callback {function}
          *
          * When calling this method, pass a callback function that expects one argument.
          * The callback will be passed the complete `kernel_info_reply` message documented
          * [here](http://ipython.org/ipython-doc/dev/development/messaging.html#kernel-info)
          */
         var callbacks;
         if (callback) {
             callbacks = { shell : { reply : callback } };
         }
         return this.send_shell_message("kernel_info_request", {}, callbacks);
     };
 
     Kernel.prototype.inspect = function (code, cursor_pos, callback) {
         /**
          * Get info on an object
          *
          * When calling this method, pass a callback function that expects one argument.
          * The callback will be passed the complete `inspect_reply` message documented
          * [here](http://ipython.org/ipython-doc/dev/development/messaging.html#object-information)
          *
          * @function inspect
          * @param code {string}
          * @param cursor_pos {integer}
          * @param callback {function}
          */
         var callbacks;
         if (callback) {
             callbacks = { shell : { reply : callback } };
         }
         
         var content = {
             code : code,
             cursor_pos : cursor_pos,
             detail_level : 0
         };
         return this.send_shell_message("inspect_request", content, callbacks);
     };
 
     Kernel.prototype.execute = function (code, callbacks, options) {
         /**
          * Execute given code into kernel, and pass result to callback.
          *
          * @async
          * @function execute
          * @param {string} code
          * @param [callbacks] {Object} With the following keys (all optional)
          *      @param callbacks.shell.reply {function}
          *      @param callbacks.shell.payload.[payload_name] {function}
          *      @param callbacks.iopub.output {function}
          *      @param callbacks.iopub.clear_output {function}
          *      @param callbacks.input {function}
          * @param {object} [options]
          *      @param [options.silent=false] {Boolean}
          *      @param [options.user_expressions=empty_dict] {Dict}
          *      @param [options.allow_stdin=false] {Boolean} true|false
          *
          * @example
          *
          * The options object should contain the options for the execute
          * call. Its default values are:
          *
          *      options = {
          *        silent : true,
          *        user_expressions : {},
          *        allow_stdin : false
          *      }
          *
          * When calling this method pass a callbacks structure of the
          * form:
          *
          *      callbacks = {
          *       shell : {
          *         reply : execute_reply_callback,
          *         payload : {
          *           set_next_input : set_next_input_callback,
          *         }
          *       },
          *       iopub : {
          *         output : output_callback,
          *         clear_output : clear_output_callback,
          *       },
          *       input : raw_input_callback
          *      }
          *
          * Each callback will be passed the entire message as a single
          * arugment.  Payload handlers will be passed the corresponding
          * payload and the execute_reply message.
          */
         var content = {
             code : code,
             silent : true,
             store_history : false,
             user_expressions : {},
             allow_stdin : false
         };
         callbacks = callbacks || {};
         if (callbacks.input !== undefined) {
             content.allow_stdin = true;
         }
         $.extend(true, content, options);
         this.events.trigger('execution_request.Kernel', {kernel: this, content: content});
         return this.send_shell_message("execute_request", content, callbacks);
     };
 
     /**
      * When calling this method, pass a function to be called with the
      * `complete_reply` message as its only argument when it arrives.
      *
      * `complete_reply` is documented
      * [here](http://ipython.org/ipython-doc/dev/development/messaging.html#complete)
      *
      * @function complete
      * @param code {string}
      * @param cursor_pos {integer}
      * @param callback {function}
      */
     Kernel.prototype.complete = function (code, cursor_pos, callback) {
         var callbacks;
         if (callback) {
             callbacks = { shell : { reply : callback } };
         }
         var content = {
             code : code,
             cursor_pos : cursor_pos
         };
         return this.send_shell_message("complete_request", content, callbacks);
     };
 
     /**
      * @function send_input_reply
      */
     Kernel.prototype.send_input_reply = function (input) {
         if (!this.is_connected()) {
             throw new Error("kernel is not connected");
         }
         var content = {
             value : input
         };
         this.events.trigger('input_reply.Kernel', {kernel: this, content: content});
         var msg = this._get_msg("input_reply", content);
-        this.channels.stdin.send(serialize.serialize(msg));
+        msg.channel = 'stdin';
+        this.ws.send(serialize.serialize(msg));
         return msg.header.msg_id;
     };
 
     /**
      * @function register_iopub_handler
      */
     Kernel.prototype.register_iopub_handler = function (msg_type, callback) {
         this._iopub_handlers[msg_type] = callback;
     };
 
     /**
      * Get the iopub handler for a specific message type.
      *
      * @function get_iopub_handler
      */
     Kernel.prototype.get_iopub_handler = function (msg_type) {
         return this._iopub_handlers[msg_type];
     };
 
     /**
      * Get callbacks for a specific message.
      *
      * @function get_callbacks_for_msg
      */
     Kernel.prototype.get_callbacks_for_msg = function (msg_id) {
         if (msg_id == this.last_msg_id) {
             return this.last_msg_callbacks;
         } else {
             return this._msg_callbacks[msg_id];
         }
     };
 
     /**
      * Clear callbacks for a specific message.
      *
      * @function clear_callbacks_for_msg
      */
     Kernel.prototype.clear_callbacks_for_msg = function (msg_id) {
         if (this._msg_callbacks[msg_id] !== undefined ) {
             delete this._msg_callbacks[msg_id];
         }
     };
     
     /**
      * @function _finish_shell
      */
     Kernel.prototype._finish_shell = function (msg_id) {
         var callbacks = this._msg_callbacks[msg_id];
         if (callbacks !== undefined) {
             callbacks.shell_done = true;
             if (callbacks.iopub_done) {
                 this.clear_callbacks_for_msg(msg_id);
             }
         }
     };
 
     /**
      * @function _finish_iopub
      */
     Kernel.prototype._finish_iopub = function (msg_id) {
         var callbacks = this._msg_callbacks[msg_id];
         if (callbacks !== undefined) {
             callbacks.iopub_done = true;
             if (callbacks.shell_done) {
                 this.clear_callbacks_for_msg(msg_id);
             }
         }
     };
     
     /**
      * Set callbacks for a particular message.
      * Callbacks should be a struct of the following form:
      * shell : {
      * 
      * }
      *
      * @function set_callbacks_for_msg
      */
     Kernel.prototype.set_callbacks_for_msg = function (msg_id, callbacks) {
         this.last_msg_id = msg_id;
         if (callbacks) {
             // shallow-copy mapping, because we will modify it at the top level
             var cbcopy = this._msg_callbacks[msg_id] = this.last_msg_callbacks = {};
             cbcopy.shell = callbacks.shell;
             cbcopy.iopub = callbacks.iopub;
             cbcopy.input = callbacks.input;
             cbcopy.shell_done = (!callbacks.shell);
             cbcopy.iopub_done = (!callbacks.iopub);
         } else {
             this.last_msg_callbacks = {};
         }
     };
 
-    /**
-     * @function _handle_shell_reply
-     */
-    Kernel.prototype._handle_shell_reply = function (e) {
-        serialize.deserialize(e.data, $.proxy(this._finish_shell_reply, this));
+    Kernel.prototype._handle_ws_message = function (e) {
+        serialize.deserialize(e.data, $.proxy(this._finish_ws_message, this));
+    };
+
+    Kernel.prototype._finish_ws_message = function (msg) {
+        switch (msg.channel) {
+            case 'shell':
+                this._handle_shell_reply(msg);
+                break;
+            case 'iopub':
+                this._handle_iopub_message(msg);
+                break;
+            case 'stdin':
+                this._handle_input_request(msg);
+                break;
+            default:
+                console.error("unrecognized message channel", msg.channel, msg);
+        }
     };
 
-    Kernel.prototype._finish_shell_reply = function (reply) {
+    Kernel.prototype._handle_shell_reply = function (reply) {
         this.events.trigger('shell_reply.Kernel', {kernel: this, reply:reply});
         var content = reply.content;
         var metadata = reply.metadata;
         var parent_id = reply.parent_header.msg_id;
         var callbacks = this.get_callbacks_for_msg(parent_id);
         if (!callbacks || !callbacks.shell) {
             return;
         }
         var shell_callbacks = callbacks.shell;
         
         // signal that shell callbacks are done
         this._finish_shell(parent_id);
         
         if (shell_callbacks.reply !== undefined) {
             shell_callbacks.reply(reply);
         }
         if (content.payload && shell_callbacks.payload) {
             this._handle_payloads(content.payload, shell_callbacks.payload, reply);
         }
     };
 
     /**
      * @function _handle_payloads
      */
     Kernel.prototype._handle_payloads = function (payloads, payload_callbacks, msg) {
         var l = payloads.length;
         // Payloads are handled by triggering events because we don't want the Kernel
         // to depend on the Notebook or Pager classes.
         for (var i=0; i<l; i++) {
             var payload = payloads[i];
             var callback = payload_callbacks[payload.source];
             if (callback) {
                 callback(payload, msg);
             }
         }
     };
 
     /**
      * @function _handle_status_message
      */
     Kernel.prototype._handle_status_message = function (msg) {
         var execution_state = msg.content.execution_state;
         var parent_id = msg.parent_header.msg_id;
         
         // dispatch status msg callbacks, if any
         var callbacks = this.get_callbacks_for_msg(parent_id);
         if (callbacks && callbacks.iopub && callbacks.iopub.status) {
             try {
                 callbacks.iopub.status(msg);
             } catch (e) {
                 console.log("Exception in status msg handler", e, e.stack);
             }
         }
         
         if (execution_state === 'busy') {
             this.events.trigger('kernel_busy.Kernel', {kernel: this});
 
         } else if (execution_state === 'idle') {
             // signal that iopub callbacks are (probably) done
             // async output may still arrive,
             // but only for the most recent request
             this._finish_iopub(parent_id);
             
             // trigger status_idle event
             this.events.trigger('kernel_idle.Kernel', {kernel: this});
 
         } else if (execution_state === 'starting') {
             this.events.trigger('kernel_starting.Kernel', {kernel: this});
             var that = this;
             this.kernel_info(function (reply) {
                 that.info_reply = reply.content;
                 that.events.trigger('kernel_ready.Kernel', {kernel: that});
             });
 
         } else if (execution_state === 'restarting') {
             // autorestarting is distinct from restarting,
             // in that it means the kernel died and the server is restarting it.
             // kernel_restarting sets the notification widget,
             // autorestart shows the more prominent dialog.
             this._autorestart_attempt = this._autorestart_attempt + 1;
             this.events.trigger('kernel_restarting.Kernel', {kernel: this});
             this.events.trigger('kernel_autorestarting.Kernel', {kernel: this, attempt: this._autorestart_attempt});
 
         } else if (execution_state === 'dead') {
             this.events.trigger('kernel_dead.Kernel', {kernel: this});
             this._kernel_dead();
         }
     };
     
     /**
      * Handle clear_output message
      *
      * @function _handle_clear_output
      */
     Kernel.prototype._handle_clear_output = function (msg) {
         var callbacks = this.get_callbacks_for_msg(msg.parent_header.msg_id);
         if (!callbacks || !callbacks.iopub) {
             return;
         }
         var callback = callbacks.iopub.clear_output;
         if (callback) {
             callback(msg);
         }
     };
 
     /**
      * handle an output message (execute_result, display_data, etc.)
      *
      * @function _handle_output_message
      */
     Kernel.prototype._handle_output_message = function (msg) {
         var callbacks = this.get_callbacks_for_msg(msg.parent_header.msg_id);
         if (!callbacks || !callbacks.iopub) {
             // The message came from another client. Let the UI decide what to
             // do with it.
             this.events.trigger('received_unsolicited_message.Kernel', msg);
             return;
         }
         var callback = callbacks.iopub.output;
         if (callback) {
             callback(msg);
         }
     };
 
     /**
      * Handle an input message (execute_input).
      *
      * @function _handle_input message
      */
     Kernel.prototype._handle_input_message = function (msg) {
         var callbacks = this.get_callbacks_for_msg(msg.parent_header.msg_id);
         if (!callbacks) {
             // The message came from another client. Let the UI decide what to
             // do with it.
             this.events.trigger('received_unsolicited_message.Kernel', msg);
         }
     };
 
     /**
      * Dispatch IOPub messages to respective handlers. Each message
      * type should have a handler.
      *
      * @function _handle_iopub_message
      */
-    Kernel.prototype._handle_iopub_message = function (e) {
-        serialize.deserialize(e.data, $.proxy(this._finish_iopub_message, this));
-    };
-
-
-    Kernel.prototype._finish_iopub_message = function (msg) {
+    Kernel.prototype._handle_iopub_message = function (msg) {
         var handler = this.get_iopub_handler(msg.header.msg_type);
         if (handler !== undefined) {
             handler(msg);
         }
     };
 
     /**
      * @function _handle_input_request
      */
-    Kernel.prototype._handle_input_request = function (e) {
-        serialize.deserialize(e.data, $.proxy(this._finish_input_request, this));
-    };
-
-
-    Kernel.prototype._finish_input_request = function (request) {
+    Kernel.prototype._handle_input_request = function (request) {
         var header = request.header;
         var content = request.content;
         var metadata = request.metadata;
         var msg_type = header.msg_type;
         if (msg_type !== 'input_request') {
             console.log("Invalid input request!", request);
             return;
         }
         var callbacks = this.get_callbacks_for_msg(request.parent_header.msg_id);
         if (callbacks) {
             if (callbacks.input) {
                 callbacks.input(request);
             }
         }
     };
 
     // Backwards compatability.
     IPython.Kernel = Kernel;
 
     return {'Kernel': Kernel};
 });