Ticket #758: tube-new-api.patch

readactivity.py

@@ -25 +25 @@
 import evince
 import gobject
 import gtk
-import telepathy
+
+from telepathy.client import Channel
+from telepathy.interfaces import CONNECTION_INTERFACE_REQUESTS, CHANNEL_INTERFACE
+from telepathy.constants import CONNECTION_HANDLE_TYPE_ROOM, SOCKET_ADDRESS_TYPE_IPV4,\
+    SOCKET_ACCESS_CONTROL_LOCALHOST
+
+from dbus import PROPERTIES_IFACE
 
 from sugar.activity import activity
 from sugar import network
@@ -44 +50 @@
 
 _logger = logging.getLogger('read-activity')
 
+# TODO: import when tube API is stable
+CHANNEL_INTERFACE_TUBE = CHANNEL_INTERFACE + ".Interface.Tube.DRAFT"
+CHANNEL_TYPE_STREAM_TUBE = CHANNEL_INTERFACE + ".Type.StreamTube.DRAFT"
+
+TUBE_CHANNEL_STATE_LOCAL_PENDING = 0
+TUBE_CHANNEL_STATE_REMOTE_PENDING = 1
+TUBE_CHANNEL_STATE_OPEN = 2
+TUBE_CHANNEL_STATE_NOT_OFFERED = 3
+
 def _get_screen_dpi():
     xft_dpi = gtk.settings_get_default().get_property('gtk-xft-dpi')
     _logger.debug('Setting dpi to %f', float(xft_dpi / 1024))
@@ -155 +170 @@
         # start with sleep off
         self._sleep_inhibit = True
 
+        # set of paths of tube channel that could be used to download the file
         self.unused_download_tubes = set()
         self._want_document = True
         self._download_content_length = 0
@@ -338 +354 @@
         self._close_requested = True
         return True
 
-    def _download_result_cb(self, getter, tempfile, suggested_name, tube_id):
+    def _download_result_cb(self, getter, tempfile, suggested_name, tube_path):
         if self._download_content_type == 'text/html':
             # got an error page instead
-            self._download_error_cb(getter, 'HTTP Error', tube_id)
+            self._download_error_cb(getter, 'HTTP Error', tube_path)
             return
 
         del self.unused_download_tubes
@@ -354 +370 @@
         self._jobject.file_path = file_path
         datastore.write(self._jobject, transfer_ownership=True)
 
-        _logger.debug("Got document %s (%s) from tube %u",
-                      tempfile, suggested_name, tube_id)
+        _logger.debug("Got document %s (%s) from tube %s",
+                      tempfile, suggested_name, tube_path)
         self._load_document("file://%s" % tempfile)
         self.save()
 
-    def _download_progress_cb(self, getter, bytes_downloaded, tube_id):
+    def _download_progress_cb(self, getter, bytes_downloaded, tube_path):
         # FIXME: Draw a progress bar
         if self._download_content_length > 0:
-            _logger.debug("Downloaded %u of %u bytes from tube %u...",
+            _logger.debug("Downloaded %u of %u bytes from tube %s...",
                           bytes_downloaded, self._download_content_length, 
-                          tube_id)
+                          tube_path)
         else:
-            _logger.debug("Downloaded %u bytes from tube %u...",
-                          bytes_downloaded, tube_id)
+            _logger.debug("Downloaded %u bytes from tube %s...",
+                          bytes_downloaded, tube_path)
 
-    def _download_error_cb(self, getter, err, tube_id):
-        _logger.debug("Error getting document from tube %u: %s",
-                      tube_id, err)
+    def _download_error_cb(self, getter, err, tube_path):
+        _logger.debug("Error getting document from tube %s: %s",
+                      tube_path, err)
         self._want_document = True
         self._download_content_length = 0
         self._download_content_type = None
         gobject.idle_add(self._get_document)
 
-    def _download_document(self, tube_id, path):
+    def _download_document(self, tube_path, path):
         # FIXME: should ideally have the CM listen on a Unix socket
         # instead of IPv4 (might be more compatible with Rainbow)
-        chan = self.shared_activity.telepathy_tubes_chan
-        iface = chan[telepathy.CHANNEL_TYPE_TUBES]
-        addr = iface.AcceptStreamTube(tube_id,
-                telepathy.SOCKET_ADDRESS_TYPE_IPV4,
-                telepathy.SOCKET_ACCESS_CONTROL_LOCALHOST, 0,
-                utf8_strings=True)
-        _logger.debug('Accepted stream tube: listening address is %r', addr)
-        # SOCKET_ADDRESS_TYPE_IPV4 is defined to have addresses of type '(sq)'
-        assert isinstance(addr, dbus.Struct)
-        assert len(addr) == 2
-        assert isinstance(addr[0], str)
-        assert isinstance(addr[1], (int, long))
-        assert addr[1] > 0 and addr[1] < 65536
-        port = int(addr[1])
-
-        getter = ReadURLDownloader("http://%s:%d/document"
-                                           % (addr[0], port))
-        getter.connect("finished", self._download_result_cb, tube_id)
-        getter.connect("progress", self._download_progress_cb, tube_id)
-        getter.connect("error", self._download_error_cb, tube_id)
-        _logger.debug("Starting download to %s...", path)
-        getter.start(path)
-        self._download_content_length = getter.get_content_length()
-        self._download_content_type = getter.get_content_type()
+
+        conn = self.shared_activity.telepathy_conn
+        tube_chan = Channel(conn.dbus_proxy.bus_name, tube_path)
+
+        def accept_stream_tube_reply_cb(addr):
+            _logger.debug('Accepted stream tube: listening address is %r', addr)
+
+            # SOCKET_ADDRESS_TYPE_IPV4 is defined to have addresses of type '(sq)'
+            assert isinstance(addr, dbus.Struct)
+            assert len(addr) == 2
+            assert isinstance(addr[0], str)
+            assert isinstance(addr[1], (int, long))
+            assert addr[1] > 0 and addr[1] < 65536
+            port = int(addr[1])
+
+            getter = ReadURLDownloader("http://%s:%d/document"
+                                               % (addr[0], port))
+            getter.connect("finished", self._download_result_cb, tube_path)
+            getter.connect("progress", self._download_progress_cb, tube_path)
+            getter.connect("error", self._download_error_cb, tube_path)
+            _logger.debug("Starting download to %s...", path)
+            getter.start(path)
+            self._download_content_length = getter.get_content_length()
+            self._download_content_type = getter.get_content_type()
+
+        def accept_stream_tube_error_cb(e):
+            _logger.error('OfferStreamTube failed: %s' % e)
+
+        addr = tube_chan[CHANNEL_TYPE_STREAM_TUBE].AcceptStreamTube(
+                SOCKET_ADDRESS_TYPE_IPV4, SOCKET_ACCESS_CONTROL_LOCALHOST, 0,
+                utf8_strings=True,
+                reply_handler=accept_stream_tube_reply_cb,
+                error_handler=accept_stream_tube_error_cb)
+
         return False
 
     def _get_document(self):
@@ -419 +445 @@
 
         # Pick an arbitrary tube we can try to download the document from
         try:
-            tube_id = self.unused_download_tubes.pop()
+            tube_path = self.unused_download_tubes.pop()
         except (ValueError, KeyError), e:
             _logger.debug('No tubes to get the document from right now: %s',
                           e)
@@ -427 +453 @@
 
         # Avoid trying to download the document multiple times at once
         self._want_document = False
-        gobject.idle_add(self._download_document, tube_id, path)
+        gobject.idle_add(self._download_document, tube_path, path)
         return False
 
     def _joined_cb(self, also_self):
@@ -500 +526 @@
             self._tempfile)
 
         # Make a tube for it
-        chan = self.shared_activity.telepathy_tubes_chan
-        iface = chan[telepathy.CHANNEL_TYPE_TUBES]
-        self._fileserver_tube_id = iface.OfferStreamTube(READ_STREAM_SERVICE,
-                {},
-                telepathy.SOCKET_ADDRESS_TYPE_IPV4,
+        conn = self.shared_activity.telepathy_conn
+        room_handle = self.shared_activity.telepathy_room_handle
+
+        def create_tube_reply_cb(path, props):
+            _logger.debug("tube channel created. Offer the tube")
+
+            self._tube_chan = Channel(conn.dbus_proxy.bus_name, path)
+
+            self._tube_chan[CHANNEL_TYPE_STREAM_TUBE].OfferStreamTube(SOCKET_ADDRESS_TYPE_IPV4,
                 ('127.0.0.1', dbus.UInt16(self.port)),
-                telepathy.SOCKET_ACCESS_CONTROL_LOCALHOST, 0)
+                SOCKET_ACCESS_CONTROL_LOCALHOST, 0, {})
+
+        def create_tube_error_cb(e):
+            _logger.error('Tube channel creation failed: %s' % e)
+
+        conn[CONNECTION_INTERFACE_REQUESTS].CreateChannel({
+            CHANNEL_INTERFACE + ".ChannelType": CHANNEL_TYPE_STREAM_TUBE,
+            CHANNEL_INTERFACE + ".TargetHandleType": CONNECTION_HANDLE_TYPE_ROOM,
+            CHANNEL_INTERFACE + ".TargetHandle": room_handle,
+            CHANNEL_TYPE_STREAM_TUBE + ".Service": READ_STREAM_SERVICE},
+            reply_handler=create_tube_reply_cb,
+            error_handler=create_tube_error_cb)
 
     def watch_for_tubes(self):
         """Watch for new tubes."""
-        tubes_chan = self.shared_activity.telepathy_tubes_chan
-
-        tubes_chan[telepathy.CHANNEL_TYPE_TUBES].connect_to_signal('NewTube',
-            self._new_tube_cb)
-        tubes_chan[telepathy.CHANNEL_TYPE_TUBES].ListTubes(
-            reply_handler=self._list_tubes_reply_cb,
-            error_handler=self._list_tubes_error_cb)
-
-    def _new_tube_cb(self, tube_id, initiator, tube_type, service, params,
-                     state):
-        """Callback when a new tube becomes available."""
-        _logger.debug('New tube: ID=%d initator=%d type=%d service=%s '
-                      'params=%r state=%d', tube_id, initiator, tube_type,
-                      service, params, state)
-        if self._document is None and service == READ_STREAM_SERVICE:
-            _logger.debug('I could download from that tube')
-            self.unused_download_tubes.add(tube_id)
-            # if no download is in progress, let's fetch the document
-            if self._want_document:
-                gobject.idle_add(self._get_document)
-
-    def _list_tubes_reply_cb(self, tubes):
-        """Callback when new tubes are available."""
-        for tube_info in tubes:
-            self._new_tube_cb(*tube_info)
-
-    def _list_tubes_error_cb(self, e):
-        """Handle ListTubes error by logging."""
-        _logger.error('ListTubes() failed: %s', e)
+        conn = self.shared_activity.telepathy_conn
+
+        conn[CONNECTION_INTERFACE_REQUESTS].connect_to_signal('NewChannels',
+            self.__new_channels_cb)
+
+        # look for existing tube channels
+        conn[PROPERTIES_IFACE].Get(CONNECTION_INTERFACE_REQUESTS, 'Channels',
+            reply_handler=self.__get_channels_reply_cb,
+            error_handler=self.__get_channels_error_cb)
+
+    def __new_channels_cb(self, channels):
+        """Callback when a new channel is created. We are interested about the stream tube ones"""
+        for path, props in channels:
+            if props[CHANNEL_INTERFACE + ".ChannelType"] != CHANNEL_TYPE_STREAM_TUBE:
+                continue
+
+            _logger.debug('New tube: initator=%s service=%s',
+                props[CHANNEL_INTERFACE + '.InitiatorID'],
+                props[CHANNEL_TYPE_STREAM_TUBE + '.Service'])
+
+            if self._document is None and props[CHANNEL_TYPE_STREAM_TUBE + '.Service'] == READ_STREAM_SERVICE:
+                _logger.debug('I could download from that tube')
+                self.unused_download_tubes.add(path)
+                # if no download is in progress, let's fetch the document
+                if self._want_document:
+                    gobject.idle_add(self._get_document)
+
+    def __get_channels_reply_cb(self, channels):
+        """Callback when existing channels are available."""
+        self.__new_channels_cb(channels)
+
+    def __get_channels_error_cb(self, e):
+        """Handle get channels error by logging."""
+        _logger.error('Get channels failed: %s', e)
 
     def _shared_cb(self, activityid):
         """Callback when activity shared.