Refactor and simplification of zmqterminal.

IPython/frontend/zmqterminal/completer.py

@@ -1,11 +1,6 b''
 # -*- coding: utf-8 -*-
 import readline
-import time
-import sys
-stdout = sys.stdout
-
-class TimeoutError(Exception):
-    pass
+from Queue import Empty
 
 class ClientCompleter2p(object):
     """Client-side completion machinery.
@@ -27,22 +22,18 b' class ClientCompleter2p(object):'
                                 dict(text=text, line=line))
         # send completion request to kernel
         # Give the kernel up to 0.5s to respond
-        for i in range(5):
-            if self.km.xreq_channel.was_called():
-                msg_xreq =  self.km.xreq_channel.get_msg()
+        msg_xreq = self.km.xreq_channel.get_msg(timeout=0.5)
                 if msg["header"]['session'] == msg_xreq["parent_header"]['session'] and \
                                 msg_xreq["content"]["status"] == 'ok' and \
                                 msg_xreq["msg_type"] == "complete_reply" :
                     return msg_xreq["content"]["matches"]
-                       
-            time.sleep(0.1)
-        raise TimeoutError
+        return []
     
     def complete(self, text, state):
         if state == 0:
             try:
                 self.matches = self.complete_request(text)
-            except TimeoutError:
+            except Empty:
                 print('WARNING: Kernel timeout on tab completion.')
         
         try:

IPython/frontend/zmqterminal/frontend.py

@@ -17,32 +17,18 b' For more details, see the ipython-zmq design'
 #-----------------------------------------------------------------------------
 
 import __builtin__
-from contextlib import nested
-import time
 import sys
 import os
-import signal
-import uuid
-import cPickle as pickle
-import code
-import zmq
+from Queue import Empty
 import readline
 import rlcompleter
-import time
 
 #-----------------------------------------------------------------------------
 # Imports from ipython
 #-----------------------------------------------------------------------------
 from IPython.external.argparse import ArgumentParser
-from IPython.utils.traitlets import (
-Int, Str, CBool, CaselessStrEnum, Enum, List, Unicode
-)
-from IPython.core.interactiveshell import get_default_colors
-from IPython.core.excolors import exception_colors
-from IPython.utils import PyColorize
 from IPython.core.inputsplitter import IPythonInputSplitter
-from IPython.frontend.zmqterminal.kernelmanager import KernelManager2p as KernelManager
-from IPython.zmq.session import Session
+from IPython.zmq.blockingkernelmanager import BlockingKernelManager as KernelManager
 from IPython.frontend.zmqterminal.completer import ClientCompleter2p 
 
 #-----------------------------------------------------------------------------
@@ -134,13 +120,14 b' class Frontend(object):'
         See parent class :meth:`execute` docstring for full details.
        """
        self.km.xreq_channel.execute(source, hidden)
+        while not self.km.xreq_channel.msg_ready():
+            try:
+                self.handle_rep_channel(timeout=0.1)
+            except Empty:
+                pass
        self.handle_xreq_channel()
-       self.handle_rep_channel()
 
    def handle_xreq_channel(self):
-       # Give the kernel up to 0.5s to respond
-       for i in range(5):
-           if self.km.xreq_channel.was_called():
               self.msg_xreq =  self.km.xreq_channel.get_msg()
               if self.msg_header["session"] == self.msg_xreq["parent_header"]["session"] :
                  if self.msg_xreq["content"]["status"] == 'ok' :
@@ -151,11 +138,12 b' class Frontend(object):'
                  else:
                      etb = self.msg_xreq["content"]["traceback"]
                      print >> sys.stderr, etb[0]
+                try:        # These bits aren't there for a SyntaxError
                      print >> sys.stderr, etb[1]
                      print >> sys.stderr, etb[2]
+                except IndexError:
+                    pass
                      self.prompt_count = self.msg_xreq["content"]["execution_count"]+1
-                     break
-           time.sleep(0.1)
 
 
    def handle_sub_channel(self):
@@ -168,7 +156,7 b' class Frontend(object):'
            sub_msg:  message receive from kernel in the sub socket channel
                      capture by kernel manager.
        """
-       while self.km.sub_channel.was_called():
+       while self.km.sub_channel.msg_ready():
            sub_msg = self.km.sub_channel.get_msg()
            if self.msg_header["username"] == sub_msg['parent_header']['username'] and \
                 self.km.session.session == sub_msg['parent_header']['session']:
@@ -188,11 +176,10 b' class Frontend(object):'
                     print >> sys.stdout,"Out[%i]:"%sub_msg["content"]["execution_count"], sub_msg["content"]["data"]["text/plain"]
                     sys.stdout.flush()
 
-   def handle_rep_channel(self):
+    def handle_rep_channel(self, timeout=0.1):
        """ Method to capture raw_input
        """
-       if self.km.rep_channel.was_called() :
-          self.msg_rep = self.km.rep_channel.get_msg()
+        self.msg_rep = self.km.rep_channel.get_msg(timeout=timeout)
           if self.msg_header["session"] == self.msg_rep["parent_header"]["session"] :
              raw_data = raw_input(self.msg_rep["content"]["prompt"])
              self.km.rep_channel.input(raw_data)
@@ -270,7 +257,6 b' def start_frontend():'
 
     
     kernel_manager.start_channels()
-    time.sleep(4)
  
     frontend=Frontend(kernel_manager)
     return frontend

IPython/zmq/blockingkernelmanager.py

@@ -16,6 +16,7 b' from __future__ import print_function'
 
 # Stdlib
 from Queue import Queue, Empty
+from threading import Event
 
 # Our own
 from IPython.utils import io
@@ -104,9 +105,31 b' class BlockingShellSocketChannel(ShellSocketChannel):'
 
 class BlockingStdInSocketChannel(StdInSocketChannel):
     
+    def __init__(self, context, session, address=None):
+        super(BlockingRepSocketChannel, self).__init__(context, session, address)
+        self._in_queue = Queue()
+        
     def call_handlers(self, msg):
         #io.rprint('[[Rep]]', msg) # dbg
-        pass
+        self._in_queue.put(msg)
+        
+    def get_msg(self, block=True, timeout=None):
+        "Gets a message if there is one that is ready."
+        return self._in_queue.get(block, timeout)
+        
+    def get_msgs(self):
+        """Get all messages that are currently ready."""
+        msgs = []
+        while True:
+            try:
+                msgs.append(self.get_msg(block=False))
+            except Empty:
+                break
+        return msgs
+    
+    def msg_ready(self):
+        "Is there a message that has been received?"
+        return not self._in_queue.empty()
 
 
 class BlockingHBSocketChannel(HBSocketChannel):