Fix QueueFile to support read-to-end semantics and add some tests

util/registry/queuefile.py

@@ -1,5 +1,5 @@
 class QueueFile(object):
-  """ Class which implements a file-like interface and reads from a blocking
+  """ Class which implements a file-like interface and reads QueueResult's from a blocking
       multiprocessing queue.
   """
   def __init__(self, queue, name=None):
@@ -15,40 +15,61 @@ class QueueFile(object):
   def add_exception_handler(self, handler):
     self._exception_handlers.append(handler)
 
-  def read(self, size=8192):
+  def read(self, size=-1):
+    # If the queuefile was closed or we have finished, send back any remaining data.
     if self._closed or self._done:
+      if size == -1:
+        buf = self._buffer
+        self._buffer = ''
+        return buf
+
       buf = self._buffer[0:size]
       self._buffer = self._buffer[size:]
       return buf
 
-    while len(self._buffer) < size:
+    # Loop until we reach the requested data size (or forever if all data was requested).
+    while (len(self._buffer) < size) or (size == -1):
       result = self._queue.get(block=True)
-      if result is None:
-        self._done = True
-        break
 
-      if isinstance(result, Exception):
+      # Check for any exceptions raised by the queue process.
+      if result.exception is not None:
         self._closed = True
         self.raised_exception = True
 
+        # Fire off the exception to any registered handlers. If no handlers were registered,
+        # then raise the exception locally.
         handled = False
         for handler in self._exception_handlers:
-          handler(result)
+          handler(result.exception)
           handled = True
 
         if handled:
           return
+        else:
+          raise result.exception
 
-        raise result
+      # Check for no further data. If the QueueProcess has finished producing data, then break
+      # out of the loop to return the data already acquired.
+      if result.data is None:
+        self._done = True
+        break
 
-      self._buffer += result
-      self._total_size += len(result)
+      # Add the data to the buffer.
+      self._buffer += result.data
+      self._total_size += len(result.data)
+
+    # Return the requested slice of the buffer.
+    if size == -1:
+      buf = self._buffer
+      self._buffer = ''
+      return buf
 
     buf = self._buffer[0:size]
     self._buffer = self._buffer[size:]
     return buf
 
   def flush(self):
+    # Purposefully not implemented.
     pass
 
   def close(self):