Add ResultProcessor

awscli/customizations/s3/results.py

@@ -12,10 +12,13 @@
 # language governing permissions and limitations under the License.
 import logging
 import sys
+import threading
 from collections import namedtuple
 
 from s3transfer.subscribers import BaseSubscriber
 
+from awscli.compat import queue
+from awscli.customizations.s3.executor import ShutdownThreadRequest
 from awscli.customizations.s3.utils import relative_path
 from awscli.customizations.s3.utils import human_readable_size
 from awscli.customizations.s3.utils import uni_print
@@ -138,7 +141,13 @@ def _get_src_dest(self, future):
         return src, dest
 
 
-class ResultRecorder(object):
+class BaseResultHandler(object):
+    """Base handler class to be called in the ResultProcessor"""
+    def __call__(self, result):
+        raise NotImplementedError('__call__()')
+
+
+class ResultRecorder(BaseResultHandler):
     """Records and track transfer statistics based on results receieved"""
     def __init__(self):
         self.bytes_transferred = 0
@@ -161,7 +170,7 @@ def __init__(self):
             WarningResult: self._record_warning_result,
         }
 
-    def record_result(self, result):
+    def __call__(self, result):
         """Record the result of an individual Result object"""
         self._result_handler_map.get(type(result), self._record_noop)(
             result=result)
@@ -217,7 +226,7 @@ def _record_warning_result(self, **kwargs):
         self.files_warned += 1
 
 
-class ResultPrinter(object):
+class ResultPrinter(BaseResultHandler):
     PROGRESS_FORMAT = (
         'Completed {bytes_completed}/{expected_bytes_completed} with '
         '{remaining_files} file(s) remaining.'
@@ -260,7 +269,7 @@ def __init__(self, result_recorder, out_file=sys.stdout,
             WarningResult: self._print_warning,
         }
 
-    def print_result(self, result):
+    def __call__(self, result):
         """Print the progress of the ongoing transfer based on a result"""
         self._result_handler_map.get(type(result), self._print_noop)(
             result=result)
@@ -356,3 +365,44 @@ def _print_progress(self, **kwargs):
 
     def _print_success(self, result, **kwargs):
         pass
+
+
+class ResultProcessor(threading.Thread):
+    def __init__(self, result_queue, result_handlers=None):
+        """Thread to process results from result queue
+
+        This includes recording statistics and printing transfer status
+
+        :param result_queue: The result queue to process results from
+        :param result_handlers: A list of callables that take a result in as
+            a parameter to process the result for that handler.
+        """
+        threading.Thread.__init__(self)
+        self._result_queue = result_queue
+        self._result_handlers = result_handlers
+        if self._result_handlers is None:
+            self._result_handlers = []
+
+    def run(self):
+        while True:
+            try:
+                result = self._result_queue.get(True)
+                if isinstance(result, ShutdownThreadRequest):
+                    LOGGER.debug(
+                        'Shutdown request received in result processing '
+                        'thread, shutting down result thread.')
+                    break
+                LOGGER.debug('Received result: %s', result)
+                self._process_result(result)
+            except queue.Empty:
+                pass
+
+    def _process_result(self, result):
+        for result_handler in self._result_handlers:
+            try:
+                LOGGER.debug('Processing %s with %s', result, result_handler)
+                result_handler(result)
+            except Exception as e:
+                LOGGER.debug(
+                    'Error processing result %s with handler %s: %s',
+                    result, result_handler, e, exc_info=True)