all sends/recvs now via Session.send/recv.

This allows changes to protocols/patterns
to happen in a single location.

IPython/zmq/completer.py

@@ -1,86 +1,87 b''
 """Tab-completion over zmq"""
 
 # Trying to get print statements to work during completion, not very
 # successfully...
 from __future__ import print_function
 
 import itertools
 import readline
 import rlcompleter
 import time
 
 import session
 
 class KernelCompleter(object):
     """Kernel-side completion machinery."""
     def __init__(self, namespace):
         self.namespace = namespace
         self.completer = rlcompleter.Completer(namespace)
 
     def complete(self, line, text):
         # We'll likely use linel later even if now it's not used for anything
         matches = []
         complete = self.completer.complete
         for state in itertools.count():
             comp = complete(text, state)
             if comp is None:
                 break
             matches.append(comp)
         return matches
     
 
 class ClientCompleter(object):
     """Client-side completion machinery.
 
     How it works: self.complete will be called multiple times, with
     state=0,1,2,... When state=0 it should compute ALL the completion matches,
     and then return them for each value of state."""
     
     def __init__(self, client, session, socket):
          # ugly, but we get called asynchronously and need access to some
          # client state, like backgrounded code
         self.client = client
         self.session = session
         self.socket = socket
         self.matches = []
 
     def request_completion(self, text):
         # Get full line to give to the kernel in case it wants more info.
         line = readline.get_line_buffer()
         # send completion request to kernel
         msg = self.session.send(self.socket,
                                 'complete_request',
                                 dict(text=text, line=line))
 
         # Give the kernel up to 0.5s to respond
         for i in range(5):
-            rep = self.session.recv(self.socket)
+            ident,rep = self.session.recv(self.socket)
+            rep = Message(rep)
             if rep is not None and rep.msg_type == 'complete_reply':
                 matches = rep.content.matches
                 break
             time.sleep(0.1)
         else:
             # timeout
             print ('TIMEOUT')  # Can't see this message...
             matches = None
         return matches
     
     def complete(self, text, state):
         
         if self.client.backgrounded > 0:
             print("\n[Not completing, background tasks active]")
             print(readline.get_line_buffer(), end='')
             return None
         
         if state==0:
             matches = self.request_completion(text)
             if matches is None:
                 self.matches = []
                 print('WARNING: Kernel timeout on tab completion.')
             else:
                 self.matches = matches
 
         try:
             return self.matches[state]
         except IndexError:
             return None

IPython/zmq/displayhook.py

@@ -1,22 +1,21 b''
 import __builtin__
 
 from session import extract_header
 
 class DisplayHook(object):
 
     def __init__(self, session, pub_socket):
         self.session = session
         self.pub_socket = pub_socket
         self.parent_header = {}
 
     def __call__(self, obj):
         if obj is None:
             return
 
         __builtin__._ = obj
-        msg = self.session.msg(u'pyout', {u'data':repr(obj)},
+        msg = self.session.send(self.pub_socket, u'pyout', {u'data':repr(obj)},
                                parent=self.parent_header)
-        self.pub_socket.send_json(msg)
 
     def set_parent(self, parent):
         self.parent_header = extract_header(parent)
  No newline at end of file

IPython/zmq/frontend.py

@@ -1,195 +1,196 b''
 #!/usr/bin/env python
 """A simple interactive frontend that talks to a kernel over 0MQ.
 """
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 # stdlib
 import cPickle as pickle
 import code
 import readline
 import sys
 import time
 import uuid
 
 # our own
 import zmq
 import session
 import completer
 from IPython.utils.localinterfaces import LOCALHOST
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 class Console(code.InteractiveConsole):
 
     def __init__(self, locals=None, filename="<console>",
                  session = session,
                  request_socket=None,
                  sub_socket=None):
         code.InteractiveConsole.__init__(self, locals, filename)
         self.session = session
         self.request_socket = request_socket
         self.sub_socket = sub_socket
         self.backgrounded = 0
         self.messages = {}
 
         # Set tab completion
         self.completer = completer.ClientCompleter(self, session, request_socket)
         readline.parse_and_bind('tab: complete')
         readline.parse_and_bind('set show-all-if-ambiguous on')
         readline.set_completer(self.completer.complete)
 
         # Set system prompts
         sys.ps1 = 'Py>>> '
         sys.ps2 = '  ... '
         sys.ps3 = 'Out : '
         # Build dict of handlers for message types
         self.handlers = {}
         for msg_type in ['pyin', 'pyout', 'pyerr', 'stream']:
             self.handlers[msg_type] = getattr(self, 'handle_%s' % msg_type)
 
     def handle_pyin(self, omsg):
         if omsg.parent_header.session == self.session.session:
             return
         c = omsg.content.code.rstrip()
         if c:
             print '[IN from %s]' % omsg.parent_header.username
             print c
 
     def handle_pyout(self, omsg):
         #print omsg # dbg
         if omsg.parent_header.session == self.session.session:
             print "%s%s" % (sys.ps3, omsg.content.data)
         else:
             print '[Out from %s]' % omsg.parent_header.username
             print omsg.content.data
 
     def print_pyerr(self, err):
         print >> sys.stderr, err.etype,':', err.evalue
         print >> sys.stderr, ''.join(err.traceback)       
 
     def handle_pyerr(self, omsg):
         if omsg.parent_header.session == self.session.session:
             return
         print >> sys.stderr, '[ERR from %s]' % omsg.parent_header.username
         self.print_pyerr(omsg.content)
         
     def handle_stream(self, omsg):
         if omsg.content.name == 'stdout':
             outstream = sys.stdout
         else:
             outstream = sys.stderr
             print >> outstream, '*ERR*',
         print >> outstream, omsg.content.data,
 
     def handle_output(self, omsg):
         handler = self.handlers.get(omsg.msg_type, None)
         if handler is not None:
             handler(omsg)
 
     def recv_output(self):
         while True:
-            omsg = self.session.recv(self.sub_socket)
-            if omsg is None:
+            ident,msg = self.session.recv(self.sub_socket)
+            if msg is None:
                 break
-            self.handle_output(omsg)
+            self.handle_output(Message(msg))
 
     def handle_reply(self, rep):
         # Handle any side effects on output channels
         self.recv_output()
         # Now, dispatch on the possible reply types we must handle
         if rep is None:
             return
         if rep.content.status == 'error':
             self.print_pyerr(rep.content)            
         elif rep.content.status == 'aborted':
             print >> sys.stderr, "ERROR: ABORTED"
             ab = self.messages[rep.parent_header.msg_id].content
             if 'code' in ab:
                 print >> sys.stderr, ab.code
             else:
                 print >> sys.stderr, ab
 
     def recv_reply(self):
-        rep = self.session.recv(self.request_socket)
-        self.handle_reply(rep)
-        return rep
+        ident,rep = self.session.recv(self.request_socket)
+        mrep = Message(rep)
+        self.handle_reply(mrep)
+        return mrep
 
     def runcode(self, code):
         # We can't pickle code objects, so fetch the actual source
         src = '\n'.join(self.buffer)
 
         # for non-background inputs, if we do have previoiusly backgrounded
         # jobs, check to see if they've produced results
         if not src.endswith(';'):
             while self.backgrounded > 0:
                 #print 'checking background'
                 rep = self.recv_reply()
                 if rep:
                     self.backgrounded -= 1
                 time.sleep(0.05)
 
         # Send code execution message to kernel
         omsg = self.session.send(self.request_socket,
                                  'execute_request', dict(code=src))
         self.messages[omsg.header.msg_id] = omsg
         
         # Fake asynchronicity by letting the user put ';' at the end of the line
         if src.endswith(';'):
             self.backgrounded += 1
             return
 
         # For foreground jobs, wait for reply
         while True:
             rep = self.recv_reply()
             if rep is not None:
                 break
             self.recv_output()
             time.sleep(0.05)
         else:
             # We exited without hearing back from the kernel!
             print >> sys.stderr, 'ERROR!!! kernel never got back to us!!!'
 
 
 class InteractiveClient(object):
     def __init__(self, session, request_socket, sub_socket):
         self.session = session
         self.request_socket = request_socket
         self.sub_socket = sub_socket
         self.console = Console(None, '<zmq-console>',
                                session, request_socket, sub_socket)
         
     def interact(self):
         self.console.interact()
 
 
 def main():
     # Defaults
     #ip = '192.168.2.109'
     ip = LOCALHOST
     #ip = '99.146.222.252'
     port_base = 5575
     connection = ('tcp://%s' % ip) + ':%i'
     req_conn = connection % port_base
     sub_conn = connection % (port_base+1)
     
     # Create initial sockets
     c = zmq.Context()
     request_socket = c.socket(zmq.XREQ)
     request_socket.connect(req_conn)
     
     sub_socket = c.socket(zmq.SUB)
     sub_socket.connect(sub_conn)
     sub_socket.setsockopt(zmq.SUBSCRIBE, '')
 
     # Make session and user-facing client
     sess = session.Session()
     client = InteractiveClient(sess, request_socket, sub_socket)
     client.interact()
 
 
 if __name__ == '__main__':
     main()

IPython/zmq/iostream.py

@@ -1,85 +1,84 b''
 import sys
 import time
 from cStringIO import StringIO
 
 from session import extract_header, Message
 
 from IPython.utils import io
 
 #-----------------------------------------------------------------------------
 # Stream classes
 #-----------------------------------------------------------------------------
 
 class OutStream(object):
     """A file like object that publishes the stream to a 0MQ PUB socket."""
 
     # The time interval between automatic flushes, in seconds.
     flush_interval = 0.05
 
     def __init__(self, session, pub_socket, name):
         self.session = session
         self.pub_socket = pub_socket
         self.name = name
         self.parent_header = {}
         self._new_buffer()
 
     def set_parent(self, parent):
         self.parent_header = extract_header(parent)
 
     def close(self):
         self.pub_socket = None
 
     def flush(self):
         #io.rprint('>>>flushing output buffer: %s<<<' % self.name)  # dbg
         if self.pub_socket is None:
             raise ValueError(u'I/O operation on closed file')
         else:
             data = self._buffer.getvalue()
             if data:
                 content = {u'name':self.name, u'data':data}
-                msg = self.session.msg(u'stream', content=content,
+                msg = self.session.send(self.pub_socket, u'stream', content=content,
                                        parent=self.parent_header)
                 io.raw_print(msg)
-                self.pub_socket.send_json(msg)
                 
                 self._buffer.close()
                 self._new_buffer()
 
     def isatty(self):
         return False
 
     def next(self):
         raise IOError('Read not supported on a write only stream.')
 
     def read(self, size=-1):
         raise IOError('Read not supported on a write only stream.')
 
     def readline(self, size=-1):
         raise IOError('Read not supported on a write only stream.')
 
     def write(self, string):
         if self.pub_socket is None:
             raise ValueError('I/O operation on closed file')
         else:
             # We can only send raw bytes, not unicode objects, so we encode
             # into utf-8 for all frontends if we get unicode inputs.
             if type(string) == unicode:
                 string = string.encode('utf-8')
                 
             self._buffer.write(string)
             current_time = time.time()
             if self._start <= 0:
                 self._start = current_time
             elif current_time - self._start > self.flush_interval:
                 self.flush()
 
     def writelines(self, sequence):
         if self.pub_socket is None:
             raise ValueError('I/O operation on closed file')
         else:
             for string in sequence:
                 self.write(string)
 
     def _new_buffer(self):
         self._buffer = StringIO()
         self._start = -1

IPython/zmq/ipkernel.py

@@ -1,642 +1,629 b''
 #!/usr/bin/env python
 """A simple interactive kernel that talks to a frontend over 0MQ.
 
 Things to do:
 
 * Implement `set_parent` logic. Right before doing exec, the Kernel should
   call set_parent on all the PUB objects with the message about to be executed.
 * Implement random port and security key logic.
 * Implement control messages.
 * Implement event loop and poll version.
 """
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Standard library imports.
 import __builtin__
 import atexit
 import sys
 import time
 import traceback
 
 # System library imports.
 import zmq
 
 # Local imports.
 from IPython.config.configurable import Configurable
 from IPython.utils import io
 from IPython.utils.jsonutil import json_clean
 from IPython.lib import pylabtools
 from IPython.utils.traitlets import Instance, Float
 from entry_point import (base_launch_kernel, make_argument_parser, make_kernel,
                          start_kernel)
 from iostream import OutStream
 from session import Session, Message
 from zmqshell import ZMQInteractiveShell
 
 #-----------------------------------------------------------------------------
 # Main kernel class
 #-----------------------------------------------------------------------------
 
 class Kernel(Configurable):
 
     #---------------------------------------------------------------------------
     # Kernel interface
     #---------------------------------------------------------------------------
 
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
     session = Instance(Session)
     reply_socket = Instance('zmq.Socket')
     pub_socket = Instance('zmq.Socket')
     req_socket = Instance('zmq.Socket')
 
     # Private interface
 
     # Time to sleep after flushing the stdout/err buffers in each execute
     # cycle.  While this introduces a hard limit on the minimal latency of the
     # execute cycle, it helps prevent output synchronization problems for
     # clients.
     # Units are in seconds.  The minimum zmq latency on local host is probably
     # ~150 microseconds, set this to 500us for now.  We may need to increase it
     # a little if it's not enough after more interactive testing.
     _execute_sleep = Float(0.0005, config=True)
 
     # Frequency of the kernel's event loop.
     # Units are in seconds, kernel subclasses for GUI toolkits may need to
     # adapt to milliseconds.
     _poll_interval = Float(0.05, config=True)
 
     # If the shutdown was requested over the network, we leave here the
     # necessary reply message so it can be sent by our registered atexit
     # handler.  This ensures that the reply is only sent to clients truly at
     # the end of our shutdown process (which happens after the underlying
     # IPython shell's own shutdown).
     _shutdown_message = None
 
     # This is a dict of port number that the kernel is listening on. It is set
     # by record_ports and used by connect_request.
     _recorded_ports = None
     
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
 
         # Before we even start up the shell, register *first* our exit handlers
         # so they come before the shell's
         atexit.register(self._at_shutdown)
 
         # Initialize the InteractiveShell subclass
         self.shell = ZMQInteractiveShell.instance()
         self.shell.displayhook.session = self.session
         self.shell.displayhook.pub_socket = self.pub_socket
 
         # TMP - hack while developing
         self.shell._reply_content = None
 
         # Build dict of handlers for message types
         msg_types = [ 'execute_request', 'complete_request', 
                       'object_info_request', 'history_request',
                       'connect_request', 'shutdown_request']
         self.handlers = {}
         for msg_type in msg_types:
             self.handlers[msg_type] = getattr(self, msg_type)
 
     def do_one_iteration(self):
         """Do one iteration of the kernel's evaluation loop.
         """
-        try:
-            ident = self.reply_socket.recv(zmq.NOBLOCK)
-        except zmq.ZMQError, e:
-            if e.errno == zmq.EAGAIN:
+        ident,msg = self.session.recv(self.reply_socket, zmq.NOBLOCK)
+        if msg is None:
                 return
-            else:
-                raise
+        
         # This assert will raise in versions of zeromq 2.0.7 and lesser.
         # We now require 2.0.8 or above, so we can uncomment for safety.
-        assert self.reply_socket.rcvmore(), "Missing message part."
-        msg = self.reply_socket.recv_json()
+        # print(ident,msg, file=sys.__stdout__)
+        assert ident is not None, "Missing message part."
         
         # Print some info about this message and leave a '--->' marker, so it's
         # easier to trace visually the message chain when debugging.  Each
         # handler prints its message at the end.
         # Eventually we'll move these from stdout to a logger.
         io.raw_print('\n*** MESSAGE TYPE:', msg['msg_type'], '***')
         io.raw_print('   Content: ', msg['content'],
                      '\n   --->\n   ', sep='', end='')
 
         # Find and call actual handler for message
         handler = self.handlers.get(msg['msg_type'], None)
         if handler is None:
             io.raw_print_err("UNKNOWN MESSAGE TYPE:", msg)
         else:
             handler(ident, msg)
             
         # Check whether we should exit, in case the incoming message set the
         # exit flag on
         if self.shell.exit_now:
             io.raw_print('\nExiting IPython kernel...')
             # We do a normal, clean exit, which allows any actions registered
             # via atexit (such as history saving) to take place.
             sys.exit(0)
 
 
     def start(self):
         """ Start the kernel main loop.
         """
         while True:
             time.sleep(self._poll_interval)
             self.do_one_iteration()
 
     def record_ports(self, xrep_port, pub_port, req_port, hb_port):
         """Record the ports that this kernel is using.
 
         The creator of the Kernel instance must call this methods if they
         want the :meth:`connect_request` method to return the port numbers.
         """
         self._recorded_ports = {
             'xrep_port' : xrep_port,
             'pub_port' : pub_port,
             'req_port' : req_port,
             'hb_port' : hb_port
         }
 
     #---------------------------------------------------------------------------
     # Kernel request handlers
     #---------------------------------------------------------------------------
 
     def _publish_pyin(self, code, parent):
         """Publish the code request on the pyin stream."""
 
-        pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
-        self.pub_socket.send_json(pyin_msg)
+        pyin_msg = self.session.send(self.pub_socket, u'pyin',{u'code':code}, parent=parent)
 
     def execute_request(self, ident, parent):
         
-        status_msg = self.session.msg(
+        status_msg = self.session.send(self.pub_socket,
             u'status',
             {u'execution_state':u'busy'},
             parent=parent
         )
-        self.pub_socket.send_json(status_msg)
         
         try:
             content = parent[u'content']
             code = content[u'code']
             silent = content[u'silent'] 
         except:
             io.raw_print_err("Got bad msg: ")
             io.raw_print_err(Message(parent))
             return
 
         shell = self.shell # we'll need this a lot here
 
         # Replace raw_input. Note that is not sufficient to replace 
         # raw_input in the user namespace.
         raw_input = lambda prompt='': self._raw_input(prompt, ident, parent)
         __builtin__.raw_input = raw_input
 
         # Set the parent message of the display hook and out streams.
         shell.displayhook.set_parent(parent)
         sys.stdout.set_parent(parent)
         sys.stderr.set_parent(parent)
 
         # Re-broadcast our input for the benefit of listening clients, and
         # start computing output
         if not silent:
             self._publish_pyin(code, parent)
 
         reply_content = {}
         try:
             if silent:
                 # run_code uses 'exec' mode, so no displayhook will fire, and it
                 # doesn't call logging or history manipulations.  Print
                 # statements in that code will obviously still execute.
                 shell.run_code(code)
             else:
                 # FIXME: the shell calls the exception handler itself.
                 shell._reply_content = None
                 shell.run_cell(code)
         except:
             status = u'error'
             # FIXME: this code right now isn't being used yet by default,
             # because the runlines() call above directly fires off exception
             # reporting.  This code, therefore, is only active in the scenario
             # where runlines itself has an unhandled exception.  We need to
             # uniformize this, for all exception construction to come from a
             # single location in the codbase.
             etype, evalue, tb = sys.exc_info()
             tb_list = traceback.format_exception(etype, evalue, tb)
             reply_content.update(shell._showtraceback(etype, evalue, tb_list))
         else:
             status = u'ok'
 
         reply_content[u'status'] = status
         
         # Return the execution counter so clients can display prompts
         reply_content['execution_count'] = shell.execution_count -1
 
         # FIXME - fish exception info out of shell, possibly left there by
         # runlines.  We'll need to clean up this logic later.
         if shell._reply_content is not None:
             reply_content.update(shell._reply_content)
 
         # At this point, we can tell whether the main code execution succeeded
         # or not.  If it did, we proceed to evaluate user_variables/expressions
         if reply_content['status'] == 'ok':
             reply_content[u'user_variables'] = \
                          shell.user_variables(content[u'user_variables'])
             reply_content[u'user_expressions'] = \
                          shell.user_expressions(content[u'user_expressions'])
         else:
             # If there was an error, don't even try to compute variables or
             # expressions
             reply_content[u'user_variables'] = {}
             reply_content[u'user_expressions'] = {}
 
         # Payloads should be retrieved regardless of outcome, so we can both
         # recover partial output (that could have been generated early in a
         # block, before an error) and clear the payload system always.
         reply_content[u'payload'] = shell.payload_manager.read_payload()
         # Be agressive about clearing the payload because we don't want
         # it to sit in memory until the next execute_request comes in.
         shell.payload_manager.clear_payload()
 
         # Send the reply.
-        reply_msg = self.session.msg(u'execute_reply', reply_content, parent)
+        reply_msg = self.session.send(self.reply_socket, u'execute_reply', reply_content, parent, ident=ident)
         io.raw_print(reply_msg)
 
         # Flush output before sending the reply.
         sys.stdout.flush()
         sys.stderr.flush()
         # FIXME: on rare occasions, the flush doesn't seem to make it to the
         # clients... This seems to mitigate the problem, but we definitely need
         # to better understand what's going on.
         if self._execute_sleep:
             time.sleep(self._execute_sleep)
         
-        self.reply_socket.send(ident, zmq.SNDMORE)
-        self.reply_socket.send_json(reply_msg)
         if reply_msg['content']['status'] == u'error':
             self._abort_queue()
 
-        status_msg = self.session.msg(
+        status_msg = self.session.send(self.pub_socket,
             u'status',
             {u'execution_state':u'idle'},
             parent=parent
         )
-        self.pub_socket.send_json(status_msg)
 
     def complete_request(self, ident, parent):
         txt, matches = self._complete(parent)
         matches = {'matches' : matches,
                    'matched_text' : txt,
                    'status' : 'ok'}
         completion_msg = self.session.send(self.reply_socket, 'complete_reply',
                                            matches, parent, ident)
         io.raw_print(completion_msg)
 
     def object_info_request(self, ident, parent):
         object_info = self.shell.object_inspect(parent['content']['oname'])
         # Before we send this object over, we scrub it for JSON usage
         oinfo = json_clean(object_info)
         msg = self.session.send(self.reply_socket, 'object_info_reply',
                                 oinfo, parent, ident)
         io.raw_print(msg)
 
     def history_request(self, ident, parent):
         output = parent['content']['output']
         index = parent['content']['index']
         raw = parent['content']['raw']
         hist = self.shell.get_history(index=index, raw=raw, output=output)
         content = {'history' : hist}
         msg = self.session.send(self.reply_socket, 'history_reply',
                                 content, parent, ident)
         io.raw_print(msg)
 
     def connect_request(self, ident, parent):
         if self._recorded_ports is not None:
             content = self._recorded_ports.copy()
         else:
             content = {}
         msg = self.session.send(self.reply_socket, 'connect_reply',
                                 content, parent, ident)
         io.raw_print(msg)
 
     def shutdown_request(self, ident, parent):
         self.shell.exit_now = True
         self._shutdown_message = self.session.msg(u'shutdown_reply', parent['content'], parent)
         sys.exit(0)
 
     #---------------------------------------------------------------------------
     # Protected interface
     #---------------------------------------------------------------------------
 
     def _abort_queue(self):
         while True:
-            try:
-                ident = self.reply_socket.recv(zmq.NOBLOCK)
-            except zmq.ZMQError, e:
-                if e.errno == zmq.EAGAIN:
+            ident,msg = self.session.recv(self.reply_socket, zmq.NOBLOCK)
+            if msg is None:
                     break
             else:
-                assert self.reply_socket.rcvmore(), \
+                assert ident is not None, \
                        "Unexpected missing message part."
-                msg = self.reply_socket.recv_json()
             io.raw_print("Aborting:\n", Message(msg))
             msg_type = msg['msg_type']
             reply_type = msg_type.split('_')[0] + '_reply'
-            reply_msg = self.session.msg(reply_type, {'status' : 'aborted'}, msg)
+            reply_msg = self.session.send(self.reply_socket, reply_type, 
+                    {'status' : 'aborted'}, msg, ident=ident)
             io.raw_print(reply_msg)
-            self.reply_socket.send(ident,zmq.SNDMORE)
-            self.reply_socket.send_json(reply_msg)
             # We need to wait a bit for requests to come in. This can probably
             # be set shorter for true asynchronous clients.
             time.sleep(0.1)
 
     def _raw_input(self, prompt, ident, parent):
         # Flush output before making the request.
         sys.stderr.flush()
         sys.stdout.flush()
 
         # Send the input request.
         content = dict(prompt=prompt)
-        msg = self.session.msg(u'input_request', content, parent)
-        self.req_socket.send_json(msg)
+        msg = self.session.send(self.req_socket, u'input_request', content, parent)
 
         # Await a response.
-        reply = self.req_socket.recv_json()
+        ident, reply = self.session.recv(self.req_socket, 0)
         try:
             value = reply['content']['value']
         except:
             io.raw_print_err("Got bad raw_input reply: ")
             io.raw_print_err(Message(parent))
             value = ''
         return value
     
     def _complete(self, msg):
         c = msg['content']
         try:
             cpos = int(c['cursor_pos'])
         except:
             # If we don't get something that we can convert to an integer, at
             # least attempt the completion guessing the cursor is at the end of
             # the text, if there's any, and otherwise of the line
             cpos = len(c['text'])
             if cpos==0:
                 cpos = len(c['line'])
         return self.shell.complete(c['text'], c['line'], cpos)
 
     def _object_info(self, context):
         symbol, leftover = self._symbol_from_context(context)
         if symbol is not None and not leftover:
             doc = getattr(symbol, '__doc__', '')
         else:
             doc = ''
         object_info = dict(docstring = doc)
         return object_info
 
     def _symbol_from_context(self, context):
         if not context:
             return None, context
 
         base_symbol_string = context[0]
         symbol = self.shell.user_ns.get(base_symbol_string, None)
         if symbol is None:
             symbol = __builtin__.__dict__.get(base_symbol_string, None)
         if symbol is None:
             return None, context
 
         context = context[1:]
         for i, name in enumerate(context):
             new_symbol = getattr(symbol, name, None)
             if new_symbol is None:
                 return symbol, context[i:]
             else:
                 symbol = new_symbol
 
         return symbol, []
 
     def _at_shutdown(self):
         """Actions taken at shutdown by the kernel, called by python's atexit.
         """
         # io.rprint("Kernel at_shutdown") # dbg
         if self._shutdown_message is not None:
-            self.reply_socket.send_json(self._shutdown_message)
-            self.pub_socket.send_json(self._shutdown_message)
+            self.session.send(self.reply_socket, self._shutdown_message)
+            self.session.send(self.pub_socket, self._shutdown_message)
             io.raw_print(self._shutdown_message)
             # A very short sleep to give zmq time to flush its message buffers
             # before Python truly shuts down.
             time.sleep(0.01)
 
 
 class QtKernel(Kernel):
     """A Kernel subclass with Qt support."""
 
     def start(self):
         """Start a kernel with QtPy4 event loop integration."""
 
         from PyQt4 import QtCore
         from IPython.lib.guisupport import get_app_qt4, start_event_loop_qt4
 
         self.app = get_app_qt4([" "])
         self.app.setQuitOnLastWindowClosed(False)
         self.timer = QtCore.QTimer()
         self.timer.timeout.connect(self.do_one_iteration)
         # Units for the timer are in milliseconds
         self.timer.start(1000*self._poll_interval)
         start_event_loop_qt4(self.app)
 
 
 class WxKernel(Kernel):
     """A Kernel subclass with Wx support."""
 
     def start(self):
         """Start a kernel with wx event loop support."""
 
         import wx
         from IPython.lib.guisupport import start_event_loop_wx
 
         doi = self.do_one_iteration
          # Wx uses milliseconds
         poll_interval = int(1000*self._poll_interval)
 
         # We have to put the wx.Timer in a wx.Frame for it to fire properly.
         # We make the Frame hidden when we create it in the main app below.
         class TimerFrame(wx.Frame):
             def __init__(self, func):
                 wx.Frame.__init__(self, None, -1)
                 self.timer = wx.Timer(self)
                 # Units for the timer are in milliseconds
                 self.timer.Start(poll_interval)
                 self.Bind(wx.EVT_TIMER, self.on_timer)
                 self.func = func
 
             def on_timer(self, event):
                 self.func()
 
         # We need a custom wx.App to create our Frame subclass that has the
         # wx.Timer to drive the ZMQ event loop.
         class IPWxApp(wx.App):
             def OnInit(self):
                 self.frame = TimerFrame(doi)
                 self.frame.Show(False)
                 return True
 
         # The redirect=False here makes sure that wx doesn't replace
         # sys.stdout/stderr with its own classes.
         self.app = IPWxApp(redirect=False)
         start_event_loop_wx(self.app)
 
 
 class TkKernel(Kernel):
     """A Kernel subclass with Tk support."""
 
     def start(self):
         """Start a Tk enabled event loop."""
 
         import Tkinter
         doi = self.do_one_iteration
         # Tk uses milliseconds
         poll_interval = int(1000*self._poll_interval)
         # For Tkinter, we create a Tk object and call its withdraw method.
         class Timer(object):
             def __init__(self, func):
                 self.app = Tkinter.Tk()
                 self.app.withdraw()
                 self.func = func
                 
             def on_timer(self):
                 self.func()
                 self.app.after(poll_interval, self.on_timer)
                 
             def start(self):
                 self.on_timer()  # Call it once to get things going.
                 self.app.mainloop()
 
         self.timer = Timer(doi)
         self.timer.start()
 
 
 class GTKKernel(Kernel):
     """A Kernel subclass with GTK support."""
     
     def start(self):
         """Start the kernel, coordinating with the GTK event loop"""
         from .gui.gtkembed import GTKEmbed
         
         gtk_kernel = GTKEmbed(self)
         gtk_kernel.start()
 
 
 #-----------------------------------------------------------------------------
 # Kernel main and launch functions
 #-----------------------------------------------------------------------------
 
 def launch_kernel(ip=None, xrep_port=0, pub_port=0, req_port=0, hb_port=0,
                   independent=False, pylab=False, colors=None):
     """Launches a localhost kernel, binding to the specified ports.
 
     Parameters
     ----------
     ip  : str, optional
         The ip address the kernel will bind to.
     
     xrep_port : int, optional
         The port to use for XREP channel.
 
     pub_port : int, optional
         The port to use for the SUB channel.
 
     req_port : int, optional
         The port to use for the REQ (raw input) channel.
 
     hb_port : int, optional
         The port to use for the hearbeat REP channel.
 
     independent : bool, optional (default False) 
         If set, the kernel process is guaranteed to survive if this process
         dies. If not set, an effort is made to ensure that the kernel is killed
         when this process dies. Note that in this case it is still good practice
         to kill kernels manually before exiting.
 
     pylab : bool or string, optional (default False)
         If not False, the kernel will be launched with pylab enabled. If a
         string is passed, matplotlib will use the specified backend. Otherwise,
         matplotlib's default backend will be used.
 
     colors : None or string, optional (default None)
         If not None, specify the color scheme. One of (NoColor, LightBG, Linux)
 
     Returns
     -------
     A tuple of form:
         (kernel_process, xrep_port, pub_port, req_port)
     where kernel_process is a Popen object and the ports are integers.
     """
     extra_arguments = []
     if pylab:
         extra_arguments.append('--pylab')
         if isinstance(pylab, basestring):
             extra_arguments.append(pylab)
     if ip is not None:
         extra_arguments.append('--ip')
         if isinstance(ip, basestring):
             extra_arguments.append(ip)
     if colors is not None:
         extra_arguments.append('--colors')
         extra_arguments.append(colors)
     return base_launch_kernel('from IPython.zmq.ipkernel import main; main()',
                               xrep_port, pub_port, req_port, hb_port, 
                               independent, extra_arguments)
 
 
 def main():
     """ The IPython kernel main entry point.
     """
     parser = make_argument_parser()
     parser.add_argument('--pylab', type=str, metavar='GUI', nargs='?', 
                         const='auto', help = \
 "Pre-load matplotlib and numpy for interactive use. If GUI is not \
 given, the GUI backend is matplotlib's, otherwise use one of: \
 ['tk', 'gtk', 'qt', 'wx', 'inline'].")
     parser.add_argument('--colors',
         type=str, dest='colors',
         help="Set the color scheme (NoColor, Linux, and LightBG).",
         metavar='ZMQInteractiveShell.colors')
     namespace = parser.parse_args()
 
     kernel_class = Kernel
 
     kernel_classes = {
         'qt' : QtKernel,
         'qt4': QtKernel,
         'inline': Kernel,
         'wx' : WxKernel,
         'tk' : TkKernel,
         'gtk': GTKKernel,
     }
     if namespace.pylab:
         if namespace.pylab == 'auto':
             gui, backend = pylabtools.find_gui_and_backend()
         else:
             gui, backend = pylabtools.find_gui_and_backend(namespace.pylab)
         kernel_class = kernel_classes.get(gui)
         if kernel_class is None:
             raise ValueError('GUI is not supported: %r' % gui)
         pylabtools.activate_matplotlib(backend)
     if namespace.colors:
         ZMQInteractiveShell.colors=namespace.colors
 
     kernel = make_kernel(namespace, kernel_class, OutStream)
 
     if namespace.pylab:
         pylabtools.import_pylab(kernel.shell.user_ns, backend,
                                 shell=kernel.shell)
 
     start_kernel(namespace, kernel)
 
 
 if __name__ == '__main__':
     main()

IPython/zmq/kernelmanager.py

@@ -1,906 +1,908 b''
 """Base classes to manage the interaction with a running kernel.
 
 TODO
 * Create logger to handle debugging and console messages.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2010  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Standard library imports.
 import atexit
 from Queue import Queue, Empty
 from subprocess import Popen
 import signal
 import sys
 from threading import Thread
 import time
 
 # System library imports.
 import zmq
 from zmq import POLLIN, POLLOUT, POLLERR
 from zmq.eventloop import ioloop
 
 # Local imports.
 from IPython.utils import io
 from IPython.utils.localinterfaces import LOCALHOST, LOCAL_IPS
 from IPython.utils.traitlets import HasTraits, Any, Instance, Type, TCPAddress
-from session import Session
+from session import Session, Message
 
 #-----------------------------------------------------------------------------
 # Constants and exceptions
 #-----------------------------------------------------------------------------
 
 class InvalidPortNumber(Exception):
     pass
 
 #-----------------------------------------------------------------------------
 # Utility functions
 #-----------------------------------------------------------------------------
 
 # some utilities to validate message structure, these might get moved elsewhere
 # if they prove to have more generic utility
 
 def validate_string_list(lst):
     """Validate that the input is a list of strings.
 
     Raises ValueError if not."""
     if not isinstance(lst, list):
         raise ValueError('input %r must be a list' % lst)
     for x in lst:
         if not isinstance(x, basestring):
             raise ValueError('element %r in list must be a string' % x)
 
 
 def validate_string_dict(dct):
     """Validate that the input is a dict with string keys and values.
 
     Raises ValueError if not."""
     for k,v in dct.iteritems():
         if not isinstance(k, basestring):
             raise ValueError('key %r in dict must be a string' % k)
         if not isinstance(v, basestring):
             raise ValueError('value %r in dict must be a string' % v)
 
 
 #-----------------------------------------------------------------------------
 # ZMQ Socket Channel classes
 #-----------------------------------------------------------------------------
 
 class ZmqSocketChannel(Thread):
     """The base class for the channels that use ZMQ sockets.
     """
     context = None
     session = None
     socket = None
     ioloop = None
     iostate = None
     _address = None
 
     def __init__(self, context, session, address):
         """Create a channel
 
         Parameters
         ----------
         context : :class:`zmq.Context`
             The ZMQ context to use.
         session : :class:`session.Session`
             The session to use.
         address : tuple
             Standard (ip, port) tuple that the kernel is listening on.
         """
         super(ZmqSocketChannel, self).__init__()
         self.daemon = True
 
         self.context = context
         self.session = session
         if address[1] == 0:
             message = 'The port number for a channel cannot be 0.'
             raise InvalidPortNumber(message)
         self._address = address
 
     def stop(self):
         """Stop the channel's activity.
 
         This calls :method:`Thread.join` and returns when the thread
         terminates. :class:`RuntimeError` will be raised if 
         :method:`self.start` is called again.
         """
         self.join()
 
     @property
     def address(self):
         """Get the channel's address as an (ip, port) tuple.
         
         By the default, the address is (localhost, 0), where 0 means a random
         port.
         """
         return self._address
 
     def add_io_state(self, state):
         """Add IO state to the eventloop.
 
         Parameters
         ----------
         state : zmq.POLLIN|zmq.POLLOUT|zmq.POLLERR
             The IO state flag to set.
 
         This is thread safe as it uses the thread safe IOLoop.add_callback.
         """
         def add_io_state_callback():
             if not self.iostate & state:
                 self.iostate = self.iostate | state
                 self.ioloop.update_handler(self.socket, self.iostate)
         self.ioloop.add_callback(add_io_state_callback)
 
     def drop_io_state(self, state):
         """Drop IO state from the eventloop.
 
         Parameters
         ----------
         state : zmq.POLLIN|zmq.POLLOUT|zmq.POLLERR
             The IO state flag to set.
 
         This is thread safe as it uses the thread safe IOLoop.add_callback.
         """
         def drop_io_state_callback():
             if self.iostate & state:
                 self.iostate = self.iostate & (~state)
                 self.ioloop.update_handler(self.socket, self.iostate)
         self.ioloop.add_callback(drop_io_state_callback)
 
 
 class XReqSocketChannel(ZmqSocketChannel):
     """The XREQ channel for issues request/replies to the kernel.
     """
 
     command_queue = None
 
     def __init__(self, context, session, address):
         super(XReqSocketChannel, self).__init__(context, session, address)
         self.command_queue = Queue()
         self.ioloop = ioloop.IOLoop()
 
     def run(self):
         """The thread's main activity.  Call start() instead."""
         self.socket = self.context.socket(zmq.XREQ)
         self.socket.setsockopt(zmq.IDENTITY, self.session.session)
         self.socket.connect('tcp://%s:%i' % self.address)
         self.iostate = POLLERR|POLLIN
         self.ioloop.add_handler(self.socket, self._handle_events, 
                                 self.iostate)
         self.ioloop.start()
 
     def stop(self):
         self.ioloop.stop()
         super(XReqSocketChannel, self).stop()
 
     def call_handlers(self, msg):
         """This method is called in the ioloop thread when a message arrives.
 
         Subclasses should override this method to handle incoming messages.
         It is important to remember that this method is called in the thread
         so that some logic must be done to ensure that the application leve
         handlers are called in the application thread.
         """
         raise NotImplementedError('call_handlers must be defined in a subclass.')
 
     def execute(self, code, silent=False,
                 user_variables=None, user_expressions=None):
         """Execute code in the kernel.
 
         Parameters
         ----------
         code : str
             A string of Python code.
             
         silent : bool, optional (default False)
             If set, the kernel will execute the code as quietly possible.
 
         user_variables : list, optional
             A list of variable names to pull from the user's namespace.  They
             will come back as a dict with these names as keys and their
             :func:`repr` as values.
             
         user_expressions : dict, optional
             A dict with string keys and  to pull from the user's
             namespace.  They will come back as a dict with these names as keys
             and their :func:`repr` as values.
 
         Returns
         -------
         The msg_id of the message sent.
         """
         if user_variables is None:
             user_variables = []
         if user_expressions is None:
             user_expressions = {}
             
         # Don't waste network traffic if inputs are invalid
         if not isinstance(code, basestring):
             raise ValueError('code %r must be a string' % code)
         validate_string_list(user_variables)
         validate_string_dict(user_expressions)
 
         # Create class for content/msg creation. Related to, but possibly
         # not in Session.
         content = dict(code=code, silent=silent,
                        user_variables=user_variables,
                        user_expressions=user_expressions)
         msg = self.session.msg('execute_request', content)
         self._queue_request(msg)
         return msg['header']['msg_id']
 
     def complete(self, text, line, cursor_pos, block=None):
         """Tab complete text in the kernel's namespace.
 
         Parameters
         ----------
         text : str
             The text to complete.
         line : str
             The full line of text that is the surrounding context for the 
             text to complete.
         cursor_pos : int
             The position of the cursor in the line where the completion was
             requested.
         block : str, optional
             The full block of code in which the completion is being requested.
 
         Returns
         -------
         The msg_id of the message sent.
         """
         content = dict(text=text, line=line, block=block, cursor_pos=cursor_pos)
         msg = self.session.msg('complete_request', content)
         self._queue_request(msg)
         return msg['header']['msg_id']
 
     def object_info(self, oname):
         """Get metadata information about an object.
 
         Parameters
         ----------
         oname : str
             A string specifying the object name.
         
         Returns
         -------
         The msg_id of the message sent.
         """
         content = dict(oname=oname)
         msg = self.session.msg('object_info_request', content)
         self._queue_request(msg)
         return msg['header']['msg_id']
 
     def history(self, index=None, raw=False, output=True):
         """Get the history list.
 
         Parameters
         ----------
         index : n or (n1, n2) or None
             If n, then the last entries. If a tuple, then all in
             range(n1, n2). If None, then all entries. Raises IndexError if
             the format of index is incorrect.
         raw : bool
             If True, return the raw input.
         output : bool
             If True, then return the output as well.
 
         Returns
         -------
         The msg_id of the message sent.
         """
         content = dict(index=index, raw=raw, output=output)
         msg = self.session.msg('history_request', content)
         self._queue_request(msg)
         return msg['header']['msg_id']
 
     def shutdown(self, restart=False):
         """Request an immediate kernel shutdown.
 
         Upon receipt of the (empty) reply, client code can safely assume that
         the kernel has shut down and it's safe to forcefully terminate it if
         it's still alive.
 
         The kernel will send the reply via a function registered with Python's
         atexit module, ensuring it's truly done as the kernel is done with all
         normal operation.
         """
         # Send quit message to kernel. Once we implement kernel-side setattr,
         # this should probably be done that way, but for now this will do.
         msg = self.session.msg('shutdown_request', {'restart':restart})
         self._queue_request(msg)
         return msg['header']['msg_id']
 
     def _handle_events(self, socket, events):
         if events & POLLERR:
             self._handle_err()
         if events & POLLOUT:
             self._handle_send()
         if events & POLLIN:
             self._handle_recv()
 
     def _handle_recv(self):
-        msg = self.socket.recv_json()
+        ident,msg = self.session.recv(self.socket, 0)
         self.call_handlers(msg)
 
     def _handle_send(self):
         try:
             msg = self.command_queue.get(False)
         except Empty:
             pass
         else:
-            self.socket.send_json(msg)
+            self.session.send(self.socket,msg)
         if self.command_queue.empty():
             self.drop_io_state(POLLOUT)
 
     def _handle_err(self):
         # We don't want to let this go silently, so eventually we should log.
         raise zmq.ZMQError()
 
     def _queue_request(self, msg):
         self.command_queue.put(msg)
         self.add_io_state(POLLOUT)
 
 
 class SubSocketChannel(ZmqSocketChannel):
     """The SUB channel which listens for messages that the kernel publishes.
     """
 
     def __init__(self, context, session, address):
         super(SubSocketChannel, self).__init__(context, session, address)
         self.ioloop = ioloop.IOLoop()
 
     def run(self):
         """The thread's main activity.  Call start() instead."""
         self.socket = self.context.socket(zmq.SUB)
         self.socket.setsockopt(zmq.SUBSCRIBE,'')
         self.socket.setsockopt(zmq.IDENTITY, self.session.session)
         self.socket.connect('tcp://%s:%i' % self.address)
         self.iostate = POLLIN|POLLERR
         self.ioloop.add_handler(self.socket, self._handle_events, 
                                 self.iostate)
         self.ioloop.start()
 
     def stop(self):
         self.ioloop.stop()
         super(SubSocketChannel, self).stop()
 
     def call_handlers(self, msg):
         """This method is called in the ioloop thread when a message arrives.
 
         Subclasses should override this method to handle incoming messages.
         It is important to remember that this method is called in the thread
         so that some logic must be done to ensure that the application leve
         handlers are called in the application thread.
         """
         raise NotImplementedError('call_handlers must be defined in a subclass.')
 
     def flush(self, timeout=1.0):
         """Immediately processes all pending messages on the SUB channel.
 
         Callers should use this method to ensure that :method:`call_handlers`
         has been called for all messages that have been received on the
         0MQ SUB socket of this channel.
 
         This method is thread safe.
 
         Parameters
         ----------
         timeout : float, optional
             The maximum amount of time to spend flushing, in seconds. The
             default is one second.
         """
         # We do the IOLoop callback process twice to ensure that the IOLoop
         # gets to perform at least one full poll.
         stop_time = time.time() + timeout
         for i in xrange(2):
             self._flushed = False
             self.ioloop.add_callback(self._flush)
             while not self._flushed and time.time() < stop_time:
                 time.sleep(0.01)
 
     def _handle_events(self, socket, events):
         # Turn on and off POLLOUT depending on if we have made a request
         if events & POLLERR:
             self._handle_err()
         if events & POLLIN:
             self._handle_recv()
 
     def _handle_err(self):
         # We don't want to let this go silently, so eventually we should log.
         raise zmq.ZMQError()
 
     def _handle_recv(self):
         # Get all of the messages we can
         while True:
             try:
-                msg = self.socket.recv_json(zmq.NOBLOCK)
+                ident,msg = self.session.recv(self.socket)
             except zmq.ZMQError:
                 # Check the errno?
                 # Will this trigger POLLERR?
                 break
             else:
+                if msg is None:
+                    break
                 self.call_handlers(msg)
 
     def _flush(self):
         """Callback for :method:`self.flush`."""
         self._flushed = True
 
 
 class RepSocketChannel(ZmqSocketChannel):
     """A reply channel to handle raw_input requests that the kernel makes."""
 
     msg_queue = None
 
     def __init__(self, context, session, address):
         super(RepSocketChannel, self).__init__(context, session, address)
         self.ioloop = ioloop.IOLoop()
         self.msg_queue = Queue()
 
     def run(self):
         """The thread's main activity.  Call start() instead."""
         self.socket = self.context.socket(zmq.XREQ)
         self.socket.setsockopt(zmq.IDENTITY, self.session.session)
         self.socket.connect('tcp://%s:%i' % self.address)
         self.iostate = POLLERR|POLLIN
         self.ioloop.add_handler(self.socket, self._handle_events, 
                                 self.iostate)
         self.ioloop.start()
 
     def stop(self):
         self.ioloop.stop()
         super(RepSocketChannel, self).stop()
 
     def call_handlers(self, msg):
         """This method is called in the ioloop thread when a message arrives.
 
         Subclasses should override this method to handle incoming messages.
         It is important to remember that this method is called in the thread
         so that some logic must be done to ensure that the application leve
         handlers are called in the application thread.
         """
         raise NotImplementedError('call_handlers must be defined in a subclass.')
 
     def input(self, string):
         """Send a string of raw input to the kernel."""
         content = dict(value=string)
         msg = self.session.msg('input_reply', content)
         self._queue_reply(msg)
 
     def _handle_events(self, socket, events):
         if events & POLLERR:
             self._handle_err()
         if events & POLLOUT:
             self._handle_send()
         if events & POLLIN:
             self._handle_recv()
 
     def _handle_recv(self):
-        msg = self.socket.recv_json()
+        ident,msg = self.session.recv(self.socket, 0)
         self.call_handlers(msg)
 
     def _handle_send(self):
         try:
             msg = self.msg_queue.get(False)
         except Empty:
             pass
         else:
-            self.socket.send_json(msg)
+            self.session.send(self.socket,msg)
         if self.msg_queue.empty():
             self.drop_io_state(POLLOUT)
 
     def _handle_err(self):
         # We don't want to let this go silently, so eventually we should log.
         raise zmq.ZMQError()
 
     def _queue_reply(self, msg):
         self.msg_queue.put(msg)
         self.add_io_state(POLLOUT)
 
 
 class HBSocketChannel(ZmqSocketChannel):
     """The heartbeat channel which monitors the kernel heartbeat.
 
     Note that the heartbeat channel is paused by default. As long as you start
     this channel, the kernel manager will ensure that it is paused and un-paused
     as appropriate.
     """
 
     time_to_dead = 3.0
     socket = None
     poller = None
     _running = None
     _pause = None
 
     def __init__(self, context, session, address):
         super(HBSocketChannel, self).__init__(context, session, address)
         self._running = False
         self._pause = True
 
     def _create_socket(self):
         self.socket = self.context.socket(zmq.REQ)
         self.socket.setsockopt(zmq.IDENTITY, self.session.session)
         self.socket.connect('tcp://%s:%i' % self.address)
         self.poller = zmq.Poller()
         self.poller.register(self.socket, zmq.POLLIN)
 
     def run(self):
         """The thread's main activity.  Call start() instead."""
         self._create_socket()
         self._running = True
         while self._running:
             if self._pause:
                 time.sleep(self.time_to_dead)
             else:
                 since_last_heartbeat = 0.0
                 request_time = time.time()
                 try:
                     #io.rprint('Ping from HB channel') # dbg
-                    self.socket.send_json('ping')
+                    self.socket.send(b'ping')
                 except zmq.ZMQError, e:
                     #io.rprint('*** HB Error:', e) # dbg
                     if e.errno == zmq.EFSM:
                         #io.rprint('sleep...', self.time_to_dead) # dbg
                         time.sleep(self.time_to_dead)
                         self._create_socket()
                     else:
                         raise
                 else:
                     while True:
                         try:
-                            self.socket.recv_json(zmq.NOBLOCK)
+                            self.socket.recv(zmq.NOBLOCK)
                         except zmq.ZMQError, e:
                             #io.rprint('*** HB Error 2:', e) # dbg
                             if e.errno == zmq.EAGAIN:
                                 before_poll = time.time()
                                 until_dead = self.time_to_dead - (before_poll -
                                                                   request_time)
 
                                 # When the return value of poll() is an empty
                                 # list, that is when things have gone wrong
                                 # (zeromq bug). As long as it is not an empty
                                 # list, poll is working correctly even if it
                                 # returns quickly. Note: poll timeout is in
                                 # milliseconds.
                                 self.poller.poll(1000*until_dead)
                             
                                 since_last_heartbeat = time.time()-request_time
                                 if since_last_heartbeat > self.time_to_dead:
                                     self.call_handlers(since_last_heartbeat)
                                     break
                             else:
                                 # FIXME: We should probably log this instead.
                                 raise
                         else:
                             until_dead = self.time_to_dead - (time.time() -
                                                               request_time)
                             if until_dead > 0.0:
                                 #io.rprint('sleep...', self.time_to_dead) # dbg
                                 time.sleep(until_dead)
                             break
 
     def pause(self):
         """Pause the heartbeat."""
         self._pause = True
 
     def unpause(self):
         """Unpause the heartbeat."""
         self._pause = False
 
     def is_beating(self):
         """Is the heartbeat running and not paused."""
         if self.is_alive() and not self._pause:
             return True
         else:
             return False
 
     def stop(self):
         self._running = False
         super(HBSocketChannel, self).stop()
 
     def call_handlers(self, since_last_heartbeat):
         """This method is called in the ioloop thread when a message arrives.
 
         Subclasses should override this method to handle incoming messages.
         It is important to remember that this method is called in the thread
         so that some logic must be done to ensure that the application leve
         handlers are called in the application thread.
         """
         raise NotImplementedError('call_handlers must be defined in a subclass.')
 
 
 #-----------------------------------------------------------------------------
 # Main kernel manager class
 #-----------------------------------------------------------------------------
 
 class KernelManager(HasTraits):
     """ Manages a kernel for a frontend.
 
     The SUB channel is for the frontend to receive messages published by the
     kernel.
         
     The REQ channel is for the frontend to make requests of the kernel.
     
     The REP channel is for the kernel to request stdin (raw_input) from the
     frontend.
     """
     # The PyZMQ Context to use for communication with the kernel.
     context = Instance(zmq.Context,(),{})
 
     # The Session to use for communication with the kernel.
     session = Instance(Session,(),{})
 
     # The kernel process with which the KernelManager is communicating.
     kernel = Instance(Popen)
 
     # The addresses for the communication channels. 
     xreq_address = TCPAddress((LOCALHOST, 0))
     sub_address = TCPAddress((LOCALHOST, 0))
     rep_address = TCPAddress((LOCALHOST, 0))
     hb_address = TCPAddress((LOCALHOST, 0))
 
     # The classes to use for the various channels.
     xreq_channel_class = Type(XReqSocketChannel)
     sub_channel_class = Type(SubSocketChannel)
     rep_channel_class = Type(RepSocketChannel)
     hb_channel_class = Type(HBSocketChannel)
 
     # Protected traits.
     _launch_args = Any
     _xreq_channel = Any
     _sub_channel = Any
     _rep_channel = Any
     _hb_channel = Any
 
     def __init__(self, **kwargs):
         super(KernelManager, self).__init__(**kwargs)
         # Uncomment this to try closing the context.
         # atexit.register(self.context.close)
 
     #--------------------------------------------------------------------------
     # Channel management methods:
     #--------------------------------------------------------------------------
 
     def start_channels(self, xreq=True, sub=True, rep=True, hb=True):
         """Starts the channels for this kernel.
 
         This will create the channels if they do not exist and then start
         them. If port numbers of 0 are being used (random ports) then you
         must first call :method:`start_kernel`. If the channels have been
         stopped and you call this, :class:`RuntimeError` will be raised.
         """
         if xreq:
             self.xreq_channel.start()
         if sub:
             self.sub_channel.start()
         if rep:
             self.rep_channel.start()
         if hb:
             self.hb_channel.start()
 
     def stop_channels(self):
         """Stops all the running channels for this kernel.
         """
         if self.xreq_channel.is_alive():
             self.xreq_channel.stop()
         if self.sub_channel.is_alive():
             self.sub_channel.stop()
         if self.rep_channel.is_alive():
             self.rep_channel.stop()
         if self.hb_channel.is_alive():
             self.hb_channel.stop()
 
     @property
     def channels_running(self):
         """Are any of the channels created and running?"""
         return (self.xreq_channel.is_alive() or self.sub_channel.is_alive() or
                 self.rep_channel.is_alive() or self.hb_channel.is_alive())
 
     #--------------------------------------------------------------------------
     # Kernel process management methods:
     #--------------------------------------------------------------------------
 
     def start_kernel(self, **kw):
         """Starts a kernel process and configures the manager to use it.
 
         If random ports (port=0) are being used, this method must be called
         before the channels are created.
 
         Parameters:
         -----------
         ipython : bool, optional (default True)
              Whether to use an IPython kernel instead of a plain Python kernel.
         """
         xreq, sub, rep, hb = self.xreq_address, self.sub_address, \
             self.rep_address, self.hb_address
         if xreq[0] not in LOCAL_IPS or sub[0] not in LOCAL_IPS or \
                 rep[0] not in LOCAL_IPS or hb[0] not in LOCAL_IPS:
             raise RuntimeError("Can only launch a kernel on a local interface. "
                                "Make sure that the '*_address' attributes are "
                                "configured properly. "
                                "Currently valid addresses are: %s"%LOCAL_IPS
                                )
                     
         self._launch_args = kw.copy()
         if kw.pop('ipython', True):
             from ipkernel import launch_kernel
         else:
             from pykernel import launch_kernel
         self.kernel, xrep, pub, req, _hb = launch_kernel(
             xrep_port=xreq[1], pub_port=sub[1], 
             req_port=rep[1], hb_port=hb[1], **kw)
         self.xreq_address = (xreq[0], xrep)
         self.sub_address = (sub[0], pub)
         self.rep_address = (rep[0], req)
         self.hb_address = (hb[0], _hb)
 
     def shutdown_kernel(self, restart=False):
         """ Attempts to the stop the kernel process cleanly. If the kernel
         cannot be stopped, it is killed, if possible.
         """
         # FIXME: Shutdown does not work on Windows due to ZMQ errors!
         if sys.platform == 'win32':
             self.kill_kernel()
             return
 
         # Pause the heart beat channel if it exists.
         if self._hb_channel is not None:
             self._hb_channel.pause()
 
         # Don't send any additional kernel kill messages immediately, to give
         # the kernel a chance to properly execute shutdown actions. Wait for at
         # most 1s, checking every 0.1s.
         self.xreq_channel.shutdown(restart=restart)
         for i in range(10):
             if self.is_alive:
                 time.sleep(0.1)
             else:
                 break
         else:
             # OK, we've waited long enough.
             if self.has_kernel:
                 self.kill_kernel()
     
     def restart_kernel(self, now=False):
         """Restarts a kernel with the same arguments that were used to launch
         it. If the old kernel was launched with random ports, the same ports
         will be used for the new kernel.
 
         Parameters
         ----------
         now : bool, optional
           If True, the kernel is forcefully restarted *immediately*, without
           having a chance to do any cleanup action.  Otherwise the kernel is
           given 1s to clean up before a forceful restart is issued.
 
           In all cases the kernel is restarted, the only difference is whether
           it is given a chance to perform a clean shutdown or not.
         """
         if self._launch_args is None:
             raise RuntimeError("Cannot restart the kernel. "
                                "No previous call to 'start_kernel'.")
         else:
             if self.has_kernel:
                 if now:
                     self.kill_kernel()
                 else:
                     self.shutdown_kernel(restart=True)
             self.start_kernel(**self._launch_args)
 
             # FIXME: Messages get dropped in Windows due to probable ZMQ bug
             # unless there is some delay here.
             if sys.platform == 'win32':
                 time.sleep(0.2)
 
     @property
     def has_kernel(self):
         """Returns whether a kernel process has been specified for the kernel
         manager.
         """
         return self.kernel is not None
 
     def kill_kernel(self):
         """ Kill the running kernel. """
         if self.has_kernel:
             # Pause the heart beat channel if it exists.
             if self._hb_channel is not None:
                 self._hb_channel.pause()
 
             # Attempt to kill the kernel.
             try:
                 self.kernel.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
             self.kernel = None
         else:
             raise RuntimeError("Cannot kill kernel. No kernel is running!")
 
     def interrupt_kernel(self):
         """ Interrupts the kernel. Unlike ``signal_kernel``, this operation is
         well supported on all platforms.
         """
         if self.has_kernel:
             if sys.platform == 'win32':
                 from parentpoller import ParentPollerWindows as Poller
                 Poller.send_interrupt(self.kernel.win32_interrupt_event)
             else:
                 self.kernel.send_signal(signal.SIGINT)
         else:
             raise RuntimeError("Cannot interrupt kernel. No kernel is running!")
 
     def signal_kernel(self, signum):
         """ Sends a signal to the kernel. Note that since only SIGTERM is
         supported on Windows, this function is only useful on Unix systems.
         """
         if self.has_kernel:
             self.kernel.send_signal(signum)
         else:
             raise RuntimeError("Cannot signal kernel. No kernel is running!")
 
     @property
     def is_alive(self):
         """Is the kernel process still running?"""
         # FIXME: not using a heartbeat means this method is broken for any
         # remote kernel, it's only capable of handling local kernels.
         if self.has_kernel:
             if self.kernel.poll() is None:
                 return True
             else:
                 return False
         else:
             # We didn't start the kernel with this KernelManager so we don't
             # know if it is running. We should use a heartbeat for this case.
             return True
 
     #--------------------------------------------------------------------------
     # Channels used for communication with the kernel:
     #--------------------------------------------------------------------------
 
     @property
     def xreq_channel(self):
         """Get the REQ socket channel object to make requests of the kernel."""
         if self._xreq_channel is None:
             self._xreq_channel = self.xreq_channel_class(self.context, 
                                                          self.session,
                                                          self.xreq_address)
         return self._xreq_channel
 
     @property
     def sub_channel(self):
         """Get the SUB socket channel object."""
         if self._sub_channel is None:
             self._sub_channel = self.sub_channel_class(self.context,
                                                        self.session,
                                                        self.sub_address)
         return self._sub_channel
 
     @property
     def rep_channel(self):
         """Get the REP socket channel object to handle stdin (raw_input)."""
         if self._rep_channel is None:
             self._rep_channel = self.rep_channel_class(self.context, 
                                                        self.session,
                                                        self.rep_address)
         return self._rep_channel
 
     @property
     def hb_channel(self):
         """Get the REP socket channel object to handle stdin (raw_input)."""
         if self._hb_channel is None:
             self._hb_channel = self.hb_channel_class(self.context, 
                                                        self.session,
                                                        self.hb_address)
         return self._hb_channel

IPython/zmq/pykernel.py

@@ -1,305 +1,296 b''
 #!/usr/bin/env python
 """A simple interactive kernel that talks to a frontend over 0MQ.
 
 Things to do:
 
 * Implement `set_parent` logic. Right before doing exec, the Kernel should
   call set_parent on all the PUB objects with the message about to be executed.
 * Implement random port and security key logic.
 * Implement control messages.
 * Implement event loop and poll version.
 """
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Standard library imports.
 import __builtin__
 from code import CommandCompiler
 import sys
 import time
 import traceback
 
 # System library imports.
 import zmq
 
 # Local imports.
 from IPython.utils.traitlets import HasTraits, Instance
 from completer import KernelCompleter
 from entry_point import base_launch_kernel, make_default_main
 from session import Session, Message
 
 #-----------------------------------------------------------------------------
 # Main kernel class
 #-----------------------------------------------------------------------------
 
 class Kernel(HasTraits):
 
     # Private interface
 
     # This is a dict of port number that the kernel is listening on. It is set
     # by record_ports and used by connect_request.
     _recorded_ports = None
 
     #---------------------------------------------------------------------------
     # Kernel interface
     #---------------------------------------------------------------------------
 
     session = Instance(Session)
     reply_socket = Instance('zmq.Socket')
     pub_socket = Instance('zmq.Socket')
     req_socket = Instance('zmq.Socket')
 
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
         self.user_ns = {}
         self.history = []
         self.compiler = CommandCompiler()
         self.completer = KernelCompleter(self.user_ns)
 
         # Build dict of handlers for message types
         msg_types = [ 'execute_request', 'complete_request', 
                       'object_info_request', 'shutdown_request' ]
         self.handlers = {}
         for msg_type in msg_types:
             self.handlers[msg_type] = getattr(self, msg_type)
 
     def start(self):
         """ Start the kernel main loop.
         """
         while True:
-            ident = self.reply_socket.recv()
-            assert self.reply_socket.rcvmore(), "Missing message part."
-            msg = self.reply_socket.recv_json()
+            ident,msg = self.session.recv(self.reply_socket,0)
+            assert ident is not None, "Missing message part."
             omsg = Message(msg)
             print>>sys.__stdout__
             print>>sys.__stdout__, omsg
             handler = self.handlers.get(omsg.msg_type, None)
             if handler is None:
                 print >> sys.__stderr__, "UNKNOWN MESSAGE TYPE:", omsg
             else:
                 handler(ident, omsg)
 
     def record_ports(self, xrep_port, pub_port, req_port, hb_port):
         """Record the ports that this kernel is using.
 
         The creator of the Kernel instance must call this methods if they
         want the :meth:`connect_request` method to return the port numbers.
         """
         self._recorded_ports = {
             'xrep_port' : xrep_port,
             'pub_port' : pub_port,
             'req_port' : req_port,
             'hb_port' : hb_port
         }
 
     #---------------------------------------------------------------------------
     # Kernel request handlers
     #---------------------------------------------------------------------------
 
     def execute_request(self, ident, parent):
         try:
             code = parent[u'content'][u'code']
         except:
             print>>sys.__stderr__, "Got bad msg: "
             print>>sys.__stderr__, Message(parent)
             return
-        pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
-        self.pub_socket.send_json(pyin_msg)
+        pyin_msg = self.session.send(self.pub_socket, u'pyin',{u'code':code}, parent=parent)
 
         try:
             comp_code = self.compiler(code, '<zmq-kernel>')
 
             # Replace raw_input. Note that is not sufficient to replace 
             # raw_input in the user namespace.
             raw_input = lambda prompt='': self._raw_input(prompt, ident, parent)
             __builtin__.raw_input = raw_input
 
             # Set the parent message of the display hook and out streams.
             sys.displayhook.set_parent(parent)
             sys.stdout.set_parent(parent)
             sys.stderr.set_parent(parent)
 
             exec comp_code in self.user_ns, self.user_ns
         except:
             etype, evalue, tb = sys.exc_info()
             tb = traceback.format_exception(etype, evalue, tb)
             exc_content = {
                 u'status' : u'error',
                 u'traceback' : tb,
                 u'ename' : unicode(etype.__name__),
                 u'evalue' : unicode(evalue)
             }
-            exc_msg = self.session.msg(u'pyerr', exc_content, parent)
-            self.pub_socket.send_json(exc_msg)
+            exc_msg = self.session.send(self.pub_socket, u'pyerr', exc_content, parent)
             reply_content = exc_content
         else:
             reply_content = { 'status' : 'ok', 'payload' : {} }
             
         # Flush output before sending the reply.
         sys.stderr.flush()
         sys.stdout.flush()
 
         # Send the reply.
-        reply_msg = self.session.msg(u'execute_reply', reply_content, parent)
+        reply_msg = self.session.send(self.reply_socket, u'execute_reply', reply_content, parent, ident=ident)
         print>>sys.__stdout__, Message(reply_msg)
-        self.reply_socket.send(ident, zmq.SNDMORE)
-        self.reply_socket.send_json(reply_msg)
         if reply_msg['content']['status'] == u'error':
             self._abort_queue()
 
     def complete_request(self, ident, parent):
         matches = {'matches' : self._complete(parent),
                    'status' : 'ok'}
         completion_msg = self.session.send(self.reply_socket, 'complete_reply',
                                            matches, parent, ident)
         print >> sys.__stdout__, completion_msg
 
     def object_info_request(self, ident, parent):
         context = parent['content']['oname'].split('.')
         object_info = self._object_info(context)
         msg = self.session.send(self.reply_socket, 'object_info_reply',
                                 object_info, parent, ident)
         print >> sys.__stdout__, msg
 
     def shutdown_request(self, ident, parent):
         content = dict(parent['content'])
         msg = self.session.send(self.reply_socket, 'shutdown_reply',
                                 content, parent, ident)
         msg = self.session.send(self.pub_socket, 'shutdown_reply',
                                 content, parent, ident)
         print >> sys.__stdout__, msg
         time.sleep(0.1)
         sys.exit(0)
 
     #---------------------------------------------------------------------------
     # Protected interface
     #---------------------------------------------------------------------------
 
     def _abort_queue(self):
         while True:
             try:
-                ident = self.reply_socket.recv(zmq.NOBLOCK)
+                ident,msg = self.session.recv(self.reply_socket, zmq.NOBLOCK)
             except zmq.ZMQError, e:
                 if e.errno == zmq.EAGAIN:
                     break
             else:
-                assert self.reply_socket.rcvmore(), "Missing message part."
-                msg = self.reply_socket.recv_json()
+                assert ident is not None, "Missing message part."
             print>>sys.__stdout__, "Aborting:"
             print>>sys.__stdout__, Message(msg)
             msg_type = msg['msg_type']
             reply_type = msg_type.split('_')[0] + '_reply'
-            reply_msg = self.session.msg(reply_type, {'status':'aborted'}, msg)
+            reply_msg = self.session.send(self.reply_socket, reply_type, {'status':'aborted'}, msg, ident=ident)
             print>>sys.__stdout__, Message(reply_msg)
-            self.reply_socket.send(ident,zmq.SNDMORE)
-            self.reply_socket.send_json(reply_msg)
             # We need to wait a bit for requests to come in. This can probably
             # be set shorter for true asynchronous clients.
             time.sleep(0.1)
 
     def _raw_input(self, prompt, ident, parent):
         # Flush output before making the request.
         sys.stderr.flush()
         sys.stdout.flush()
 
         # Send the input request.
         content = dict(prompt=prompt)
-        msg = self.session.msg(u'input_request', content, parent)
-        self.req_socket.send_json(msg)
+        msg = self.session.send(self.req_socket, u'input_request', content, parent)
 
         # Await a response.
-        reply = self.req_socket.recv_json()
+        ident,reply = self.session.recv(self.req_socket, 0)
         try:
             value = reply['content']['value']
         except:
             print>>sys.__stderr__, "Got bad raw_input reply: "
             print>>sys.__stderr__, Message(parent)
             value = ''
         return value
 
     def _complete(self, msg):
         return self.completer.complete(msg.content.line, msg.content.text)
 
     def _object_info(self, context):
         symbol, leftover = self._symbol_from_context(context)
         if symbol is not None and not leftover:
             doc = getattr(symbol, '__doc__', '')
         else:
             doc = ''
         object_info = dict(docstring = doc)
         return object_info
 
     def _symbol_from_context(self, context):
         if not context:
             return None, context
 
         base_symbol_string = context[0]
         symbol = self.user_ns.get(base_symbol_string, None)
         if symbol is None:
             symbol = __builtin__.__dict__.get(base_symbol_string, None)
         if symbol is None:
             return None, context
 
         context = context[1:]
         for i, name in enumerate(context):
             new_symbol = getattr(symbol, name, None)
             if new_symbol is None:
                 return symbol, context[i:]
             else:
                 symbol = new_symbol
 
         return symbol, []
 
 #-----------------------------------------------------------------------------
 # Kernel main and launch functions
 #-----------------------------------------------------------------------------
 
 def launch_kernel(ip=None, xrep_port=0, pub_port=0, req_port=0, hb_port=0,
                   independent=False):
     """ Launches a localhost kernel, binding to the specified ports.
 
     Parameters
     ----------
     ip  : str, optional
         The ip address the kernel will bind to.
     
     xrep_port : int, optional
         The port to use for XREP channel.
 
     pub_port : int, optional
         The port to use for the SUB channel.
 
     req_port : int, optional
         The port to use for the REQ (raw input) channel.
 
     hb_port : int, optional
         The port to use for the hearbeat REP channel.
 
     independent : bool, optional (default False) 
         If set, the kernel process is guaranteed to survive if this process
         dies. If not set, an effort is made to ensure that the kernel is killed
         when this process dies. Note that in this case it is still good practice
         to kill kernels manually before exiting.
 
     Returns
     -------
     A tuple of form:
         (kernel_process, xrep_port, pub_port, req_port)
     where kernel_process is a Popen object and the ports are integers.
     """
     extra_arguments = []
     if ip is not None:
         extra_arguments.append('--ip')
         if isinstance(ip, basestring):
             extra_arguments.append(ip)
     
     return base_launch_kernel('from IPython.zmq.pykernel import main; main()',
                               xrep_port, pub_port, req_port, hb_port,
                               independent, extra_arguments=extra_arguments)
 
 main = make_default_main(Kernel)
 
 if __name__ == '__main__':
     main()

IPython/zmq/session.py

@@ -1,122 +1,135 b''
 import os
 import uuid
 import pprint
 
 import zmq
 
+from zmq.utils import jsonapi as json
+
 class Message(object):
     """A simple message object that maps dict keys to attributes.
 
     A Message can be created from a dict and a dict from a Message instance
     simply by calling dict(msg_obj)."""
     
     def __init__(self, msg_dict):
         dct = self.__dict__
         for k, v in msg_dict.iteritems():
             if isinstance(v, dict):
                 v = Message(v)
             dct[k] = v
 
     # Having this iterator lets dict(msg_obj) work out of the box.
     def __iter__(self):
         return iter(self.__dict__.iteritems())
     
     def __repr__(self):
         return repr(self.__dict__)
 
     def __str__(self):
         return pprint.pformat(self.__dict__)
 
     def __contains__(self, k):
         return k in self.__dict__
 
     def __getitem__(self, k):
         return self.__dict__[k]
 
 
 def msg_header(msg_id, username, session):
     return {
         'msg_id' : msg_id,
         'username' : username,
         'session' : session
     }
 
 
 def extract_header(msg_or_header):
     """Given a message or header, return the header."""
     if not msg_or_header:
         return {}
     try:
         # See if msg_or_header is the entire message.
         h = msg_or_header['header']
     except KeyError:
         try:
             # See if msg_or_header is just the header
             h = msg_or_header['msg_id']
         except KeyError:
             raise
         else:
             h = msg_or_header
     if not isinstance(h, dict):
         h = dict(h)
     return h
 
 
 class Session(object):
 
     def __init__(self, username=os.environ.get('USER','username'), session=None):
         self.username = username
         if session is None:
             self.session = str(uuid.uuid4())
         else:
             self.session = session
         self.msg_id = 0
 
     def msg_header(self):
         h = msg_header(self.msg_id, self.username, self.session)
         self.msg_id += 1
         return h
 
     def msg(self, msg_type, content=None, parent=None):
         msg = {}
         msg['header'] = self.msg_header()
         msg['parent_header'] = {} if parent is None else extract_header(parent)
         msg['msg_type'] = msg_type
         msg['content'] = {} if content is None else content
         return msg
 
     def send(self, socket, msg_type, content=None, parent=None, ident=None):
+        if isinstance(msg_type, (Message, dict)):
+            msg = dict(msg_type)
+        else:
         msg = self.msg(msg_type, content, parent)
         if ident is not None:
             socket.send(ident, zmq.SNDMORE)
         socket.send_json(msg)
-        omsg = Message(msg)
-        return omsg
+        # omsg = Message(msg)
+        return msg
 
     def recv(self, socket, mode=zmq.NOBLOCK):
         try:
-            msg = socket.recv_json(mode)
+            msg = socket.recv_multipart(mode)
         except zmq.ZMQError, e:
             if e.errno == zmq.EAGAIN:
                 # We can convert EAGAIN to None as we know in this case
                 # recv_json won't return None.
-                return None
+                return None,None
             else:
                 raise
-        return Message(msg)
+        if len(msg) == 1:
+            ident=None
+            msg = msg[0]
+        elif len(msg) == 2:
+            ident, msg = msg
+        else:
+            raise ValueError("Got message with length > 2, which is invalid")
+        
+        return ident, json.loads(msg)
 
 def test_msg2obj():
     am = dict(x=1)
     ao = Message(am)
     assert ao.x == am['x']
 
     am['y'] = dict(z=1)
     ao = Message(am)
     assert ao.y.z == am['y']['z']
     
     k1, k2 = 'y', 'z'
     assert ao[k1][k2] == am[k1][k2]
     
     am2 = dict(ao)
     assert am['x'] == am2['x']
     assert am['y']['z'] == am2['y']['z']

IPython/zmq/zmqshell.py

@@ -1,581 +1,580 b''
 """A ZMQ-based subclass of InteractiveShell.
 
 This code is meant to ease the refactoring of the base InteractiveShell into
 something with a cleaner architecture for 2-process use, without actually
 breaking InteractiveShell itself.  So we're doing something a bit ugly, where
 we subclass and override what we want to fix.  Once this is working well, we
 can go back to the base class and refactor the code for a cleaner inheritance
 implementation that doesn't rely on so much monkeypatching.
 
 But this lets us maintain a fully working IPython as we develop the new
 machinery.  This should thus be thought of as scaffolding.
 """
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib
 import inspect
 import os
 import re
 
 # Our own
 from IPython.core.interactiveshell import (
     InteractiveShell, InteractiveShellABC
 )
 from IPython.core import page
 from IPython.core.displayhook import DisplayHook
 from IPython.core.macro import Macro
 from IPython.core.payloadpage import install_payload_page
 from IPython.utils import io
 from IPython.utils.path import get_py_filename
 from IPython.utils.text import StringTypes
 from IPython.utils.traitlets import Instance, Type, Dict
 from IPython.utils.warn import warn
 from IPython.zmq.session import extract_header
 from session import Session
 
 #-----------------------------------------------------------------------------
 # Globals and side-effects
 #-----------------------------------------------------------------------------
 
 # Install the payload version of page.
 install_payload_page()
 
 #-----------------------------------------------------------------------------
 # Functions and classes
 #-----------------------------------------------------------------------------
 
 class ZMQDisplayHook(DisplayHook):
 
     session = Instance(Session)
     pub_socket = Instance('zmq.Socket')
     parent_header = Dict({})
 
     def set_parent(self, parent):
         """Set the parent for outbound messages."""
         self.parent_header = extract_header(parent)
 
     def start_displayhook(self):
         self.msg = self.session.msg(u'pyout', {}, parent=self.parent_header)
 
     def write_output_prompt(self):
         """Write the output prompt."""
         if self.do_full_cache:
             self.msg['content']['execution_count'] = self.prompt_count
 
     def write_result_repr(self, result_repr, extra_formats):
         self.msg['content']['data'] = result_repr
         self.msg['content']['extra_formats'] = extra_formats
 
     def finish_displayhook(self):
         """Finish up all displayhook activities."""
-        self.pub_socket.send_json(self.msg)
+        self.session.send(self.pub_socket, self.msg)
         self.msg = None
 
 
 class ZMQInteractiveShell(InteractiveShell):
     """A subclass of InteractiveShell for ZMQ."""
 
     displayhook_class = Type(ZMQDisplayHook)
     keepkernel_on_exit = None
 
     def init_environment(self):
         """Configure the user's environment.
 
         """
         env = os.environ
         # These two ensure 'ls' produces nice coloring on BSD-derived systems
         env['TERM'] = 'xterm-color'
         env['CLICOLOR'] = '1'
         # Since normal pagers don't work at all (over pexpect we don't have
         # single-key control of the subprocess), try to disable paging in
         # subprocesses as much as possible.
         env['PAGER'] = 'cat'
         env['GIT_PAGER'] = 'cat'
 
     def auto_rewrite_input(self, cmd):
         """Called to show the auto-rewritten input for autocall and friends.
 
         FIXME: this payload is currently not correctly processed by the
         frontend.
         """
         new = self.displayhook.prompt1.auto_rewrite() + cmd
         payload = dict(
             source='IPython.zmq.zmqshell.ZMQInteractiveShell.auto_rewrite_input',
             transformed_input=new,
             )
         self.payload_manager.write_payload(payload)
         
     def ask_exit(self):
         """Engage the exit actions."""
         payload = dict(
             source='IPython.zmq.zmqshell.ZMQInteractiveShell.ask_exit',
             exit=True,
             keepkernel=self.keepkernel_on_exit,
             )
         self.payload_manager.write_payload(payload)
 
     def _showtraceback(self, etype, evalue, stb):
 
         exc_content = {
             u'traceback' : stb,
             u'ename' : unicode(etype.__name__),
             u'evalue' : unicode(evalue)
         }
 
         dh = self.displayhook
-        exc_msg = dh.session.msg(u'pyerr', exc_content, dh.parent_header)
         # Send exception info over pub socket for other clients than the caller
         # to pick up
-        dh.pub_socket.send_json(exc_msg)
+        exc_msg = dh.session.send(dh.pub_socket, u'pyerr', exc_content, dh.parent_header)
 
         # FIXME - Hack: store exception info in shell object.  Right now, the
         # caller is reading this info after the fact, we need to fix this logic
         # to remove this hack.  Even uglier, we need to store the error status
         # here, because in the main loop, the logic that sets it is being
         # skipped because runlines swallows the exceptions.
         exc_content[u'status'] = u'error'
         self._reply_content = exc_content
         # /FIXME
         
         return exc_content
 
     #------------------------------------------------------------------------
     # Magic overrides
     #------------------------------------------------------------------------
     # Once the base class stops inheriting from magic, this code needs to be
     # moved into a separate machinery as well.  For now, at least isolate here
     # the magics which this class needs to implement differently from the base
     # class, or that are unique to it.
 
     def magic_doctest_mode(self,parameter_s=''):
         """Toggle doctest mode on and off.
 
         This mode is intended to make IPython behave as much as possible like a
         plain Python shell, from the perspective of how its prompts, exceptions
         and output look.  This makes it easy to copy and paste parts of a
         session into doctests.  It does so by:
 
         - Changing the prompts to the classic ``>>>`` ones.
         - Changing the exception reporting mode to 'Plain'.
         - Disabling pretty-printing of output.
 
         Note that IPython also supports the pasting of code snippets that have
         leading '>>>' and '...' prompts in them.  This means that you can paste
         doctests from files or docstrings (even if they have leading
         whitespace), and the code will execute correctly.  You can then use
         '%history -t' to see the translated history; this will give you the
         input after removal of all the leading prompts and whitespace, which
         can be pasted back into an editor.
 
         With these features, you can switch into this mode easily whenever you
         need to do testing and changes to doctests, without having to leave
         your existing IPython session.
         """
 
         from IPython.utils.ipstruct import Struct
 
         # Shorthands
         shell = self.shell
         # dstore is a data store kept in the instance metadata bag to track any
         # changes we make, so we can undo them later.
         dstore = shell.meta.setdefault('doctest_mode', Struct())
         save_dstore = dstore.setdefault
 
         # save a few values we'll need to recover later
         mode = save_dstore('mode', False)
         save_dstore('rc_pprint', shell.pprint)
         save_dstore('xmode', shell.InteractiveTB.mode)
         
         if mode == False:
             # turn on
             shell.pprint = False
             shell.magic_xmode('Plain')
         else:
             # turn off
             shell.pprint = dstore.rc_pprint
             shell.magic_xmode(dstore.xmode)
 
         # Store new mode and inform on console
         dstore.mode = bool(1-int(mode))
         mode_label = ['OFF','ON'][dstore.mode]
         print('Doctest mode is:', mode_label)
         
         # Send the payload back so that clients can modify their prompt display
         payload = dict(
             source='IPython.zmq.zmqshell.ZMQInteractiveShell.magic_doctest_mode',
             mode=dstore.mode)
         self.payload_manager.write_payload(payload)
 
     def magic_edit(self,parameter_s='',last_call=['','']):
         """Bring up an editor and execute the resulting code.
 
         Usage:
           %edit [options] [args]
 
         %edit runs IPython's editor hook.  The default version of this hook is
         set to call the __IPYTHON__.rc.editor command.  This is read from your
         environment variable $EDITOR.  If this isn't found, it will default to
         vi under Linux/Unix and to notepad under Windows.  See the end of this
         docstring for how to change the editor hook.
 
         You can also set the value of this editor via the command line option
         '-editor' or in your ipythonrc file. This is useful if you wish to use
         specifically for IPython an editor different from your typical default
         (and for Windows users who typically don't set environment variables).
 
         This command allows you to conveniently edit multi-line code right in
         your IPython session.
         
         If called without arguments, %edit opens up an empty editor with a
         temporary file and will execute the contents of this file when you
         close it (don't forget to save it!).
 
 
         Options:
 
         -n <number>: open the editor at a specified line number.  By default,
         the IPython editor hook uses the unix syntax 'editor +N filename', but
         you can configure this by providing your own modified hook if your
         favorite editor supports line-number specifications with a different
         syntax.
         
         -p: this will call the editor with the same data as the previous time
         it was used, regardless of how long ago (in your current session) it
         was.
 
         -r: use 'raw' input.  This option only applies to input taken from the
         user's history.  By default, the 'processed' history is used, so that
         magics are loaded in their transformed version to valid Python.  If
         this option is given, the raw input as typed as the command line is
         used instead.  When you exit the editor, it will be executed by
         IPython's own processor.
         
         -x: do not execute the edited code immediately upon exit. This is
         mainly useful if you are editing programs which need to be called with
         command line arguments, which you can then do using %run.
 
 
         Arguments:
 
         If arguments are given, the following possibilites exist:
 
         - The arguments are numbers or pairs of colon-separated numbers (like
         1 4:8 9). These are interpreted as lines of previous input to be
         loaded into the editor. The syntax is the same of the %macro command.
 
         - If the argument doesn't start with a number, it is evaluated as a
         variable and its contents loaded into the editor. You can thus edit
         any string which contains python code (including the result of
         previous edits).
 
         - If the argument is the name of an object (other than a string),
         IPython will try to locate the file where it was defined and open the
         editor at the point where it is defined. You can use `%edit function`
         to load an editor exactly at the point where 'function' is defined,
         edit it and have the file be executed automatically.
 
         If the object is a macro (see %macro for details), this opens up your
         specified editor with a temporary file containing the macro's data.
         Upon exit, the macro is reloaded with the contents of the file.
 
         Note: opening at an exact line is only supported under Unix, and some
         editors (like kedit and gedit up to Gnome 2.8) do not understand the
         '+NUMBER' parameter necessary for this feature. Good editors like
         (X)Emacs, vi, jed, pico and joe all do.
 
         - If the argument is not found as a variable, IPython will look for a
         file with that name (adding .py if necessary) and load it into the
         editor. It will execute its contents with execfile() when you exit,
         loading any code in the file into your interactive namespace.
 
         After executing your code, %edit will return as output the code you
         typed in the editor (except when it was an existing file). This way
         you can reload the code in further invocations of %edit as a variable,
         via _<NUMBER> or Out[<NUMBER>], where <NUMBER> is the prompt number of
         the output.
 
         Note that %edit is also available through the alias %ed.
 
         This is an example of creating a simple function inside the editor and
         then modifying it. First, start up the editor:
 
         In [1]: ed
         Editing... done. Executing edited code...
         Out[1]: 'def foo():n    print "foo() was defined in an editing session"n'
 
         We can then call the function foo():
         
         In [2]: foo()
         foo() was defined in an editing session
 
         Now we edit foo.  IPython automatically loads the editor with the
         (temporary) file where foo() was previously defined:
         
         In [3]: ed foo
         Editing... done. Executing edited code...
 
         And if we call foo() again we get the modified version:
         
         In [4]: foo()
         foo() has now been changed!
 
         Here is an example of how to edit a code snippet successive
         times. First we call the editor:
 
         In [5]: ed
         Editing... done. Executing edited code...
         hello
         Out[5]: "print 'hello'n"
 
         Now we call it again with the previous output (stored in _):
 
         In [6]: ed _
         Editing... done. Executing edited code...
         hello world
         Out[6]: "print 'hello world'n"
 
         Now we call it with the output #8 (stored in _8, also as Out[8]):
 
         In [7]: ed _8
         Editing... done. Executing edited code...
         hello again
         Out[7]: "print 'hello again'n"
 
 
         Changing the default editor hook:
 
         If you wish to write your own editor hook, you can put it in a
         configuration file which you load at startup time.  The default hook
         is defined in the IPython.core.hooks module, and you can use that as a
         starting example for further modifications.  That file also has
         general instructions on how to set a new hook for use once you've
         defined it."""
         
         # FIXME: This function has become a convoluted mess.  It needs a
         # ground-up rewrite with clean, simple logic.
 
         def make_filename(arg):
             "Make a filename from the given args"
             try:
                 filename = get_py_filename(arg)
             except IOError:
                 if args.endswith('.py'):
                     filename = arg
                 else:
                     filename = None
             return filename
 
         # custom exceptions
         class DataIsObject(Exception): pass
 
         opts,args = self.parse_options(parameter_s,'prn:')
         # Set a few locals from the options for convenience:
         opts_p = opts.has_key('p')
         opts_r = opts.has_key('r')
         
         # Default line number value
         lineno = opts.get('n',None)
         if lineno is not None:
             try:
                 lineno = int(lineno)
             except:
                 warn("The -n argument must be an integer.")
                 return
 
         if opts_p:
             args = '_%s' % last_call[0]
             if not self.shell.user_ns.has_key(args):
                 args = last_call[1]
             
         # use last_call to remember the state of the previous call, but don't
         # let it be clobbered by successive '-p' calls.
         try:
             last_call[0] = self.shell.displayhook.prompt_count
             if not opts_p:
                 last_call[1] = parameter_s
         except:
             pass
 
         # by default this is done with temp files, except when the given
         # arg is a filename
         use_temp = 1
 
         if re.match(r'\d',args):
             # Mode where user specifies ranges of lines, like in %macro.
             # This means that you can't edit files whose names begin with
             # numbers this way. Tough.
             ranges = args.split()
             data = ''.join(self.extract_input_slices(ranges,opts_r))
         elif args.endswith('.py'):
             filename = make_filename(args)
             data = ''
             use_temp = 0
         elif args:
             try:
                 # Load the parameter given as a variable. If not a string,
                 # process it as an object instead (below)
 
                 #print '*** args',args,'type',type(args)  # dbg
                 data = eval(args,self.shell.user_ns)
                 if not type(data) in StringTypes:
                     raise DataIsObject
 
             except (NameError,SyntaxError):
                 # given argument is not a variable, try as a filename
                 filename = make_filename(args)
                 if filename is None:
                     warn("Argument given (%s) can't be found as a variable "
                          "or as a filename." % args)
                     return
 
                 data = ''
                 use_temp = 0
             except DataIsObject:
 
                 # macros have a special edit function
                 if isinstance(data,Macro):
                     self._edit_macro(args,data)
                     return
                                 
                 # For objects, try to edit the file where they are defined
                 try:
                     filename = inspect.getabsfile(data)
                     if 'fakemodule' in filename.lower() and inspect.isclass(data):                     
                         # class created by %edit? Try to find source
                         # by looking for method definitions instead, the
                         # __module__ in those classes is FakeModule.
                         attrs = [getattr(data, aname) for aname in dir(data)]
                         for attr in attrs:
                             if not inspect.ismethod(attr):
                                 continue
                             filename = inspect.getabsfile(attr)
                             if filename and 'fakemodule' not in filename.lower():
                                 # change the attribute to be the edit target instead
                                 data = attr 
                                 break
                     
                     datafile = 1
                 except TypeError:
                     filename = make_filename(args)
                     datafile = 1
                     warn('Could not find file where `%s` is defined.\n'
                          'Opening a file named `%s`' % (args,filename))
                 # Now, make sure we can actually read the source (if it was in
                 # a temp file it's gone by now).
                 if datafile:
                     try:
                         if lineno is None:
                             lineno = inspect.getsourcelines(data)[1]
                     except IOError:
                         filename = make_filename(args)
                         if filename is None:
                             warn('The file `%s` where `%s` was defined cannot '
                                  'be read.' % (filename,data))
                             return
                 use_temp = 0
         else:
             data = ''
 
         if use_temp:
             filename = self.shell.mktempfile(data)
             print('IPython will make a temporary file named:', filename)
 
         # Make sure we send to the client an absolute path, in case the working
         # directory of client and kernel don't match
         filename = os.path.abspath(filename)
 
         payload = {
             'source' : 'IPython.zmq.zmqshell.ZMQInteractiveShell.edit_magic',
             'filename' : filename,
             'line_number' : lineno
         }
         self.payload_manager.write_payload(payload)
 
     def magic_gui(self, *args, **kwargs):
         raise NotImplementedError(
             'GUI support must be enabled in command line options.')
 
     def magic_pylab(self, *args, **kwargs):
         raise NotImplementedError(
             'pylab support must be enabled in command line options.')
 
     # A few magics that are adapted to the specifics of using pexpect and a
     # remote terminal
 
     def magic_clear(self, arg_s):
         """Clear the terminal."""
         if os.name == 'posix':
             self.shell.system("clear")
         else:
             self.shell.system("cls")
 
     if os.name == 'nt':
         # This is the usual name in windows
         magic_cls = magic_clear
 
     # Terminal pagers won't work over pexpect, but we do have our own pager
     
     def magic_less(self, arg_s):
         """Show a file through the pager.
 
         Files ending in .py are syntax-highlighted."""
         cont = open(arg_s).read()
         if arg_s.endswith('.py'):
             cont = self.shell.pycolorize(cont)
         page.page(cont)
 
     magic_more = magic_less
 
     # Man calls a pager, so we also need to redefine it
     if os.name == 'posix':
         def magic_man(self, arg_s):
             """Find the man page for the given command and display in pager."""
             page.page(self.shell.getoutput('man %s | col -b' % arg_s,
                                            split=False))
 
     # FIXME: this is specific to the GUI, so we should let the gui app load
     # magics at startup that are only for the gui.  Once the gui app has proper
     # profile and configuration management, we can have it initialize a kernel
     # with a special config file that provides these.
     def magic_guiref(self, arg_s):
         """Show a basic reference about the GUI console."""
         from IPython.core.usage import gui_reference
         page.page(gui_reference, auto_html=True)
 
     def magic_loadpy(self, arg_s):
         """Load a .py python script into the GUI console.
 
         This magic command can either take a local filename or a url::
 
         %loadpy myscript.py
         %loadpy http://www.example.com/myscript.py
         """
         if not arg_s.endswith('.py'):
             raise ValueError('%%load only works with .py files: %s' % arg_s)
         if arg_s.startswith('http'):
             import urllib2
             response = urllib2.urlopen(arg_s)
             content = response.read()
         else:
             content = open(arg_s).read()
         payload = dict(
             source='IPython.zmq.zmqshell.ZMQInteractiveShell.magic_loadpy', 
             text=content
         )
         self.payload_manager.write_payload(payload)
         
     def magic_Exit(self, parameter_s=''):
         """Exit IPython. If the -k option is provided, the kernel will be left
         running. Otherwise, it will shutdown without prompting.
         """
         opts,args = self.parse_options(parameter_s,'k')
         self.shell.keepkernel_on_exit = opts.has_key('k')
         self.shell.ask_exit()
 
     # Add aliases as magics so all common forms work: exit, quit, Exit, Quit.
     magic_exit = magic_quit = magic_Quit = magic_Exit
 
 InteractiveShellABC.register(ZMQInteractiveShell)