Merge lp:~pedronis/ubuntuone-client/transfer-progress-events into lp:ubuntuone-client

[Samuele Pedroni]

addressing the demands of Bug #570747, introduces internal events AQ_DOWNLOAD/UPLOAD_FILE_PROGRESS that then get converted into D-Bus signals Download/UploadFileProgress.

addressing the demands of Bug #570747, introduces internal events AQ_DOWNLOAD/UPLOAD_FILE_PROGRESS that then get converted into D-Bus signals Download/UploadFileProgress. The information conveyed is similar to that of the current_downloads method, information about deflated_size and numbers of *compressed* bytes transfered are reported. Events are generated only if at least 64Kb of progress have been made.

[Natalia Bidart (nataliabidart)]

 * This docstring seems incorrect:

+class DownloadTestCase(ConnectedBaseTestCase):
+ """Test for ListDir ActionQueueCommand."""

 * class MyGetContent(GetContentMixin) (and all the other auxiliar classes) need docstring, and the self shadows the self from the outter class.

 * most of the (new) tests needs their docstring.

 * use assertEqual instead of assertEquals (the latter will be deprecated with unittest2).

 * PEP-8 indicates 2 blank lines between classes.

 * the adding of new events to the EVENTS collection should be tested

review: Needs Fixing

[Samuele Pedroni (pedronis)]

Now all style issues should have been addressed, also the progress hooks have been simplified to take only the byte count.

[Rodrigo Moya (rodrigo-moya)]

As for the libubuntuone needs, this is ok, even a bigger threshold would be ok

[Natalia Bidart (nataliabidart)]

Thank you!

review: Approve

[Rick McBride (rmcbride)]

Looks good!

review: Approve

[Samuele Pedroni (pedronis)]

there's a bug that was not there about resetting the n_bytes_*_last counters that need fixing

review: Needs Fixing

[Samuele Pedroni (pedronis)]

I pushed fixes for the issue, plus testing

review: Abstain

[Rick McBride (rmcbride)]

review: Abstain

[Stuart Colville (muffinresearch)]

Tests pass - looks good.

review: Approve

[Facundo Batista (facundo)]

Looks ok now

review: Approve

=== modified file 'tests/syncdaemon/test_action_queue.py'
--- tests/syncdaemon/test_action_queue.py	2010-05-03 23:40:25 +0000
+++ tests/syncdaemon/test_action_queue.py	2010-05-05 18:49:23 +0000
@@ -40,6 +40,7 @@
 from contrib.testing.testcase import (
     BaseTwistedTestCase, MementoHandler, DummyClass
 )
+from contrib.mocker import Mocker
 
 from ubuntuone.storageprotocol import client, errors, protocol_pb2, volumes
 from ubuntuone.syncdaemon import states
@@ -47,8 +48,10 @@
 from ubuntuone.syncdaemon.main import Main
 from ubuntuone.syncdaemon.action_queue import (
     ActionQueue, ActionQueueCommand, ChangePublicAccess, CreateUDF,
-    DeleteVolume, ListDir, ListVolumes, NoisyRequestQueue, RequestQueue,
-    Upload, CreateShare, GetPublicFiles,
+    DeleteVolume, Download, GetContentMixin, ListDir, ListVolumes,
+    NoisyRequestQueue, RequestQueue, UploadProgressWrapper, Upload,
+    CreateShare, GetPublicFiles,
+    TRANSFER_PROGRESS_THRESHOLD
 )
 from ubuntuone.syncdaemon.event_queue import EventQueue, EVENTS
 from ubuntuone.syncdaemon.volume_manager import UDF
@@ -1262,6 +1265,49 @@
         self.assertTrue(res is None)
 
 
+class GetContentMixinTestCase(ConnectedBaseTestCase):
+    """Test for GetContentMixin ActionQueueCommand subclass."""
+
+    def setUp(self):
+        """Init."""
+        res = super(GetContentMixinTestCase, self).setUp()
+
+        self.rq = RequestQueue(name='FOO', action_queue=self.action_queue)
+        return res
+
+    def test_progress(self):
+        """Test progress_hook invocation."""
+        mocker = Mocker()
+        gunzip = mocker.mock()
+        gunzip.decompress('some compressed bytes')
+        mocker.result('some bytes')
+        mocker.replay()
+
+        calls = []
+        class MyGetContent(GetContentMixin):
+            """Subclass to check progress_hook invocation."""
+            def progress_hook(innerself, n_bytes):
+                calls.append((innerself.fileobj.getvalue(),
+                              n_bytes))
+                # check just that is there
+                super(MyGetContent, innerself).progress_hook(n_bytes)
+
+        command = MyGetContent(self.rq, share_id='a_share_id',
+                               node_id='a_node_id', server_hash='a_server_hash',
+                               fileobj_factory=lambda: None)
+        command.gunzip = gunzip
+        command.fileobj = StringIO()
+        dloading = {
+            'n_bytes_read': 0
+        }
+        command.action_queue.downloading['a_share_id', 'a_node_id'] = dloading
+        command.cb('some compressed bytes')
+        expected = len('some compressed bytes')
+        self.assertEqual(calls, [('some bytes', expected)])
+        mocker.restore()
+        mocker.verify()
+
+
 class ListDirTestCase(ConnectedBaseTestCase):
     """Test for ListDir ActionQueueCommand."""
 
@@ -1322,12 +1368,177 @@
         self.assertTrue(res is None)
 
     def test_AQ_DOWNLOAD_DOES_NOT_EXIST_is_valid_event(self):
-        """AQ_DOWNLOAD_DOES_NOT_EXIST is a valdi event."""
+        """AQ_DOWNLOAD_DOES_NOT_EXIST is a valid event."""
         event = 'AQ_DOWNLOAD_DOES_NOT_EXIST'
         self.assertTrue(event in EVENTS)
         self.assertEquals(('share_id', 'node_id'), EVENTS[event])
 
 
+class DownloadUnconnectedTestCase(FactoryBaseTestCase):
+    """Test for Download ActionQueueCommand, no connection"""
+    
+    def setUp(self):
+        """Init."""
+        super(DownloadUnconnectedTestCase, self).setUp()
+
+        self.rq = request_queue = RequestQueue(name='FOO',
+                                               action_queue=self.action_queue)
+        self.command = Download(request_queue, share_id='a_share_id',
+                               node_id='a_node_id', server_hash='a_server_hash',
+                               fileobj_factory=lambda: None)
+        self.command.start_unqueued() # create the logger
+
+    def test_progress_information_setup(self):
+        """Test the setting up of the progress information in ._run()."""
+        calls = []
+        d = defer.Deferred()
+        class FakedRequest():
+            """Fake Request."""
+            pass
+        class FakedClient(object):
+            """Fake Client."""
+            def get_content_request(innerself, *args, **kwds):
+                """Fake a get content request with its deferred."""
+                calls.append(kwds.get('offset'))
+                req = FakedRequest()
+                req.deferred = d
+                return req
+
+        self.command.action_queue.connect_in_progress = False
+        self.command.action_queue.client = FakedClient()
+        self.command._run()
+
+        self.assertEqual(self.command.n_bytes_read_last, 0)
+        self.assertEqual(calls, [0])
+
+        calls = []
+        dloading = self.command.action_queue.downloading['a_share_id',
+                                                         'a_node_id']
+        dloading['n_bytes_read'] = 20
+        self.command._run()
+        
+        self.assertEqual(self.command.n_bytes_read_last, 20)        
+        self.assertEqual(calls, [20])
+        
+class DownloadTestCase(ConnectedBaseTestCase):
+    """Test for Download ActionQueueCommand."""
+
+    def setUp(self):
+        """Init."""
+        res = super(DownloadTestCase, self).setUp()
+
+        request_queue = RequestQueue(name='FOO', action_queue=self.action_queue)
+        self.command = Download(request_queue, share_id='a_share_id',
+                               node_id='a_node_id', server_hash='a_server_hash',
+                               fileobj_factory=lambda: None)
+        self.command.start_unqueued() # create the logger
+
+        return res
+
+    def test_AQ_DOWNLOAD_FILE_PROGRESS_is_valid_event(self):
+        """AQ_DOWNLOAD_FILE_PROGRESS is a valid event."""
+        event = 'AQ_DOWNLOAD_FILE_PROGRESS'
+        self.assertTrue(event in EVENTS)
+        self.assertEqual(('share_id', 'node_id', 'n_bytes_read',
+                          'deflated_size'), EVENTS[event])
+
+    def test_progress_hook(self):
+        # would first get the node attribute including this
+        self.command.action_queue.downloading['a_share_id', 'a_node_id'] = {}
+        self.command.nacb(deflated_size = 2*TRANSFER_PROGRESS_THRESHOLD)
+        self.command.progress_start(0)
+
+        self.command.progress_hook(5)
+        self.assertEqual([], self.command.action_queue.event_queue.events)
+        self.assertEqual(self.command.n_bytes_read_last, 5)
+
+        self.command.progress_hook(TRANSFER_PROGRESS_THRESHOLD)
+        self.assertEqual([], self.command.action_queue.event_queue.events)
+        self.assertEqual(self.command.n_bytes_read_last,
+                         TRANSFER_PROGRESS_THRESHOLD)
+
+        self.command.progress_start(5)
+        self.command.progress_hook(TRANSFER_PROGRESS_THRESHOLD+5)
+        kwargs = {'share_id': 'a_share_id', 'node_id': 'a_node_id',
+                  'deflated_size': 2*TRANSFER_PROGRESS_THRESHOLD,
+                  'n_bytes_read': 5+TRANSFER_PROGRESS_THRESHOLD}
+        events = [('AQ_DOWNLOAD_FILE_PROGRESS', (), kwargs)]
+        self.assertEqual(events, self.command.action_queue.event_queue.events)
+        self.assertEqual(self.command.n_bytes_read_last,
+                         5+TRANSFER_PROGRESS_THRESHOLD)
+
+
+class UploadUnconnectedTestCase(FactoryBaseTestCase):
+    """Test for Upload ActionQueueCommand, no connection"""
+    
+    def setUp(self):
+        """Init."""
+        super(UploadUnconnectedTestCase, self).setUp()
+
+        self.rq = request_queue = RequestQueue(name='FOO',
+                                               action_queue=self.action_queue)
+        self.command = Upload(request_queue, share_id='a_share_id',
+                              node_id='a_node_id', previous_hash='prev_hash',
+                              hash='yadda', crc32=0, size=0,
+                              fileobj_factory=lambda: None,
+                              tempfile_factory=lambda: None)
+        self.command.start_unqueued() # create the logger
+
+    def test_upload_progress_wrapper_setup(self):
+        """Test the setting up of the progress wrapper in ._run()."""
+        calls = []
+        d = defer.Deferred()
+        class FakedRequest():
+            """Fake Request."""
+            pass
+        class FakedClient(object):
+            """Fake Client."""
+            def put_content_request(innerself, *args):
+                """Fake a put content request with its deferred."""
+                calls.append(args)
+                req = FakedRequest()
+                req.deferred = d
+                return req
+
+        self.command.action_queue.connect_in_progress = False
+        self.command.action_queue.client = FakedClient()
+        self.command._run()
+
+        self.assertEqual(len(calls), 1)
+        upload_wrapper = calls[0][-1]
+        uploading = self.action_queue.uploading['a_share_id', 'a_node_id']
+        self.assertTrue(upload_wrapper.data_dict is uploading)
+        self.assertEqual(upload_wrapper.progress_hook,
+                         self.command.progress_hook)
+        self.assertEqual(self.command.n_bytes_written_last, 0)
+
+        self.command.n_bytes_written_last = 20
+        self.command._run()
+        self.assertEqual(self.command.n_bytes_written_last, 0)        
+
+class UploadProgressWrapperTestCase(BaseTwistedTestCase):
+    """Test for the UploadProgressWrapper helper class."""
+
+    def test_read(self):
+        """Test the read method."""
+        progress = []
+        
+        def progress_hook(n_bytes_written):
+            progress.append(n_bytes_written)
+
+        info = {'n_bytes_written': 0}
+        f = StringIO("x"*10+"y"*5)
+        upw = UploadProgressWrapper(f, info, progress_hook)
+
+        res = upw.read(10)
+        self.assertEqual(res, "x"*10)
+        self.assertEqual(progress, [0])
+
+        res = upw.read(5)
+        self.assertEqual(res, "y"*5)        
+        self.assertEqual(progress, [0, 10])        
+        
+
 class UploadTestCase(ConnectedBaseTestCase):
     """Test for Upload ActionQueueCommand."""
 
@@ -1411,6 +1622,38 @@
         events = [('AQ_UPLOAD_ERROR', (), kwargs)]
         self.assertEquals(events, self.command.action_queue.event_queue.events)
 
+    def test_AQ_UPLOAD_FILE_PROGRESS_is_valid_event(self):
+        """AQ_UPLOAD_FILE_PROGRESS is a valid event."""
+        event = 'AQ_UPLOAD_FILE_PROGRESS'
+        self.assertTrue(event in EVENTS)
+        self.assertEqual(('share_id', 'node_id', 'n_bytes_written',
+                          'deflated_size'), EVENTS[event])
+
+    def test_progress_hook(self):
+        """Test the progress hook."""
+        self.command.deflated_size = 2*TRANSFER_PROGRESS_THRESHOLD
+        self.command.n_bytes_written_last = 0
+        
+        self.command.progress_hook(5)
+        self.assertEqual([], self.command.action_queue.event_queue.events)
+        self.assertEqual(self.command.n_bytes_written_last, 5)
+
+        self.command.progress_hook(TRANSFER_PROGRESS_THRESHOLD)
+        self.assertEqual([], self.command.action_queue.event_queue.events)
+        self.assertEqual(self.command.n_bytes_written_last,
+                         TRANSFER_PROGRESS_THRESHOLD)        
+        
+        self.command.n_bytes_written_last = 5
+        self.command.progress_hook(TRANSFER_PROGRESS_THRESHOLD+14)        
+
+        kwargs = {'share_id': 'a_share_id', 'node_id': 'a_node_id',
+                  'deflated_size': 2*TRANSFER_PROGRESS_THRESHOLD,
+                  'n_bytes_written': 14+TRANSFER_PROGRESS_THRESHOLD }
+        events = [('AQ_UPLOAD_FILE_PROGRESS', (), kwargs)]
+        self.assertEqual(events, self.command.action_queue.event_queue.events)
+        self.assertEqual(self.command.n_bytes_written_last,
+                         14+TRANSFER_PROGRESS_THRESHOLD)
+
 
 class CreateShareTestCase(ConnectedBaseTestCase):
     """Test for CreateShare ActionQueueCommand."""

=== modified file 'tests/syncdaemon/test_dbus.py'
--- tests/syncdaemon/test_dbus.py	2010-04-09 18:21:13 +0000
+++ tests/syncdaemon/test_dbus.py	2010-05-05 18:49:23 +0000
@@ -981,6 +981,30 @@
         self.main.event_q.push('AQ_DOWNLOAD_STARTED', '', 'node_id', '')
         return d
 
+    def test_download_file_progress(self):
+        """Test the DBus signals in Status."""
+        a_dir = os.path.join(self.root_dir, u'ñoño'.encode('utf-8'))
+        self.fs_manager.create(a_dir, "", is_dir=False)
+        self.fs_manager.set_node_id(a_dir, "node_id")
+
+        d = defer.Deferred()
+        def download_handler(path, info):
+            """Handler for DownloadFileProgress signal."""
+            self.assertEqual(a_dir, path.encode('utf-8'))
+            self.assertEqual(info, {'n_bytes_read': '10',
+                                    'deflated_size': '20'})
+            d.callback(True)
+
+        match = self.bus.add_signal_receiver(download_handler,
+                                     signal_name='DownloadFileProgress')
+        self.signal_receivers.add(match)
+        self.main.event_q.push('AQ_DOWNLOAD_FILE_PROGRESS',
+                               share_id='',
+                               node_id='node_id',
+                               n_bytes_read=10,
+                               deflated_size=20)
+        return d
+    
     def test_download_finished(self):
         """ Test the DBus signals in Status """
         a_dir = os.path.join(self.root_dir, u'ñoño'.encode('utf-8'))
@@ -1056,6 +1080,30 @@
         self.main.event_q.push('AQ_UPLOAD_STARTED', '', 'node_id', '')
         return d
 
+    def test_upload_file_progress(self):
+        """Test the DBus signals in Status."""
+        a_dir = os.path.join(self.root_dir, u'ñoño'.encode('utf-8'))
+        self.fs_manager.create(a_dir, "", is_dir=False)
+        self.fs_manager.set_node_id(a_dir, "node_id")
+
+        d = defer.Deferred()
+        def upload_handler(path, info):
+            """Handler for UploadFileProgress signal."""
+            self.assertEqual(a_dir, path.encode('utf-8'))
+            self.assertEqual(info, {'n_bytes_written': '10',
+                                     'deflated_size': '20'})
+            d.callback(True)
+
+        match = self.bus.add_signal_receiver(upload_handler,
+                                     signal_name='UploadFileProgress')
+        self.signal_receivers.add(match)
+        self.main.event_q.push('AQ_UPLOAD_FILE_PROGRESS',
+                               share_id='',
+                               node_id='node_id',
+                               n_bytes_written=10,
+                               deflated_size=20)
+        return d
+
     def test_upload_finished(self):
         """ Test the DBus signals in Status """
         a_dir = os.path.join(self.root_dir, u'ñoño'.encode('utf-8'))

=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2010-05-03 23:40:25 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2010-05-05 18:49:23 +0000
@@ -66,6 +66,8 @@
 # Regular expression to validate an e-mail address
 EREGEX = "^.+\\@(\\[?)[a-zA-Z0-9\\-\\.]+\\.([a-zA-Z]{2,3}|[0-9]{1,3})(\\]?)$"
 
+# progress threshold to emit a download/upload progress event: 64Kb
+TRANSFER_PROGRESS_THRESHOLD = 64*1024*1024
 
 def passit(func):
     """Pass the value on for the next deferred, while calling func with it."""
@@ -598,13 +600,13 @@
     """A wrapper around the file-like object used for Uploads.
 
     It can be used to keep track of the number of bytes that have been
-    written to the store.
+    written to the store and invokes a hook on progress.
 
     """
 
-    __slots__ = ('fd', 'data_dict', 'n_bytes_read')
+    __slots__ = ('fd', 'data_dict', 'n_bytes_read', 'progress_hook')
 
-    def __init__(self, fd, data_dict):
+    def __init__(self, fd, data_dict, progress_hook):
         """
         fd is the file-like object used for uploads. data_dict is the
         entry in the uploading dictionary.
@@ -612,6 +614,7 @@
         self.fd = fd
         self.data_dict = data_dict
         self.n_bytes_read = 0
+        self.progress_hook = progress_hook
 
     def read(self, size=None):
         """
@@ -623,6 +626,8 @@
         latter is done directly in the data_dict.
         """
         self.data_dict['n_bytes_written'] = self.n_bytes_read
+        self.progress_hook(self.n_bytes_read)
+        
         data = self.fd.read(size)
         self.n_bytes_read += len(data)
         return data
@@ -2192,7 +2197,7 @@
 
     __slots__ = ('share_id', 'node_id', 'server_hash', 'fileobj_factory',
                  'fileobj', 'gunzip', 'marker_maybe', 'cancelled',
-                 'download_req')
+                 'download_req', 'deflated_size')
     logged_attrs = ('share_id', 'node_id', 'server_hash', 'fileobj_factory')
 
     def __init__(self, request_queue, share_id, node_id, server_hash,
@@ -2243,7 +2248,7 @@
                                                         'command': self}
         assert downloading[self.share_id, self.node_id]['command'] is self
         offset = downloading[self.share_id, self.node_id]['n_bytes_read']
-
+        self.progress_start(offset)
         self.action_queue.event_queue.push('AQ_DOWNLOAD_STARTED',
                                            share_id=self.share_id,
                                            node_id=self.node_id,
@@ -2318,9 +2323,19 @@
         self.fileobj.write(self.gunzip.decompress(bytes))
         self.fileobj.flush()     # not strictly necessary but nice to
                                  # see the downloaded size
-
+        self.progress_hook(dloading['n_bytes_read'])
+
+    def progress_start(self, n_bytes_read_already):
+        """Hook to start tracking progress."""
+        pass
+
+    def progress_hook(self, n_bytes_read):
+        """Hook to track progress."""
+        pass
+        
     def nacb(self, **kwargs):
         """Set the node attrs in the 'currently downloading' dict."""
+        self.deflated_size = kwargs['deflated_size']
         self.action_queue.downloading[self.share_id,
                                       self.node_id].update(kwargs)
 
@@ -2353,10 +2368,29 @@
 class Download(GetContentMixin, ActionQueueCommand):
     """Get the contents of a file."""
 
-    __slots__ = ()
+    __slots__ = ('n_bytes_read_last')
 
     is_dir = False
 
+    def progress_start(self, n_bytes_read_already):
+        """Start tracking progress.
+
+        Consider that n_bytes_read_already have been already read.
+        """
+        self.n_bytes_read_last = n_bytes_read_already
+
+    def progress_hook(self, n_bytes_read):
+        """Convert downloading progress into an event."""
+        n_bytes_read_last = self.n_bytes_read_last
+        self.n_bytes_read_last = n_bytes_read
+        # produce an event only if there has been a threshold-sized progress
+        if n_bytes_read - n_bytes_read_last < TRANSFER_PROGRESS_THRESHOLD:
+            return
+        self.action_queue.event_queue.push('AQ_DOWNLOAD_FILE_PROGRESS',
+                                      share_id=self.share_id,
+                                      node_id=self.node_id,
+                                      n_bytes_read=n_bytes_read,
+                                      deflated_size=self.deflated_size)
 
 class Upload(ActionQueueCommand):
     """Upload stuff to a file."""
@@ -2364,7 +2398,7 @@
     __slots__ = ('share_id', 'node_id', 'previous_hash', 'hash', 'crc32',
                  'size', 'fileobj_factory', 'tempfile_factory',
                  'deflated_size', 'tempfile', 'cancelled', 'upload_req',
-                 'marker_maybe')
+                 'marker_maybe', 'n_bytes_written_last')
 
     logged_attrs = ('share_id', 'node_id', 'previous_hash', 'hash', 'crc32',
                     'size', 'fileobj_factory')
@@ -2389,6 +2423,7 @@
         self.marker_maybe = None
         if (self.share_id, self.node_id) in self.action_queue.uploading:
             self.action_queue.cancel_upload(self.share_id, self.node_id)
+        self.n_bytes_written_last = None # set by _run
 
     def _is_runnable(self):
         """Returns True if there is sufficient space available to complete
@@ -2451,7 +2486,8 @@
 
         if getattr(self.tempfile, 'name', None) is not None:
             self.tempfile = open(self.tempfile.name)
-        f = UploadProgressWrapper(self.tempfile, uploading)
+        self.n_bytes_written_last = 0
+        f = UploadProgressWrapper(self.tempfile, uploading, self.progress_hook)
         req = self.action_queue.client.put_content_request(
             self.share_id, self.node_id, self.previous_hash, self.hash,
             self.crc32, self.size, self.deflated_size, f)
@@ -2465,6 +2501,19 @@
         d.addBoth(passit(lambda _: self.tempfile.close()))
         return d
 
+    def progress_hook(self, n_bytes_written):
+        """Convert uploading progress into an event."""
+        n_bytes_written_last = self.n_bytes_written_last
+        self.n_bytes_written_last = n_bytes_written
+        # produce an event only if there has been a threshold-sized progress
+        if n_bytes_written - n_bytes_written_last < TRANSFER_PROGRESS_THRESHOLD:
+            return
+        self.action_queue.event_queue.push('AQ_UPLOAD_FILE_PROGRESS',
+                                      share_id=self.share_id,
+                                      node_id=self.node_id,
+                                      n_bytes_written=n_bytes_written,
+                                      deflated_size=self.deflated_size)
+
     def handle_success(self, _):
         """
         It worked! Push the event.

=== modified file 'ubuntuone/syncdaemon/dbus_interface.py'
--- ubuntuone/syncdaemon/dbus_interface.py	2010-04-29 20:56:27 +0000
+++ ubuntuone/syncdaemon/dbus_interface.py	2010-05-05 18:49:23 +0000
@@ -276,6 +276,12 @@
 
     @dbus.service.signal(DBUS_IFACE_STATUS_NAME,
                          signature='sa{ss}')
+    def DownloadFileProgress(self, path, info):
+        """Fire a D-BUS signal, notifying about a download progress."""
+        pass    
+
+    @dbus.service.signal(DBUS_IFACE_STATUS_NAME,
+                         signature='sa{ss}')
     def DownloadFinished(self, path, info):
         """ Fire a D-BUS signal, notifying a download has finished.  """
         pass
@@ -287,6 +293,12 @@
 
     @dbus.service.signal(DBUS_IFACE_STATUS_NAME,
                          signature='sa{ss}')
+    def UploadFileProgress(self, path, info):
+        """Fire a D-BUS signal, notifying about an upload progress."""
+        pass    
+
+    @dbus.service.signal(DBUS_IFACE_STATUS_NAME,
+                         signature='sa{ss}')
     def UploadFinished(self, path, info):
         """ Fire a D-BUS signal, notifying an upload has finished.  """
         pass
@@ -356,8 +368,14 @@
         """ Emits the signal """
         self.DownloadStarted(download)
 
+    def emit_download_file_progress(self, download, **info):
+        """Emits the signal."""
+        for k, v in info.copy().items():
+            info[str(k)] = str(v)
+        self.DownloadFileProgress(download, info)
+
     def emit_download_finished(self, download, **info):
-        """ Emits the signal """
+        """Emits the signal."""
         for k, v in info.copy().items():
             info[str(k)] = str(v)
         self.DownloadFinished(download, info)
@@ -366,6 +384,12 @@
         """ Emits the signal """
         self.UploadStarted(upload)
 
+    def emit_upload_file_progress(self, upload, **info):
+        """Emits the signal."""
+        for k, v in info.copy().items():
+            info[str(k)] = str(v)
+        self.UploadFileProgress(upload, info)
+
     def emit_upload_finished(self, upload, **info):
         """ Emits the signal """
         for k, v in info.copy().items():
@@ -457,6 +481,28 @@
                         node_id=str(node_id))
             self.dbus_iface.status.emit_signal_error('DownloadStarted', args)
 
+    def handle_AQ_DOWNLOAD_FILE_PROGRESS(self, share_id, node_id,
+                                         n_bytes_read, deflated_size):
+        """Handle AQ_DOWNLOAD_FILE_PROGRESS."""
+        self.handle_default('AQ_DOWNLOAD_FILE_PROGRESS', share_id, node_id,
+                            n_bytes_read, deflated_size)
+        try:
+            mdobj = self.dbus_iface.fs_manager.get_by_node_id(share_id, node_id)
+        except KeyError, e:
+            args = dict(message='The md is gone before sending '
+                        'DownloadFileProgress signal',
+                        error=str(e),
+                        share_id=str(share_id),
+                        node_id=str(node_id))
+            self.dbus_iface.status.emit_signal_error('DownloadFileProgress',
+                                                     args)
+        else:
+            path = self.dbus_iface.fs_manager.get_abspath(share_id, mdobj.path)
+            self.dbus_iface.status.emit_download_file_progress(path,
+                                                 n_bytes_read=n_bytes_read,
+                                                 deflated_size=deflated_size
+                                                               )
+            
     def handle_AQ_DOWNLOAD_FINISHED(self, share_id, node_id, server_hash):
         """ handle AQ_DOWNLOAD_FINISHED """
         self.handle_default('AQ_DOWNLOAD_FINISHED', share_id,
@@ -519,6 +565,28 @@
                         node_id=str(node_id))
             self.dbus_iface.status.emit_signal_error('UploadStarted', args)
 
+    def handle_AQ_UPLOAD_FILE_PROGRESS(self, share_id, node_id,
+                                         n_bytes_written, deflated_size):
+        """Handle AQ_UPLOAD_FILE_PROGRESS."""
+        self.handle_default('AQ_UPLOAD_FILE_PROGRESS', share_id, node_id,
+                            n_bytes_written, deflated_size)
+        try:
+            mdobj = self.dbus_iface.fs_manager.get_by_node_id(share_id, node_id)
+        except KeyError, e:
+            args = dict(message='The md is gone before sending '
+                        'UploadFileProgress signal',
+                        error=str(e),
+                        share_id=str(share_id),
+                        node_id=str(node_id))
+            self.dbus_iface.status.emit_signal_error('UploadFileProgress',
+                                                     args)
+        else:
+            path = self.dbus_iface.fs_manager.get_abspath(share_id, mdobj.path)
+            self.dbus_iface.status.emit_upload_file_progress(path,
+                                                n_bytes_written=n_bytes_written,
+                                                deflated_size=deflated_size
+                                                             )
+            
     def handle_AQ_UPLOAD_FINISHED(self, share_id, node_id, hash):
         """ handle AQ_UPLOAD_FINISHED """
         self.handle_default('AQ_UPLOAD_FINISHED', share_id, node_id, hash)

=== modified file 'ubuntuone/syncdaemon/event_queue.py'
--- ubuntuone/syncdaemon/event_queue.py	2010-04-12 17:58:20 +0000
+++ ubuntuone/syncdaemon/event_queue.py	2010-05-05 18:49:23 +0000
@@ -53,12 +53,16 @@
     'AQ_UNLINK_OK': ('share_id', 'parent_id', 'node_id'),
     'AQ_UNLINK_ERROR': ('share_id', 'parent_id', 'node_id', 'error'),
     'AQ_DOWNLOAD_STARTED': ('share_id', 'node_id', 'server_hash'),
+    'AQ_DOWNLOAD_FILE_PROGRESS': ('share_id', 'node_id',
+                                  'n_bytes_read', 'deflated_size'),
     'AQ_DOWNLOAD_COMMIT': ('share_id', 'node_id', 'server_hash'),
     'AQ_DOWNLOAD_FINISHED': ('share_id', 'node_id', 'server_hash'),
     'AQ_DOWNLOAD_ERROR': ('share_id', 'node_id', 'server_hash', 'error'),
     'AQ_DOWNLOAD_CANCELLED': ('share_id', 'node_id', 'server_hash'),
     'AQ_DOWNLOAD_DOES_NOT_EXIST': ('share_id', 'node_id'),
     'AQ_UPLOAD_STARTED' : ('share_id', 'node_id', 'hash'),
+    'AQ_UPLOAD_FILE_PROGRESS': ('share_id', 'node_id',
+                                'n_bytes_written', 'deflated_size'),
     'AQ_UPLOAD_FINISHED': ('share_id', 'node_id', 'hash'),
     'AQ_UPLOAD_ERROR': ('share_id', 'node_id', 'error', 'hash'),
     'AQ_SHARES_LIST': ('shares_list',),