use idle status message to signal when outputs are complete

IPython/parallel/client/asyncresult.py

@@ -158,6 +158,8 b' class AsyncResult(object):'
                 self._success = True
             finally:
                 self._metadata = map(self._client.metadata.get, self.msg_ids)
+                self._wait_for_outputs()
+            
 
 
     def successful(self):
@@ -424,6 +426,19 b' class AsyncResult(object):'
         if self.pyout is not None:
             display(self.get())
     
+    def _wait_for_outputs(self, timeout=-1):
+        """wait for the 'status=idle' message that indicates we have all outputs
+        """
+        if not self._success:
+            # don't wait on errors
+            return
+        tic = time.time()
+        while not all(md['outputs_ready'] for md in self._metadata):
+            time.sleep(0.01)
+            self._client._flush_iopub(self._client._iopub_socket)
+            if timeout >= 0 and time.time() > tic + timeout:
+                break
+    
     @check_ready
     def display_outputs(self, groupby="type"):
         """republish the outputs of the computation
@@ -453,8 +468,6 b' class AsyncResult(object):'
                 several plots, and you would like to see all of the first
                 plots together, then all of the second plots, and so on.
         """
-        # flush iopub, just in case
-        self._client._flush_iopub(self._client._iopub_socket)
         if self._single_result:
             self._display_single_result()
             return
@@ -619,7 +632,6 b' class AsyncMapResult(AsyncResult):'
                 yield r
 
 
-
 class AsyncHubResult(AsyncResult):
     """Class to wrap pending results that must be requested from the Hub.
 
@@ -627,6 +639,10 b' class AsyncHubResult(AsyncResult):'
     so use `AsyncHubResult.wait()` sparingly.
     """
 
+    def _wait_for_outputs(self, timeout=None):
+        """no-op, because HubResults are never incomplete"""
+        return
+    
     def wait(self, timeout=-1):
         """wait for result to complete."""
         start = time.time()

IPython/parallel/client/client.py

@@ -184,6 +184,7 b' class Metadata(dict):'
               'stdout' : '',
               'stderr' : '',
               'outputs' : [],
+              'outputs_ready' : False,
             }
         self.update(md)
         self.update(dict(*args, **kwargs))
@@ -880,6 +881,10 b' class Client(HasTraits):'
                 md['outputs'].append(content)
             elif msg_type == 'pyout':
                 md['pyout'] = content
+            elif msg_type == 'status':
+                # idle message comes after all outputs
+                if content['execution_state'] == 'idle':
+                    md['outputs_ready'] = True
             else:
                 # unhandled msg_type (status, etc.)
                 pass