diff options
| -rw-r--r-- | src/Makefile.am | 3 | ||||
| -rw-r--r-- | src/activity.py | 83 | ||||
| -rw-r--r-- | src/buddy.py | 430 | ||||
| -rw-r--r-- | src/buddyiconcache.py | 2 | ||||
| -rw-r--r-- | src/linklocal_plugin.py | 175 | ||||
| -rw-r--r-- | src/presenceservice.py | 454 | ||||
| -rw-r--r-- | src/pstest.py | 5 | ||||
| -rw-r--r-- | src/server_plugin.py | 959 | ||||
| -rw-r--r-- | src/telepathy_plugin.py | 460 |
9 files changed, 1486 insertions, 1085 deletions
diff --git a/src/Makefile.am b/src/Makefile.am index 86a4e8f..893c163 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -18,7 +18,8 @@ psdata_PYTHON = \ presenceservice.py \ pstest.py \ psutils.py \ - server_plugin.py + server_plugin.py \ + telepathy_plugin.py dist_bin_SCRIPTS = sugar-presence-service diff --git a/src/activity.py b/src/activity.py index 153d2e1..b9a1973 100644 --- a/src/activity.py +++ b/src/activity.py @@ -24,8 +24,9 @@ import logging from telepathy.client import Channel from telepathy.constants import (CHANNEL_GROUP_FLAG_CHANNEL_SPECIFIC_HANDLES, - PROPERTY_FLAG_WRITE) + PROPERTY_FLAG_WRITE, HANDLE_TYPE_ROOM) from telepathy.interfaces import (CHANNEL_INTERFACE, CHANNEL_INTERFACE_GROUP, + CHANNEL_TYPE_TEXT, CONN_INTERFACE, PROPERTIES_INTERFACE) @@ -143,6 +144,8 @@ class Activity(ExportedGObject): self._self_handle = None self._text_channel = None self._text_channel_group_flags = 0 + #: list of SignalMatch associated with the text channel, or None + self._text_channel_matches = None self._valid = False self._id = None @@ -184,6 +187,13 @@ class Activity(ExportedGObject): reply_handler=self.set_properties, error_handler=got_properties_err) + @property + def room_details(self): + """Return the Telepathy plugin on which this Activity can be joined + and the handle of the room representing it. + """ + return (self._tp, self._room) + def do_get_property(self, pspec): """Gets the value of a property associated with this activity. @@ -339,7 +349,7 @@ class Activity(ExportedGObject): unsuccessful """ - self.join(lambda unused: async_cb(), async_err_cb, False) + self.join(async_cb, async_err_cb, False) @dbus.service.method(_ACTIVITY_INTERFACE, in_signature="", out_signature="ao") @@ -420,7 +430,12 @@ class Activity(ExportedGObject): This method is called by the PresenceService on the local machine. """ - if not self._joined: + if self._joined: + _logger.debug("Ignoring alleged join to activity %s that I'm in: " + "I can already see who's there", self._id) + else: + _logger.debug("%s says they joined activity %s that I'm not in", + buddy.props.objid, self._id) self._add_buddies((buddy,)) def _add_buddies(self, buddies): @@ -470,6 +485,13 @@ class Activity(ExportedGObject): self._text_channel_group_flags |= added self._text_channel_group_flags &= ~removed + def _clean_up_matches(self): + matches = self._text_channel_matches + self._text_channel_matches = [] + if matches is not None: + for match in matches: + match.remove() + def _handle_share_join(self, text_channel): """Called when a join to a network activity was successful. @@ -481,22 +503,27 @@ class Activity(ExportedGObject): raise RuntimeError("Plugin returned invalid text channel") self._text_channel = text_channel - self._text_channel[CHANNEL_INTERFACE].connect_to_signal('Closed', + self._clean_up_matches() + + m = self._text_channel[CHANNEL_INTERFACE].connect_to_signal('Closed', self._text_channel_closed_cb) + self._text_channel_matches.append(m) if CHANNEL_INTERFACE_GROUP in self._text_channel: group = self._text_channel[CHANNEL_INTERFACE_GROUP] # FIXME: make these method calls async? - group.connect_to_signal('GroupFlagsChanged', - self._text_channel_group_flags_changed_cb) + m = group.connect_to_signal('GroupFlagsChanged', + self._text_channel_group_flags_changed_cb) + self._text_channel_matches.append(m) self._text_channel_group_flags = group.GetGroupFlags() self._self_handle = group.GetSelfHandle() # by the time we hook this, we need to know the group flags - group.connect_to_signal('MembersChanged', - self._text_channel_members_changed_cb) + m = group.connect_to_signal('MembersChanged', + self._text_channel_members_changed_cb) + self._text_channel_matches.append(m) # bootstrap by getting the current state. This is where we find # out whether anyone was lying to us in their PEP info members = set(group.GetMembers()) @@ -518,7 +545,7 @@ class Activity(ExportedGObject): def _joined_cb(self, text_channel): """XXX - not documented yet """ - self._tp.emit_joined_activity(self._id, self._room) + self._ps.owner.add_owner_activity(self._tp, self._id, self._room) verb = self._join_is_sharing and 'Share' or 'Join' @@ -527,7 +554,7 @@ class Activity(ExportedGObject): if self._join_is_sharing: self.send_properties() self._ps.owner.add_activity(self) - self._join_cb(dbus.ObjectPath(self._object_path)) + self._join_cb() _logger.debug("%s of activity %s succeeded" % (verb, self._id)) except Exception, e: self._join_failed_cb(e) @@ -569,9 +596,7 @@ class Activity(ExportedGObject): else: self._joined_cb(channel) - def _join_activity_create_channel_cb(self, handle, chan_path): - self._room = handle - + def _join_activity_create_channel_cb(self, chan_path): channel = Channel(self._tp.get_connection().service_name, chan_path) channel[PROPERTIES_INTERFACE].ListProperties( reply_handler=lambda prop_specs: @@ -579,12 +604,24 @@ class Activity(ExportedGObject): channel, prop_specs), error_handler=self._join_failed_cb) + def _join_activity_got_handles_cb(self, handles): + assert len(handles) == 1 + + self._room = handles[0] + + conn = self._tp.get_connection() + conn[CONN_INTERFACE].RequestChannel(CHANNEL_TYPE_TEXT, + HANDLE_TYPE_ROOM, self._room, True, + reply_handler=self._join_activity_create_channel_cb, + error_handler=self._join_failed_cb) + def join(self, async_cb, async_err_cb, sharing): """Local method for the local user to attempt to join the activity. - async_cb -- Callback method to be called if join attempt is successful - async_err_cb -- Callback method to be called if join attempt is - unsuccessful + async_cb -- Callback method to be called with no parameters + if join attempt is successful + async_err_cb -- Callback method to be called with an Exception + parameter if join attempt is unsuccessful The two callbacks are passed to the server_plugin ("tp") object, which in turn passes them back as parameters in a callback to the @@ -607,8 +644,18 @@ class Activity(ExportedGObject): self._join_err_cb = async_err_cb self._join_is_sharing = sharing - self._tp.join_activity(self._id, self._join_activity_create_channel_cb, - self._join_failed_cb) + if self._room: + # we're probably sharing a local activity. + # FIXME: assert that this is the case? + self._join_activity_got_handles_cb((self._room,)) + else: + conn = self._tp.get_connection() + + conn[CONN_INTERFACE].RequestHandles(HANDLE_TYPE_ROOM, + [self._tp.suggest_room_for_activity(self._id)], + reply_handler=self._join_activity_got_handles_cb, + error_handler=self._join_failed_cb) + _logger.debug("triggered share/join attempt on activity %s", self._id) def get_channels(self): diff --git a/src/buddy.py b/src/buddy.py index 68b9ac0..c1e0c9e 100644 --- a/src/buddy.py +++ b/src/buddy.py @@ -17,14 +17,30 @@ # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA import os +import logging +try: + # Python >= 2.5 + from hashlib import md5 as new_md5 +except ImportError: + from md5 import new as new_md5 + import gobject +import gtk import dbus +import dbus.proxies import dbus.service from dbus.gobject_service import ExportedGObject -import psutils +from telepathy.constants import CONNECTION_STATUS_CONNECTED +from telepathy.interfaces import (CONN_INTERFACE_ALIASING, + CONN_INTERFACE_AVATARS) from sugar import env, profile -import logging + +import psutils +from buddyiconcache import buddy_icon_cache + + +CONN_INTERFACE_BUDDY_INFO = 'org.laptop.Telepathy.BuddyInfo' _BUDDY_PATH = "/org/laptop/Sugar/Presence/Buddies/" _BUDDY_INTERFACE = "org.laptop.Sugar.Presence.Buddy" @@ -44,6 +60,41 @@ _PROP_IP4_ADDRESS = "ip4-address" _logger = logging.getLogger('s-p-s.buddy') +def _noop(*args, **kwargs): + pass + +def _buddy_icon_save_cb(buf, data): + data[0] += buf + return True + +def _get_buddy_icon_at_size(icon, maxw, maxh, maxsize): + loader = gtk.gdk.PixbufLoader() + loader.write(icon) + loader.close() + unscaled_pixbuf = loader.get_pixbuf() + del loader + + pixbuf = unscaled_pixbuf.scale_simple(maxw, maxh, gtk.gdk.INTERP_BILINEAR) + del unscaled_pixbuf + + data = [""] + quality = 90 + img_size = maxsize + 1 + while img_size > maxsize: + data = [""] + pixbuf.save_to_callback(_buddy_icon_save_cb, "jpeg", + {"quality":"%d" % quality}, data) + quality -= 10 + img_size = len(data[0]) + del pixbuf + + if img_size > maxsize: + data = [""] + raise RuntimeError("could not size image less than %d bytes" % maxsize) + + return str(data[0]) + + class Buddy(ExportedGObject): """Person on the network (tracks properties and shared activites) @@ -62,8 +113,8 @@ class Buddy(ExportedGObject): _activities -- dictionary mapping activity ID to activity.Activity objects - handles -- dictionary mapping Telepathy client plugin to - contact handle (an integer representing the JID or unique ID); + _handles -- dictionary mapping Telepathy client plugin to + tuples (contact handle, corresponding unique ID); channel-specific handles do not appear here """ @@ -71,7 +122,8 @@ class Buddy(ExportedGObject): 'validity-changed': # The buddy's validity changed. # Validity starts off False, and becomes True when the buddy - # has a color, a nick and a key. + # either has, or has tried and failed to get, a color, a nick + # and a key. # * the new validity: bool (gobject.SIGNAL_RUN_FIRST, None, [bool]), 'property-changed': @@ -90,15 +142,25 @@ class Buddy(ExportedGObject): } __gproperties__ = { - _PROP_KEY : (str, None, None, None, gobject.PARAM_READWRITE), - _PROP_ICON : (object, None, None, gobject.PARAM_READWRITE), - _PROP_NICK : (str, None, None, None, gobject.PARAM_READWRITE), - _PROP_COLOR : (str, None, None, None, gobject.PARAM_READWRITE), - _PROP_CURACT : (str, None, None, None, gobject.PARAM_READWRITE), + _PROP_KEY : (str, None, None, None, + gobject.PARAM_CONSTRUCT_ONLY | + gobject.PARAM_READWRITE), + _PROP_ICON : (object, None, None, gobject.PARAM_READABLE), + _PROP_NICK : (str, None, None, None, + gobject.PARAM_CONSTRUCT_ONLY | + gobject.PARAM_READWRITE), + _PROP_COLOR : (str, None, None, None, + gobject.PARAM_CONSTRUCT_ONLY | + gobject.PARAM_READWRITE), + _PROP_CURACT : (str, None, None, None, + gobject.PARAM_CONSTRUCT_ONLY | + gobject.PARAM_READWRITE), _PROP_VALID : (bool, None, None, False, gobject.PARAM_READABLE), _PROP_OWNER : (bool, None, None, False, gobject.PARAM_READABLE), _PROP_OBJID : (str, None, None, None, gobject.PARAM_READABLE), - _PROP_IP4_ADDRESS : (str, None, None, None, gobject.PARAM_READWRITE) + _PROP_IP4_ADDRESS : (str, None, None, None, + gobject.PARAM_CONSTRUCT_ONLY | + gobject.PARAM_READWRITE) } def __init__(self, bus, object_id, **kwargs): @@ -116,11 +178,13 @@ class Buddy(ExportedGObject): self._object_id = object_id self._object_path = dbus.ObjectPath(_BUDDY_PATH + object_id) - self._activities = {} # Activity ID -> Activity + #: activity ID -> activity + self._activities = {} self._activity_sigids = {} - self.handles = {} # tp client -> handle + #: Telepathy plugin -> (handle, identifier e.g. JID) + self._handles = {} - self._valid = False + self._awaiting = set(('alias', 'properties')) self._owner = False self._key = None self._icon = '' @@ -146,8 +210,9 @@ class Buddy(ExportedGObject): ExportedGObject.__init__(self, bus, self._object_path, gobject_properties=kwargs) - if icon_data: - self.props.icon = icon_data + if icon_data is not None: + self._icon = str(icon_data) + self.IconChanged(self._icon) def do_get_property(self, pspec): """Retrieve current value for the given property specifier @@ -171,7 +236,7 @@ class Buddy(ExportedGObject): return None return self._current_activity elif pspec.name == _PROP_VALID: - return self._valid + return not self._awaiting elif pspec.name == _PROP_OWNER: return self._owner elif pspec.name == _PROP_IP4_ADDRESS: @@ -184,13 +249,11 @@ class Buddy(ExportedGObject): value -- value to set emits 'icon-changed' signal on icon setting - calls _update_validity on all calls """ if pspec.name == _PROP_ICON: if str(value) != self._icon: self._icon = str(value) self.IconChanged(self._icon) - self.emit('icon-changed', self._icon) elif pspec.name == _PROP_NICK: self._nick = value elif pspec.name == _PROP_COLOR: @@ -204,8 +267,6 @@ class Buddy(ExportedGObject): elif pspec.name == _PROP_IP4_ADDRESS: self._ip4_address = value - self._update_validity() - # dbus signals @dbus.service.signal(_BUDDY_INTERFACE, signature="ay") @@ -238,11 +299,11 @@ class Buddy(ExportedGObject): full set of properties, just the changes. """ - def add_telepathy_handle(self, tp_client, handle): + def add_telepathy_handle(self, tp_client, handle, uid): """Add a Telepathy handle.""" conn = tp_client.get_connection() + self._handles[tp_client] = (handle, uid) self.TelepathyHandleAdded(conn.service_name, conn.object_path, handle) - self.handles[tp_client] = handle @dbus.service.signal(_BUDDY_INTERFACE, signature='sou') def TelepathyHandleAdded(self, tp_conn_name, tp_conn_path, handle): @@ -258,21 +319,19 @@ class Buddy(ExportedGObject): newly associated with the buddy """ - def remove_telepathy_handle(self, tp_client, handle): + def remove_telepathy_handle(self, tp_client): """Remove a Telepathy handle.""" conn = tp_client.get_connection() - my_handle = self.handles.get(tp_client, 0) - if my_handle == handle: - del self.handles[tp_client] - self.TelepathyHandleRemoved(conn.service_name, conn.object_path, - handle) - # the Owner can't disappear - that would be silly - if not self.handles and not self._owner: - self.emit('disappeared') - else: - _logger.debug('Telepathy handle %u supposedly removed, but ' - 'my handle on that connection is %u - ignoring', - handle, my_handle) + try: + handle, identifier = self._handles.pop(tp_client) + except KeyError: + return + + self.TelepathyHandleRemoved(conn.service_name, conn.object_path, + handle) + # the Owner can't disappear - that would be silly + if not self._handles and not self._owner: + self.emit('disappeared') @dbus.service.signal(_BUDDY_INTERFACE, signature='sou') def TelepathyHandleRemoved(self, tp_conn_name, tp_conn_path, handle): @@ -324,20 +383,12 @@ class Buddy(ExportedGObject): "" if no current activity """ props = {} - props[_PROP_NICK] = self.props.nick - props[_PROP_OWNER] = self.props.owner - props[_PROP_KEY] = self.props.key - props[_PROP_COLOR] = self.props.color - - if self.props.ip4_address: - props[_PROP_IP4_ADDRESS] = self.props.ip4_address - else: - props[_PROP_IP4_ADDRESS] = "" - - if self.props.current_activity: - props[_PROP_CURACT] = self.props.current_activity - else: - props[_PROP_CURACT] = "" + props[_PROP_NICK] = self.props.nick or '' + props[_PROP_OWNER] = self.props.owner or '' + props[_PROP_KEY] = self.props.key or '' + props[_PROP_COLOR] = self.props.color or '' + props[_PROP_IP4_ADDRESS] = self.props.ip4_address or '' + props[_PROP_CURACT] = self.props.current_activity or '' return props @dbus.service.method(_BUDDY_INTERFACE, @@ -351,10 +402,10 @@ class Buddy(ExportedGObject): object path, handle). """ ret = [] - for plugin in self.handles: + for plugin in self._handles: conn = plugin.get_connection() ret.append((str(conn.service_name), conn.object_path, - self.handles[plugin])) + self._handles[plugin][0])) # methods def object_path(self): @@ -415,8 +466,7 @@ class Buddy(ExportedGObject): properties -- set of property values to set if no change, no events generated - if change, generates property-changed and - calls _update_validity + if change, generates property-changed """ changed = False changed_props = {} @@ -459,7 +509,7 @@ class Buddy(ExportedGObject): # Try emitting PropertyChanged before updating validity # to avoid leaking a PropertyChanged signal before the buddy is # actually valid the first time after creation - if self._valid: + if not self._awaiting: dbus_changed = {} for key, value in changed_props.items(): if value: @@ -468,28 +518,68 @@ class Buddy(ExportedGObject): dbus_changed[key] = "" self.PropertyChanged(dbus_changed) - self.emit('property-changed', changed_props) + self._property_changed(changed_props) - self._update_validity() + def _property_changed(self, changed_props): + pass - def _update_validity(self): - """Check whether we are now valid + def update_buddy_properties(self, tp, props): + """Update the buddy properties (those that come from the GetProperties + method of the org.laptop.Telepathy.BuddyInfo interface) from the + given Telepathy connection. - validity is True if color, nick and key are non-null + Other properties, such as 'nick', may not be set via this method. + """ + self.set_properties(props) + # If the properties didn't contain the key or color, then we're never + # going to get one. + try: + self._awaiting.remove('properties') + except KeyError: + pass + else: + if not self._awaiting: + self.emit('validity-changed', True) - emits validity-changed if we have changed validity + def update_alias(self, tp, alias): + """Update the alias from the given Telepathy connection. """ + self.set_properties({'nick': alias}) try: - old_valid = self._valid - if self._color and self._nick and self._key: - self._valid = True - else: - self._valid = False + self._awaiting.remove('alias') + except KeyError: + pass + else: + if not self._awaiting: + self.emit('validity-changed', True) - if old_valid != self._valid: - self.emit("validity-changed", self._valid) - except AttributeError: - self._valid = False + def update_current_activity(self, tp, current_activity): + """Update the current activity from the given Telepathy connection. + """ + self.set_properties({'current-activity': current_activity}) + + def update_avatar(self, tp, new_avatar_token, icon=None, mime_type=None): + """Handle update of the avatar""" + conn = tp.get_connection() + handle, identifier = self._handles[tp] + + if icon is None: + icon = buddy_icon_cache.get_icon(conn.object_path, identifier, + new_avatar_token) + else: + buddy_icon_cache.store_icon(conn.object_path, identifier, + new_avatar_token, icon) + + if icon is None: + # this was AvatarUpdated not AvatarRetrieved, and then we got a + # cache miss - request an AvatarRetrieved signal so we can get the + # actual icon + conn[CONN_INTERFACE_AVATARS].RequestAvatars([handle], + ignore_reply=True) + else: + if self._icon != icon: + self._icon = icon + self.IconChanged(self._icon) class GenericOwner(Buddy): @@ -517,6 +607,9 @@ class GenericOwner(Buddy): self._key_hash = kwargs.pop("key_hash", None) self._registered = kwargs.pop("registered", False) + #: Telepathy plugin -> dict { activity ID -> room handle } + self._activities_by_connection = {} + self._ip4_addr_monitor = psutils.IP4AddressMonitor.get_instance() self._ip4_addr_monitor.connect("address-changed", self._ip4_address_changed_cb) @@ -526,7 +619,189 @@ class GenericOwner(Buddy): Buddy.__init__(self, bus, object_id, **kwargs) self._owner = True - self._bus = dbus.SessionBus() + self._bus = bus + + def add_owner_activity(self, tp, activity_id, activity_room): + # FIXME: this probably duplicates something else (_activities?) + # but for now I'll keep the same duplication as before. + # Equivalent code used to be in ServerPlugin. + id_to_act = self._activities_by_connection.setdefault(tp, {}) + id_to_act[activity_id] = activity_room + + self._set_self_activities(tp) + + def _set_self_activities(self, tp): + """Forward set of joined activities to network + + uses SetActivities on BuddyInfo channel + """ + conn = tp.get_connection() + conn[CONN_INTERFACE_BUDDY_INFO].SetActivities( + self._activities_by_connection[tp].iteritems(), + reply_handler=_noop, + error_handler=lambda e: + _logger.warning("setting activities failed: %s", e)) + + def _set_self_current_activity(self, tp): + """Forward our current activity (or "") to network + """ + cur_activity = self._current_activity + if not cur_activity: + cur_activity = "" + cur_activity_handle = 0 + else: + id_to_act = self._activities_by_connection.setdefault(tp, {}) + cur_activity_handle = id_to_act.get(cur_activity) + if cur_activity_handle is None: + # don't advertise a current activity that's not shared on + # this connection + # FIXME: this gives us a different current activity on each + # connection - need to make sure clients are OK with this + # (at the moment, PS isn't!) + cur_activity = "" + + _logger.debug("Setting current activity to '%s' (handle %s)", + cur_activity, cur_activity_handle) + conn = tp.get_connection() + conn[CONN_INTERFACE_BUDDY_INFO].SetCurrentActivity(cur_activity, + cur_activity_handle, + reply_handler=_noop, + error_handler=lambda e: + _logger.warning("setting current activity failed: %s", e)) + + def _set_self_alias(self, tp): + self_handle = self._handles[tp][0] + conn = tp.get_connection() + conn[CONN_INTERFACE_ALIASING].SetAliases({self_handle: self._nick}, + reply_handler=_noop, + error_handler=lambda e: + _logger.warning('Error setting alias: %s', e)) + # Hack so we can use this as a timeout handler + return False + + def set_properties_before_connect(self, tp): + self._set_self_olpc_properties(tp, connected=False) + + def _set_self_olpc_properties(self, tp, connected=True): + conn = tp.get_connection() + # FIXME: omit color/key/ip4-address if None? + + props = dbus.Dictionary({ + 'color': self._color or '', + 'key': dbus.ByteArray(self._key or ''), + 'ip4-address': self._ip4_address or '', + }, signature='sv') + + # FIXME: clarify whether we're meant to support random extra properties + # (Salut doesn't) + if tp._PROTOCOL == 'salut': + del props['ip4-address'] + + if connected: + conn[CONN_INTERFACE_BUDDY_INFO].SetProperties(props, + reply_handler=_noop, + error_handler=lambda e: + _logger.warning('Error setting OLPC properties: %s', e)) + else: + # we don't yet know whether the connection supports setting buddy + # properties + # FIXME: remove this hack, and the import of dbus.proxies, when + # we have a newer tp-python that makes dbus_object public + try: + obj = conn.dbus_object + if not isinstance(obj, dbus.proxies.ProxyObject): + raise AttributeError + except AttributeError: + obj = conn._dbus_object + + obj.SetProperties(props, dbus_interface=CONN_INTERFACE_BUDDY_INFO, + reply_handler=lambda: + _logger.debug('Successfully preloaded buddy props'), + error_handler=lambda e: + _logger.debug('Failed to preload buddy properties, ' + 'will try again after Connect(): %s', e)) + + # Hack so we can use this as a timeout handler + return False + + def add_telepathy_handle(self, tp_client, handle, uid): + Buddy.add_telepathy_handle(self, tp_client, handle, uid) + self._activities_by_connection.setdefault(tp_client, {}) + + self._set_self_olpc_properties(tp_client) + self._set_self_alias(tp_client) + # Hack; send twice to make sure the server gets it + #gobject.timeout_add(1000, lambda: self._set_self_alias(tp_client)) + + self._set_self_activities(tp_client) + self._set_self_current_activity(tp_client) + + self._set_self_avatar(tp_client) + + def IconChanged(self, icon_data): + # As well as emitting the D-Bus signal, prod the Telepathy + # connection manager + Buddy.IconChanged(self, icon_data) + for tp in self._handles.iterkeys(): + self._set_self_avatar(tp) + + def _set_self_avatar(self, tp): + conn = tp.get_connection() + icon_data = self._icon + + m = new_md5() + m.update(icon_data) + digest = m.hexdigest() + + self_handle = self._handles[tp][0] + token = conn[CONN_INTERFACE_AVATARS].GetAvatarTokens( + [self_handle])[0] + + if buddy_icon_cache.check_avatar(conn.object_path, digest, + token): + # avatar is up to date + return + + def set_self_avatar_cb(token): + buddy_icon_cache.set_avatar(conn.object_path, digest, token) + + types, minw, minh, maxw, maxh, maxsize = \ + conn[CONN_INTERFACE_AVATARS].GetAvatarRequirements() + if not "image/jpeg" in types: + _logger.debug("server does not accept JPEG format avatars.") + return + + width = 96 + height = 96 + size = 8192 + if maxw > 0 and width > maxw: + width = maxw + if maxw > 0 and height > maxh: + height = maxh + if maxsize > 0 and size > maxsize: + size = maxsize + + img_data = _get_buddy_icon_at_size(icon_data, width, height, size) + conn[CONN_INTERFACE_AVATARS].SetAvatar(img_data, "image/jpeg", + reply_handler=set_self_avatar_cb, + error_handler=lambda e: + _logger.warning('Error setting avatar: %s', e)) + + def _property_changed(self, changed_props): + for tp in self._handles.iterkeys(): + + if changed_props.has_key("current-activity"): + self._set_self_current_activity(tp) + + if changed_props.has_key("nick"): + self._set_self_alias(tp) + # Hack; send twice to make sure the server gets it + gobject.timeout_add(1000, lambda: self._set_self_alias(tp)) + + if (changed_props.has_key("color") or + changed_props.has_key("ip4-address")): + if tp.status == CONNECTION_STATUS_CONNECTED: + self._set_self_olpc_properties(tp) def _ip4_address_changed_cb(self, monitor, address): """Handle IPv4 address change, set property to generate event""" @@ -551,6 +826,11 @@ class GenericOwner(Buddy): """Customisation point: handle the registration of the owner""" raise RuntimeError("Subclasses must implement") + def update_avatar(self, tp, new_avatar_token, icon=None, mime_type=None): + # This should never get called because Owner avatar changes are + # driven by the Sugar shell, but just in case: + _logger.warning('GenericOwner.update_avatar() should not be called') + class ShellOwner(GenericOwner): """Representation of the local-machine owner using Sugar's Shell @@ -604,6 +884,9 @@ class ShellOwner(GenericOwner): bus_name=self._SHELL_SERVICE, path=self._SHELL_PATH) + # we already know our own nick, color, key + self._awaiting = None + def set_registered(self, value): """Handle notification that we have been registered""" if value: @@ -611,7 +894,10 @@ class ShellOwner(GenericOwner): def _icon_changed_cb(self, icon): """Handle icon change, set property to generate event""" - self.props.icon = icon + icon = str(icon) + if icon != self._icon: + self._icon = icon + self.IconChanged(icon) def _color_changed_cb(self, color): """Handle color change, set property to generate event""" diff --git a/src/buddyiconcache.py b/src/buddyiconcache.py index b7da614..b95ea98 100644 --- a/src/buddyiconcache.py +++ b/src/buddyiconcache.py @@ -81,6 +81,8 @@ class BuddyIconCache(object): self._token = token self._save_cache() +buddy_icon_cache = BuddyIconCache() + if __name__ == "__main__": my_cache = BuddyIconCache() diff --git a/src/linklocal_plugin.py b/src/linklocal_plugin.py index b8f6445..b60eb15 100644 --- a/src/linklocal_plugin.py +++ b/src/linklocal_plugin.py @@ -1,3 +1,4 @@ +"""Link-local plugin for Presence Service""" # Copyright (C) 2007, Red Hat, Inc. # Copyright (C) 2007, Collabora Ltd. # @@ -15,13 +16,177 @@ # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA +# Standard library +import logging +from itertools import izip +from string import hexdigits + +# Other libraries import gobject +from dbus import SystemBus +from telepathy.client import (ConnectionManager, Connection) +from telepathy.interfaces import (CONN_MGR_INTERFACE, CONN_INTERFACE, + CHANNEL_INTERFACE_GROUP) +from telepathy.constants import (HANDLE_TYPE_CONTACT, + CONNECTION_STATUS_CONNECTED, CONNECTION_STATUS_DISCONNECTED, + CHANNEL_GROUP_FLAG_CHANNEL_SPECIFIC_HANDLES) + +# Presence Service local modules +import psutils +from telepathy_plugin import TelepathyPlugin + + +CONN_INTERFACE_BUDDY_INFO = 'org.laptop.Telepathy.BuddyInfo' + +_logger = logging.getLogger('s-p-s.linklocal_plugin') + + +class LinkLocalPlugin(TelepathyPlugin): + """Telepathy-python-based presence server interface + + The ServerPlugin instance translates network events from + Telepathy Python into GObject events. It provides direct + python calls to perform the required network operations + to implement the PresenceService. + """ + + _TP_CONN_MANAGER = 'salut' + _PROTOCOL = 'salut' + _OBJ_PATH_PREFIX = "/org/freedesktop/Telepathy/Connection/salut/salut/" -class LinkLocalPlugin(gobject.GObject): def __init__(self, registry, owner): - gobject.GObject.__init__(self) - self._registry = registry - self._owner = owner + TelepathyPlugin.__init__(self, registry, owner) + + self._sys_bus = SystemBus() + self._have_avahi = False + self._watch = self._sys_bus.watch_name_owner('org.freedesktop.Avahi', + self._avahi_owner_cb) + + def _avahi_owner_cb(self, unique_name): + had_avahi = self._have_avahi + + if unique_name: + self._have_avahi = True + if not had_avahi: + _logger.info('Avahi appeared on the system bus (%s) - ' + 'starting...', unique_name) + self.start() + else: + self._have_avahi = False + if had_avahi: + _logger.info('Avahi disappeared from the system bus - ' + 'stopping...') + self.stop() def cleanup(self): - pass + TelepathyPlugin.cleanup(self) + if self._watch is not None: + self._watch.cancel() + self._watch = None + + def _could_connect(self): + return self._have_avahi + + def _get_account_info(self): + """Retrieve connection manager parameters for this account + """ + server = self._owner.get_server() + khash = psutils.pubkey_to_keyid(self._owner.props.key) + + return { + 'nickname': '%s' % self._owner.props.nick, + 'first-name': ' ', + 'last-name': '%s' % self._owner.props.nick, + 'jid': '%s@%s' % (khash, server), + 'published-name': '%s' % self._owner.props.nick, + } + + def _find_existing_connection(self): + """Try to find an existing Telepathy connection to this server + + filters the set of connections from + telepathy.client.Connection.get_connections + to find a connection using our protocol with the + "self handle" of that connection being a handle + which matches our account (see _get_account_info) + + returns connection or None + """ + # Search existing connections, if any, that we might be able to use + connections = Connection.get_connections() + for item in connections: + if not item.object_path.startswith(self._OBJ_PATH_PREFIX): + continue + if item[CONN_INTERFACE].GetProtocol() != self._PROTOCOL: + continue + # Any Salut instance will do + return item + return None + + def identify_contacts(self, tp_chan, handles, identifiers=None): + """Work out the "best" unique identifier we can for the given handles, + in the context of the given channel (which may be None), using only + 'fast' connection manager API (that does not involve network + round-trips). + + For the XMPP server case, we proceed as follows: + + * Find the owners of the given handles, if the channel has + channel-specific handles + * If the owner (globally-valid JID) is on a trusted server, return + 'keyid/' plus the 'key fingerprint' (the user part of their JID, + currently implemented as the SHA-1 of the Base64 blob in + owner.key.pub) + * If the owner (globally-valid JID) cannot be found or is on an + untrusted server, return 'xmpp/' plus an escaped form of the JID + + The idea is that we identify buddies by key-ID (i.e. by key, assuming + no collisions) if we can find it without making network round-trips, + but if that's not possible we just use their JIDs. + + :Parameters: + `tp_chan` : telepathy.client.Channel or None + The channel in which the handles were found, or None if they + are known to be channel-specific handles + `handles` : iterable over (int or long) + The contacts' handles in that channel + :Returns: + A dict mapping the provided handles to the best available + unique identifier, which is a string that could be used as a + suffix to an object path + """ + # we need to be able to index into handles, so force them to + # be a sequence + if not isinstance(handles, (tuple, list)): + handles = tuple(handles) + + # we happen to know that Salut has no channel-specific handles + + if identifiers is None: + identifiers = self._conn[CONN_INTERFACE].InspectHandles( + HANDLE_TYPE_CONTACT, handles) + + ret = {} + for handle, ident in izip(handles, identifiers): + # special-case the Owner - we always know who we are + if handle == self.self_handle: + ret[handle] = self._owner.props.objid + continue + + # we also happen to know that on Salut, getting properties + # is immediate, and the key is (well, will be) trustworthy + + if CONN_INTERFACE_BUDDY_INFO in self._conn: + props = self._conn[CONN_INTERFACE_BUDDY_INFO].GetProperties( + handle, byte_arrays=True, utf8_strings=True) + key = props.get('key') + else: + key = None + + if key is not None: + khash = psutils.pubkey_to_keyid(key) + ret[handle] = 'keyid/' + khash + else: + ret[handle] = 'salut/' + psutils.escape_identifier(ident) + + return ret diff --git a/src/presenceservice.py b/src/presenceservice.py index 8aeaac6..e6fb895 100644 --- a/src/presenceservice.py +++ b/src/presenceservice.py @@ -16,6 +16,8 @@ # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA import logging +from itertools import izip +from os import environ from weakref import WeakValueDictionary import dbus @@ -25,7 +27,8 @@ from dbus.exceptions import DBusException from dbus.gobject_service import ExportedGObject from dbus.mainloop.glib import DBusGMainLoop from telepathy.client import ManagerRegistry, Connection -from telepathy.interfaces import (CONN_MGR_INTERFACE, CONN_INTERFACE) +from telepathy.interfaces import (CONN_MGR_INTERFACE, CONN_INTERFACE, + CONN_INTERFACE_AVATARS, CONN_INTERFACE_ALIASING) from telepathy.constants import (CONNECTION_STATUS_CONNECTING, CONNECTION_STATUS_CONNECTED, CONNECTION_STATUS_DISCONNECTED) @@ -35,10 +38,12 @@ from sugar import util from server_plugin import ServerPlugin from linklocal_plugin import LinkLocalPlugin from buddy import Buddy, ShellOwner -from buddyiconcache import BuddyIconCache from activity import Activity from psutils import pubkey_to_keyid +CONN_INTERFACE_BUDDY_INFO = 'org.laptop.Telepathy.BuddyInfo' +CONN_INTERFACE_ACTIVITY_PROPERTIES = 'org.laptop.Telepathy.ActivityProperties' + _PRESENCE_SERVICE = "org.laptop.Sugar.Presence" _PRESENCE_INTERFACE = "org.laptop.Sugar.Presence" _PRESENCE_PATH = "/org/laptop/Sugar/Presence" @@ -46,7 +51,6 @@ _PRESENCE_PATH = "/org/laptop/Sugar/Presence" _logger = logging.getLogger('s-p-s.presenceservice') - class NotFoundError(DBusException): def __init__(self, msg): DBusException.__init__(self, msg) @@ -55,20 +59,12 @@ class NotFoundError(DBusException): class PresenceService(ExportedGObject): __gtype_name__ = "PresenceService" - __gsignals__ = { - 'connection-status': (gobject.SIGNAL_RUN_FIRST, gobject.TYPE_NONE, - ([gobject.TYPE_BOOLEAN])) - } - def _create_owner(self): # Overridden by TestPresenceService return ShellOwner(self, self._session_bus) def __init__(self): self._next_object_id = 0 - self._connected = False - - self._icon_cache = BuddyIconCache() # all Buddy objects # identifier -> Buddy, GC'd when no more refs exist @@ -83,7 +79,12 @@ class PresenceService(ExportedGObject): self._handles_buddies = {} # activity id -> Activity - self._activities = {} + self._activities_by_id = {} + #: Tp plugin -> (handle -> Activity) + self._activities_by_handle = {} + + #: Connection -> list of SignalMatch + self._conn_matches = {} self._session_bus = dbus.SessionBus() self._session_bus.add_signal_receiver(self._connection_disconnected_cb, @@ -100,30 +101,36 @@ class PresenceService(ExportedGObject): self._registry = ManagerRegistry() self._registry.LoadManagers() - # Set up the server connection - self._server_plugin = ServerPlugin(self._registry, self._owner, - self._icon_cache) - self._handles_buddies[self._server_plugin] = {} - - self._server_plugin.connect('status', self._server_status_cb) - self._server_plugin.connect('contact-online', self._contact_online) - self._server_plugin.connect('contact-offline', self._contact_offline) - self._server_plugin.connect('avatar-updated', self._avatar_updated) - self._server_plugin.connect('buddy-properties-changed', - self._buddy_properties_changed) - self._server_plugin.connect('buddy-activities-changed', - self._buddy_activities_changed) - self._server_plugin.connect('activity-invitation', - self._activity_invitation) - self._server_plugin.connect('private-invitation', - self._private_invitation) - self._server_plugin.connect('activity-properties-changed', - self._activity_properties_changed) - self._server_plugin.start() - - # Set up the link local connection - self._ll_plugin = LinkLocalPlugin(self._registry, self._owner) - self._handles_buddies[self._ll_plugin] = {} + # Set up the Telepathy plugins + self._plugins = [] + debug_flags = set(environ.get('PRESENCE_SERVICE_DEBUG', '').split(',')) + _logger.debug('Debug flags: %r', debug_flags) + if 'disable-gabble' in debug_flags: + self._server_plugin = None + else: + self._server_plugin = ServerPlugin(self._registry, self._owner) + self._plugins.append(self._server_plugin) + if 'disable-salut' in debug_flags: + self._ll_plugin = None + else: + self._ll_plugin = LinkLocalPlugin(self._registry, self._owner) + self._plugins.append(self._ll_plugin) + self._connected_plugins = set() + + for tp in self._plugins: + self._handles_buddies[tp] = {} + self._activities_by_handle[tp] = {} + + tp.connect('status', self._tp_status_cb) + tp.connect('contacts-online', self._contacts_online) + tp.connect('contacts-offline', self._contacts_offline) + tp.connect('activity-invitation', + self._activity_invitation) + tp.connect('private-invitation', + self._private_invitation) + tp.start() + + self._contacts_online_queue = [] ExportedGObject.__init__(self, self._session_bus, _PRESENCE_PATH) @@ -141,23 +148,111 @@ class PresenceService(ExportedGObject): """Log event when D-Bus kicks us off the bus for some reason""" _logger.debug("Disconnected from session bus!!!") - def _server_status_cb(self, plugin, status, reason): - - # FIXME: figure out connection status when we have a salut plugin too - old_status = self._connected + def _tp_status_cb(self, plugin, status, reason): if status == CONNECTION_STATUS_CONNECTED: - self._connected = True - self._handles_buddies[plugin][plugin.self_handle] = self._owner - self._owner.add_telepathy_handle(plugin, plugin.self_handle) + self._tp_connected(plugin) else: - self._connected = False - if plugin.self_handle is not None: - self._handles_buddies.setdefault(plugin, {}).pop( - plugin.self_handle, None) - self._owner.remove_telepathy_handle(plugin, plugin.self_handle) + self._tp_disconnected(plugin) + + def _tp_connected(self, tp): + self._connected_plugins.add(tp) + self._handles_buddies[tp][tp.self_handle] = self._owner + self._owner.add_telepathy_handle(tp, tp.self_handle, + tp.self_identifier) + + conn = tp.get_connection() - if self._connected != old_status: - self.emit('connection-status', self._connected) + self._conn_matches[conn] = [] + + if CONN_INTERFACE_ACTIVITY_PROPERTIES in conn: + def activity_properties_changed(room, properties): + self._activity_properties_changed(tp, room, properties) + m = conn[CONN_INTERFACE_ACTIVITY_PROPERTIES].connect_to_signal( + 'ActivityPropertiesChanged', + activity_properties_changed) + self._conn_matches[conn].append(m) + else: + _logger.warning('Connection %s does not support OLPC activity ' + 'properties', conn.object_path) + + if CONN_INTERFACE_BUDDY_INFO in conn: + def buddy_activities_changed(contact, activities): + self._buddy_activities_changed(tp, contact, activities) + m = conn[CONN_INTERFACE_BUDDY_INFO].connect_to_signal( + 'ActivitiesChanged', buddy_activities_changed) + self._conn_matches[conn].append(m) + + def buddy_properties_changed(contact, properties): + buddy = self._handles_buddies[tp].get(contact) + if buddy is not None and buddy is not self._owner: + buddy.update_buddy_properties(tp, properties) + m = conn[CONN_INTERFACE_BUDDY_INFO].connect_to_signal( + 'PropertiesChanged', buddy_properties_changed) + self._conn_matches[conn].append(m) + + def buddy_curact_changed(contact, act_id, room): + if (act_id == '' or not util.validate_activity_id(act_id) or + room == 0): + act_id = '' + room = 0 + buddy = self._handles_buddies[tp].get(contact) + if buddy is not None and buddy is not self._owner: + buddy.update_current_activity(tp, act_id) + # FIXME: do something useful with the room handle? + m = conn[CONN_INTERFACE_BUDDY_INFO].connect_to_signal( + 'CurrentActivityChanged', buddy_curact_changed) + self._conn_matches[conn].append(m) + else: + _logger.warning('Connection %s does not support OLPC buddy info', + conn.object_path) + + if CONN_INTERFACE_AVATARS in conn: + def avatar_retrieved(contact, avatar_token, avatar, mime_type): + self._avatar_updated(tp, contact, avatar_token, avatar, + mime_type) + m = conn[CONN_INTERFACE_AVATARS].connect_to_signal( + 'AvatarRetrieved', avatar_retrieved) + self._conn_matches[conn].append(m) + + def avatar_updated(contact, avatar_token): + self._avatar_updated(tp, contact, avatar_token) + m = conn[CONN_INTERFACE_AVATARS].connect_to_signal('AvatarUpdated', + avatar_updated) + self._conn_matches[conn].append(m) + else: + _logger.warning('Connection %s does not support avatars', + conn.object_path) + + if CONN_INTERFACE_ALIASING in conn: + def aliases_changed(aliases): + for contact, alias in aliases: + buddy = self._handles_buddies[tp].get(contact) + if buddy is not None and buddy is not self._owner: + buddy.update_alias(tp, alias) + m = conn[CONN_INTERFACE_ALIASING].connect_to_signal( + 'AliasesChanged', aliases_changed) + self._conn_matches[conn].append(m) + else: + _logger.warning('Connection %s does not support aliasing', + conn.object_path) + + def _tp_disconnected(self, tp): + self._connected_plugins.discard(tp) + if tp.self_handle is not None: + self._handles_buddies.setdefault(tp, {}).pop( + tp.self_handle, None) + self._owner.remove_telepathy_handle(tp) + + conn = tp.get_connection() + + matches = self._conn_matches.get(conn) + try: + del self._conn_matches[conn] + except KeyError: + pass + if matches is not None: + for match in matches: + match.remove() def get_buddy(self, objid): buddy = self._buddies.get(objid) @@ -170,19 +265,134 @@ class PresenceService(ExportedGObject): self._buddies[objid] = buddy return buddy - def _contact_online(self, tp, objid, handle, props): - _logger.debug('Handle %u, .../%s is now online', handle, objid) - buddy = self.get_buddy(objid) + def _contacts_online(self, tp, objids, handles, identifiers): + # we'll iterate over handles many times, so make sure that will + # work + if not isinstance(handles, (list, tuple)): + handles = tuple(handles) - self._handles_buddies[tp][handle] = buddy - # store the handle of the buddy for this CM - buddy.add_telepathy_handle(tp, handle) - buddy.set_properties(props) + for objid, handle, identifier in izip(objids, handles, identifiers): + _logger.debug('Handle %u, .../%s is now online', handle, objid) + buddy = self.get_buddy(objid) + + self._handles_buddies[tp][handle] = buddy + # Store the handle of the buddy for this CM. This doesn't + # fetch anything over D-Bus, to avoid reaching the pending-call + # limit. + buddy.add_telepathy_handle(tp, handle, identifier) + + conn = tp.get_connection() + + if not self._contacts_online_queue: + gobject.idle_add(self._run_contacts_online_queue) + + def handle_error(e, when): + gobject.idle_add(self._run_contacts_online_queue) + _logger.warning('Error %s: %s', when, e) + + if CONN_INTERFACE_ALIASING in conn: + def got_aliases(aliases): + gobject.idle_add(self._run_contacts_online_queue) + for contact, alias in izip(handles, aliases): + buddy = self._handles_buddies[tp].get(contact) + if buddy is not None and buddy is not self._owner: + buddy.update_alias(tp, alias) + def request_aliases(): + try: + conn[CONN_INTERFACE_ALIASING].RequestAliases(handles, + reply_handler=got_aliases, + error_handler=lambda e: + handle_error(e, 'fetching aliases')) + except Exception, e: + gobject.idle_add(self._run_contacts_online_queue) + handle_error(e, 'fetching aliases') + self._contacts_online_queue.append(request_aliases) + + for handle in handles: + self._queue_contact_online(tp, handle) + + if CONN_INTERFACE_AVATARS in conn: + def got_avatar_tokens(tokens): + gobject.idle_add(self._run_contacts_online_queue) + for contact, token in izip(handles, tokens): + self._avatar_updated(tp, contact, token) + def get_avatar_tokens(): + try: + conn[CONN_INTERFACE_AVATARS].GetAvatarTokens(handles, + reply_handler=got_avatar_tokens, + error_handler=lambda e: + handle_error(e, 'fetching avatar tokens')) + except Exception, e: + gobject.idle_add(self._run_contacts_online_queue) + handle_error(e, 'fetching avatar tokens') + self._contacts_online_queue.append(get_avatar_tokens) + + def _queue_contact_online(self, tp, contact): + conn = tp.get_connection() + + if CONN_INTERFACE_BUDDY_INFO in conn: + def handle_error(e, when): + gobject.idle_add(self._run_contacts_online_queue) + _logger.warning('Error %s: %s', when, e) + def got_properties(props): + gobject.idle_add(self._run_contacts_online_queue) + buddy = self._handles_buddies[tp].get(contact) + if buddy is not None and buddy is not self._owner: + buddy.update_buddy_properties(tp, props) + def get_properties(): + try: + conn[CONN_INTERFACE_BUDDY_INFO].GetProperties(contact, + byte_arrays=True, reply_handler=got_properties, + error_handler=lambda e: + handle_error(e, 'fetching buddy properties')) + except Exception, e: + gobject.idle_add(self._run_contacts_online_queue) + handle_error(e, 'fetching buddy properties') + def got_current_activity(current_activity, room): + gobject.idle_add(self._run_contacts_online_queue) + buddy = self._handles_buddies[tp].get(contact) + if buddy is not None and buddy is not self._owner: + buddy.update_current_activity(tp, current_activity) + def get_current_activity(): + try: + conn[CONN_INTERFACE_BUDDY_INFO].GetCurrentActivity(contact, + reply_handler=got_current_activity, + error_handler=lambda e: + handle_error(e, 'fetching current activity')) + except Exception, e: + gobject.idle_add(self._run_contacts_online_queue) + handle_error(e, 'fetching current activity') + def got_activities(activities): + gobject.idle_add(self._run_contacts_online_queue) + self._buddy_activities_changed(tp, contact, activities) + def get_activities(): + try: + conn[CONN_INTERFACE_BUDDY_INFO].GetActivities(contact, + reply_handler=got_activities, + error_handler=lambda e: + handle_error(e, 'fetching activities')) + except Exception, e: + gobject.idle_add(self._run_contacts_online_queue) + handle_error(e, 'fetching activities') + + self._contacts_online_queue.append(get_properties) + self._contacts_online_queue.append(get_current_activity) + self._contacts_online_queue.append(get_activities) + + def _run_contacts_online_queue(self): + try: + callback = self._contacts_online_queue.pop(0) + except IndexError: + pass + else: + callback() + return False def _buddy_validity_changed_cb(self, buddy, valid): if valid: self.BuddyAppeared(buddy.object_path()) - self._buddies_by_pubkey[buddy.props.key] = buddy + if buddy.props.key is not None: + self._buddies_by_pubkey[buddy.props.key] = buddy _logger.debug("New Buddy: %s (%s)", buddy.props.nick, buddy.props.color) else: @@ -195,33 +405,26 @@ class PresenceService(ExportedGObject): if buddy.props.valid: self._buddy_validity_changed_cb(buddy, False) - def _contact_offline(self, tp, handle): - if not self._handles_buddies[tp].has_key(handle): - return - - buddy = self._handles_buddies[tp].pop(handle) - # the handle of the buddy for this CM is not valid anymore - # (this might trigger _buddy_disappeared_cb if they are not visible - # via any CM) - buddy.remove_telepathy_handle(tp, handle) + def _contacts_offline(self, tp, handles): + for handle in handles: + buddy = self._handles_buddies[tp].pop(handle, None) + # the handle of the buddy for this CM is not valid anymore + # (this might trigger _buddy_disappeared_cb if they are not + # visible via any CM) + if buddy is not None: + buddy.remove_telepathy_handle(tp) def _get_next_object_id(self): """Increment and return the object ID counter.""" self._next_object_id = self._next_object_id + 1 return self._next_object_id - def _avatar_updated(self, tp, handle, avatar): + def _avatar_updated(self, tp, handle, new_avatar_token, avatar=None, + mime_type=None): buddy = self._handles_buddies[tp].get(handle) - if buddy and not buddy.props.owner: + if buddy is not None and buddy is not self._owner: _logger.debug("Buddy %s icon updated" % buddy.props.nick) - buddy.props.icon = avatar - - def _buddy_properties_changed(self, tp, handle, properties): - buddy = self._handles_buddies[tp].get(handle) - if buddy: - buddy.set_properties(properties) - _logger.debug("Buddy %s properties updated: %s", buddy.props.nick, - properties.keys()) + buddy.update_avatar(tp, new_avatar_token, avatar, mime_type) def _new_activity(self, activity_id, tp, room): try: @@ -231,25 +434,45 @@ class PresenceService(ExportedGObject): except Exception: # FIXME: catching bare Exception considered harmful _logger.debug("Invalid activity:", exc_info=1) + try: + del self._activities_by_handle[tp][room] + except KeyError: + pass return None activity.connect("validity-changed", self._activity_validity_changed_cb) activity.connect("disappeared", self._activity_disappeared_cb) - self._activities[activity_id] = activity + self._activities_by_id[activity_id] = activity + self._activities_by_handle[tp][room] = activity return activity def _activity_disappeared_cb(self, activity): _logger.debug("activity %s disappeared" % activity.props.id) self.ActivityDisappeared(activity.object_path()) - del self._activities[activity.props.id] + try: + del self._activities_by_id[activity.props.id] + except KeyError: + pass + tp, room = activity.room_details + try: + del self._activities_by_handle[tp][room] + except KeyError: + pass def _buddy_activities_changed(self, tp, contact_handle, activities): - _logger.debug("Handle %s activities changed: %s", contact_handle, activities) + activities = dict(activities) + _logger.debug("Handle %s activities changed: %s", contact_handle, + activities) buddies = self._handles_buddies[tp] buddy = buddies.get(contact_handle) + if buddy is self._owner: + # ignore network events for Owner activity changes since those + # are handled locally + return + if not buddy: # We don't know this buddy # FIXME: What should we do here? @@ -264,28 +487,40 @@ class PresenceService(ExportedGObject): new_activities = set(activities.iterkeys()) activities_joined = new_activities - old_activities + for act in activities_joined: - _logger.debug("Handle %s joined activity %s", contact_handle, act) - activity = self._activities.get(act) + room_handle = activities[act] + _logger.debug("Handle %s claims to have joined activity %s", + contact_handle, act) + activity = self._activities_by_id.get(act) if activity is None: # new activity, can fail - activity = self._new_activity(act, tp, activities[act]) + _logger.debug('No activity object for %s, creating one', act) + activity = self._new_activity(act, tp, room_handle) - if activity is not None: + if activity is None: + _logger.debug('Failed to create activity object for %s', act) + else: activity.buddy_apparently_joined(buddy) activities_left = old_activities - new_activities for act in activities_left: - _logger.debug("Handle %s left activity %s", contact_handle, act) - activity = self._activities.get(act) - if not activity: + _logger.debug("Handle %s claims to have left activity %s", + contact_handle, act) + activity = self._activities_by_id.get(act) + if activity is None: + # don't bother creating an Activity just so someone can leave continue activity.buddy_apparently_left(buddy) - def _activity_invitation(self, tp, act_id, act_handle): - activity = self._activities.get(act_id) - if activity: + def _activity_invitation(self, tp, act_handle): + activity = self._activities_by_handle[tp].get(act_handle) + if activity is None: + # FIXME: we should synthesize an activity somehow, for the case of + # an invite to a non-public room + pass + else: self.ActivityInvitation(activity.object_path()) def _private_invitation(self, tp, chan_path): @@ -321,7 +556,7 @@ class PresenceService(ExportedGObject): out_signature="ao") def GetActivities(self): ret = [] - for act in self._activities.values(): + for act in self._activities_by_id.values(): if act.props.valid: ret.append(act.object_path()) return ret @@ -329,7 +564,7 @@ class PresenceService(ExportedGObject): @dbus.service.method(_PRESENCE_INTERFACE, in_signature="s", out_signature="o") def GetActivityById(self, actid): - act = self._activities.get(actid, None) + act = self._activities_by_id.get(actid, None) if not act or not act.props.valid: raise NotFoundError("The activity was not found.") return act.object_path() @@ -459,10 +694,19 @@ class PresenceService(ExportedGObject): self._share_activity(actid, atype, name, properties, async_cb, async_err_cb) + def _get_preferred_plugin(self): + for tp in self._plugins: + if tp in self._connected_plugins: + return tp + return None + @dbus.service.method(_PRESENCE_INTERFACE, in_signature='', out_signature="so") def GetPreferredConnection(self): - conn = self._server_plugin.get_connection() + tp = self._get_preferred_plugin + if tp is None: + raise NotFoundError('No connection is available') + conn = tp.get_connection() return str(conn.service_name), conn.object_path def cleanup(self): @@ -472,16 +716,23 @@ class PresenceService(ExportedGObject): def _share_activity(self, actid, atype, name, properties, async_cb, async_err_cb): objid = self._get_next_object_id() - # FIXME check which tp client we should use to share the activity + # FIXME: is the preferred Telepathy plugin always the right way to + # share the activity? color = self._owner.props.color activity = Activity(self._session_bus, objid, self, - self._server_plugin, 0, + self._get_preferred_plugin(), 0, id=actid, type=atype, name=name, color=color, local=True) activity.connect("validity-changed", self._activity_validity_changed_cb) - self._activities[actid] = activity - activity.join(async_cb, async_err_cb, True) + self._activities_by_id[actid] = activity + + def activity_shared(): + tp, room = activity.room_details + self._activities_by_handle[tp][room] = activity + async_cb(activity.object_path()) + + activity.join(activity_shared, async_err_cb, True) # local activities are valid at creation by definition, but we can't # connect to the activity's validity-changed signal until its already @@ -499,9 +750,12 @@ class PresenceService(ExportedGObject): _logger.debug("Activity disappeared: %s (%s)", activity.props.name, activity.props.id) - def _activity_properties_changed(self, tp, act_id, act_handle, props): - activity = self._activities.get(act_id) - if activity: + def _activity_properties_changed(self, tp, act_handle, props): + activity = self._activities_by_handle[tp].get(act_handle) + if activity is None: + # FIXME: synthesize an activity + pass + else: activity.set_properties(props) diff --git a/src/pstest.py b/src/pstest.py index 7715fd3..094e15b 100644 --- a/src/pstest.py +++ b/src/pstest.py @@ -68,6 +68,9 @@ class TestOwner(GenericOwner): key=pubkey, nick=nick, color=color, icon=icon, registered=registered, key_hash=privkey_hash) + # we already know our own nick, color, key + self._awaiting = None + # Only do the random stuff if randomize is true if randomize: self._ps.connect('connection-status', self._ps_connection_status_cb) @@ -176,7 +179,7 @@ class TestPresenceService(PresenceService): self.__test_num, self.__randomize) def internal_get_activity(self, actid): - return self._activities.get(actid, None) + return self._activities_by_id.get(actid, None) def _extract_public_key(keyfile): diff --git a/src/server_plugin.py b/src/server_plugin.py index 053170b..526a639 100644 --- a/src/server_plugin.py +++ b/src/server_plugin.py @@ -1,4 +1,4 @@ -"""Telepathy-python presence server interface/implementation plugin""" +"""XMPP server plugin for Presence Service""" # Copyright (C) 2007, Red Hat, Inc. # Copyright (C) 2007, Collabora Ltd. # @@ -18,82 +18,28 @@ # Standard library import logging -import os -import sys +from itertools import izip from string import hexdigits -try: - # Python >= 2.5 - from hashlib import md5 -except ImportError: - from md5 import new as md5 # Other libraries -import dbus import gobject -import gtk -from telepathy.client import (ConnectionManager, ManagerRegistry, Connection, - Channel) +from telepathy.client import (ConnectionManager, Connection) from telepathy.interfaces import (CONN_MGR_INTERFACE, CONN_INTERFACE, - CHANNEL_TYPE_CONTACT_LIST, CHANNEL_INTERFACE_GROUP, - CONN_INTERFACE_ALIASING, CONN_INTERFACE_AVATARS, CONN_INTERFACE_PRESENCE, - CHANNEL_TYPE_TEXT, CHANNEL_TYPE_STREAMED_MEDIA, PROPERTIES_INTERFACE) + CHANNEL_INTERFACE_GROUP) from telepathy.constants import (HANDLE_TYPE_CONTACT, - HANDLE_TYPE_LIST, HANDLE_TYPE_CONTACT, HANDLE_TYPE_ROOM, CONNECTION_STATUS_CONNECTED, CONNECTION_STATUS_DISCONNECTED, - CONNECTION_STATUS_CONNECTING, - CONNECTION_STATUS_REASON_AUTHENTICATION_FAILED, - CONNECTION_STATUS_REASON_NONE_SPECIFIED, - CHANNEL_GROUP_FLAG_CHANNEL_SPECIFIC_HANDLES, - PROPERTY_FLAG_WRITE) -from sugar import util + CHANNEL_GROUP_FLAG_CHANNEL_SPECIFIC_HANDLES) # Presence Service local modules import psutils +from telepathy_plugin import TelepathyPlugin -CONN_INTERFACE_BUDDY_INFO = 'org.laptop.Telepathy.BuddyInfo' -CONN_INTERFACE_ACTIVITY_PROPERTIES = 'org.laptop.Telepathy.ActivityProperties' - -_PROTOCOL = "jabber" -_OBJ_PATH_PREFIX = "/org/freedesktop/Telepathy/Connection/gabble/jabber/" _logger = logging.getLogger('s-p-s.server_plugin') -_RECONNECT_TIMEOUT = 5000 - -def _buddy_icon_save_cb(buf, data): - data[0] += buf - return True - -def _get_buddy_icon_at_size(icon, maxw, maxh, maxsize): - loader = gtk.gdk.PixbufLoader() - loader.write(icon) - loader.close() - unscaled_pixbuf = loader.get_pixbuf() - del loader - - pixbuf = unscaled_pixbuf.scale_simple(maxw, maxh, gtk.gdk.INTERP_BILINEAR) - del unscaled_pixbuf - - data = [""] - quality = 90 - img_size = maxsize + 1 - while img_size > maxsize: - data = [""] - pixbuf.save_to_callback(_buddy_icon_save_cb, "jpeg", - {"quality":"%d" % quality}, data) - quality -= 10 - img_size = len(data[0]) - del pixbuf - - if img_size > maxsize: - data = [""] - raise RuntimeError("could not size image less than %d bytes" % maxsize) - return str(data[0]) - - -class ServerPlugin(gobject.GObject): +class ServerPlugin(TelepathyPlugin): """Telepathy-python-based presence server interface The ServerPlugin instance translates network events from @@ -101,105 +47,21 @@ class ServerPlugin(gobject.GObject): python calls to perform the required network operations to implement the PresenceService. """ - __gsignals__ = { - 'contact-online': - # Contact has come online and we've discovered all their buddy - # properties. - # args: - # contact identification (based on key ID or JID): str - # contact handle: int or long - # dict {name: str => property: object} - (gobject.SIGNAL_RUN_FIRST, None, [str, object, object]), - 'contact-offline': - # Contact has gone offline. - # args: contact handle - (gobject.SIGNAL_RUN_FIRST, None, [object]), - 'status': - # Connection status changed. - # args: status, reason as for Telepathy StatusChanged - (gobject.SIGNAL_RUN_FIRST, None, [int, int]), - 'avatar-updated': - # Contact's avatar has changed - # args: - # contact handle: int - # icon data: str - (gobject.SIGNAL_RUN_FIRST, None, [object, object]), - 'buddy-properties-changed': - # OLPC buddy properties changed; as for PropertiesChanged - # args: - # contact handle: int - # properties: dict {name: str => property: object} - # FIXME: are these all the properties or just those that changed? - (gobject.SIGNAL_RUN_FIRST, None, [object, object]), - 'buddy-activities-changed': - # OLPC activities changed - # args: - # contact handle: int - # activities: dict {activity_id: str => room: int or long} - (gobject.SIGNAL_RUN_FIRST, None, [object, object]), - 'activity-invitation': - # We were invited to join an activity - # args: - # activity ID: str - # activity room handle: int or long - (gobject.SIGNAL_RUN_FIRST, None, [object, object]), - 'private-invitation': - # We were invited to join a chat or a media call - # args: - # channel object path - (gobject.SIGNAL_RUN_FIRST, None, [object]), - 'activity-properties-changed': - # An activity's properties changed; as for - # ActivityPropertiesChanged - # args: - # activity ID: str - # activity room handle: int or long - # properties: dict { str => object } - # FIXME: are these all the properties or just those that changed? - (gobject.SIGNAL_RUN_FIRST, None, [object, object, object]), - } - - def __init__(self, registry, owner, icon_cache): - """Initialize the ServerPlugin instance - - registry -- telepathy.client.ManagerRegistry from the - PresenceService, used to find the "gabble" connection - manager in this case... - owner -- presence.buddy.GenericOwner instance (normally a - presence.buddy.ShellOwner instance) - """ - gobject.GObject.__init__(self) - - self._conn = None - self._icon_cache = icon_cache - self._registry = registry - self._online_contacts = {} # handle -> jid + _TP_CONN_MANAGER = 'gabble' + _PROTOCOL = 'jabber' + _OBJ_PATH_PREFIX = "/org/freedesktop/Telepathy/Connection/gabble/jabber/" - # activity id -> handle - self._activities = {} - # (activity_id, handle of the activity channel) - self._joined_activities = [] - - self._owner = owner - self._owner.connect("property-changed", - self._owner_property_changed_cb) - self._owner.connect("icon-changed", self._owner_icon_changed_cb) - self.self_handle = None - - self._account = self._get_account_info() - self._conn_status = CONNECTION_STATUS_DISCONNECTED - self._reconnect_id = 0 + def __init__(self, registry, owner): + TelepathyPlugin.__init__(self, registry, owner) # Monitor IPv4 address as an indicator of the network connection self._ip4am = psutils.IP4AddressMonitor.get_instance() - self._ip4am.connect('address-changed', self._ip4_address_changed_cb) + self._ip4am_sigid = self._ip4am.connect('address-changed', self._ip4_address_changed_cb) - self._publish_channel = None - self._subscribe_channel = None - self._subscribe_members = set() - self._subscribe_local_pending = set() - self._subscribe_remote_pending = set() + def cleanup(self): + TelepathyPlugin.cleanup(self) + self._ip4am.disconnect(self._ip4am_sigid) def _ip4_address_changed_cb(self, ip4am, address): _logger.debug("::: IP4 address now %s", address) @@ -211,65 +73,20 @@ class ServerPlugin(gobject.GObject): self.start() else: _logger.debug("::: invalid IP4 address, will disconnect") - self.cleanup() - - def _owner_property_changed_cb(self, owner, properties): - """Local user's configuration properties have changed - - owner -- the Buddy object for the local user - properties -- set of updated properties - - calls: - - _set_self_current_activity current-activity - _set_self_alias nick - _set_self_olpc_properties color - - depending on which properties are present in the - set of properties. - """ - _logger.debug("Owner properties changed: %s", properties) - - if properties.has_key("current-activity"): - self._set_self_current_activity() - - if properties.has_key("nick"): - self._set_self_alias() - # Hack; send twice to make sure the server gets it - gobject.timeout_add(1000, self._set_self_alias) - - if properties.has_key("color") or properties.has_key("ip4-address"): - if self._conn_status == CONNECTION_STATUS_CONNECTED: - self._set_self_olpc_properties() - - def _owner_icon_changed_cb(self, owner, icon): - """Owner has changed their icon, forward to network""" - _logger.debug("Owner icon changed to size %d", len(str(icon))) - self._set_self_avatar(icon) + self.stop() def _get_account_info(self): - """Retrieve metadata dictionary describing this account - - returns dictionary with: - - server : server url from owner - account : printable-ssh-key-hash@server - password : ssh-key-hash - register : whether to register (i.e. whether not yet - registered) + """Retrieve connection manager parameters for this account """ - account_info = {} - - account_info['server'] = self._owner.get_server() - + server = self._owner.get_server() khash = psutils.pubkey_to_keyid(self._owner.props.key) - account_info['account'] = "%s@%s" % (khash, account_info['server']) - - account_info['password'] = self._owner.get_key_hash() - account_info['register'] = not self._owner.get_registered() - print "ACCT: %s" % account_info - return account_info + return { + 'account': "%s@%s" % (khash, server), + 'fallback-conference-server': "conference.%s" % server, + 'password': self._owner.get_key_hash(), + 'register': not self._owner.get_registered(), + } def _find_existing_connection(self): """Try to find an existing Telepathy connection to this server @@ -287,9 +104,9 @@ class ServerPlugin(gobject.GObject): # Search existing connections, if any, that we might be able to use connections = Connection.get_connections() for item in connections: - if not item.object_path.startswith(_OBJ_PATH_PREFIX): + if not item.object_path.startswith(self._OBJ_PATH_PREFIX): continue - if item[CONN_INTERFACE].GetProtocol() != _PROTOCOL: + if item[CONN_INTERFACE].GetProtocol() != self._PROTOCOL: continue if item[CONN_INTERFACE].GetStatus() == CONNECTION_STATUS_CONNECTED: test_handle = item[CONN_INTERFACE].RequestHandles( @@ -299,686 +116,15 @@ class ServerPlugin(gobject.GObject): return item return None - def get_connection(self): - """Retrieve our telepathy.client.Connection object""" - return self._conn - - def _init_connection(self): - """Set up our connection - - if there is no existing connection - (_find_existing_connection returns None) - produce a new connection with our protocol for our - account. - - if there is an existing connection, reuse it by - registering for various of events on it. - """ - conn = self._find_existing_connection() - if not conn: - acct = self._account.copy() - - # Create a new connection - gabble_mgr = self._registry.GetManager('gabble') - name, path = gabble_mgr[CONN_MGR_INTERFACE].RequestConnection( - _PROTOCOL, acct) - conn = Connection(name, path) - del acct - - conn[CONN_INTERFACE].connect_to_signal('StatusChanged', - self._status_changed_cb) - conn[CONN_INTERFACE].connect_to_signal('NewChannel', - self._new_channel_cb) - - # hack - conn._valid_interfaces.add(CONN_INTERFACE_PRESENCE) - conn._valid_interfaces.add(CONN_INTERFACE_BUDDY_INFO) - conn._valid_interfaces.add(CONN_INTERFACE_ACTIVITY_PROPERTIES) - conn._valid_interfaces.add(CONN_INTERFACE_AVATARS) - conn._valid_interfaces.add(CONN_INTERFACE_ALIASING) - - conn[CONN_INTERFACE_PRESENCE].connect_to_signal('PresenceUpdate', - self._presence_update_cb) - - self._conn = conn - status = self._conn[CONN_INTERFACE].GetStatus() - - if status == CONNECTION_STATUS_DISCONNECTED: - def connect_reply(): - _logger.debug('Connect() succeeded') - def connect_error(e): - _logger.debug('Connect() failed: %s', e) - if not self._reconnect_id: - self._reconnect_id = gobject.timeout_add(_RECONNECT_TIMEOUT, - self._reconnect_cb) - - self._conn[CONN_INTERFACE].Connect(reply_handler=connect_reply, - error_handler=connect_error) - - self._handle_connection_status_change(status, - CONNECTION_STATUS_REASON_NONE_SPECIFIED) + def _could_connect(self): + return bool(self._ip4am.props.address) def _connected_cb(self): - """Callback on successful connection to a server - """ - if self._account['register']: # we successfully register this account self._owner.set_registered(True) - # request both handles at the same time to reduce round-trips - pub_handle, sub_handle = self._conn[CONN_INTERFACE].RequestHandles( - HANDLE_TYPE_LIST, ['publish', 'subscribe']) - - # the group of contacts who may receive your presence - publish = self._conn.request_channel(CHANNEL_TYPE_CONTACT_LIST, - HANDLE_TYPE_LIST, pub_handle, True) - self._publish_channel = publish - publish[CHANNEL_INTERFACE_GROUP].connect_to_signal('MembersChanged', - self._publish_members_changed_cb) - publish_handles, local_pending, remote_pending = \ - publish[CHANNEL_INTERFACE_GROUP].GetAllMembers() - - # the group of contacts for whom you wish to receive presence - subscribe = self._conn.request_channel(CHANNEL_TYPE_CONTACT_LIST, - HANDLE_TYPE_LIST, sub_handle, True) - self._subscribe_channel = subscribe - subscribe[CHANNEL_INTERFACE_GROUP].connect_to_signal('MembersChanged', - self._subscribe_members_changed_cb) - subscribe_handles, subscribe_lp, subscribe_rp = \ - subscribe[CHANNEL_INTERFACE_GROUP].GetAllMembers() - self._subscribe_members = set(subscribe_handles) - self._subscribe_local_pending = set(subscribe_lp) - self._subscribe_remote_pending = set(subscribe_rp) - - if local_pending: - # accept pending subscriptions - publish[CHANNEL_INTERFACE_GROUP].AddMembers(local_pending, '') - - self.self_handle = self._conn[CONN_INTERFACE].GetSelfHandle() - self._online_contacts[self.self_handle] = self._account['account'] - - # request subscriptions from people subscribed to us if we're not - # subscribed to them - not_subscribed = list(set(publish_handles) - set(subscribe_handles)) - subscribe[CHANNEL_INTERFACE_GROUP].AddMembers(not_subscribed, '') - - if CONN_INTERFACE_BUDDY_INFO not in self._conn.get_valid_interfaces(): - _logger.debug('OLPC information not available') - return False - - self._conn[CONN_INTERFACE_BUDDY_INFO].connect_to_signal( - 'PropertiesChanged', self._buddy_properties_changed_cb) - self._conn[CONN_INTERFACE_BUDDY_INFO].connect_to_signal( - 'ActivitiesChanged', self._buddy_activities_changed_cb) - self._conn[CONN_INTERFACE_BUDDY_INFO].connect_to_signal( - 'CurrentActivityChanged', - self._buddy_current_activity_changed_cb) - - self._conn[CONN_INTERFACE_AVATARS].connect_to_signal('AvatarUpdated', - self._avatar_updated_cb) - self._conn[CONN_INTERFACE_ALIASING].connect_to_signal('AliasesChanged', - self._alias_changed_cb) - self._conn[CONN_INTERFACE_ACTIVITY_PROPERTIES].connect_to_signal( - 'ActivityPropertiesChanged', - self._activity_properties_changed_cb) - - # Set initial buddy properties, avatar, and activities - self._set_self_olpc_properties() - self._set_self_alias() - # Hack; send twice to make sure the server gets it - gobject.timeout_add(1000, self._set_self_alias) - self._set_self_activities() - self._set_self_current_activity() - self._set_self_avatar() - - # Request presence for everyone we're subscribed to - self._conn[CONN_INTERFACE_PRESENCE].RequestPresence(subscribe_handles) - return True - - def _set_self_avatar(self, icon_data=None): - if not icon_data: - icon_data = self._owner.props.icon - - m = md5() - m.update(icon_data) - digest = m.hexdigest() - - self_handle = self._conn[CONN_INTERFACE].GetSelfHandle() - token = self._conn[CONN_INTERFACE_AVATARS].GetAvatarTokens( - [self_handle])[0] - - if self._icon_cache.check_avatar(self._conn.object_path, digest, - token): - # avatar is up to date - return - - def set_self_avatar_cb(token): - self._icon_cache.set_avatar(self._conn.object_path, digest, token) - - types, minw, minh, maxw, maxh, maxsize = \ - self._conn[CONN_INTERFACE_AVATARS].GetAvatarRequirements() - if not "image/jpeg" in types: - _logger.debug("server does not accept JPEG format avatars.") - return - - img_data = _get_buddy_icon_at_size(icon_data, min(maxw, 96), - min(maxh, 96), maxsize) - self._conn[CONN_INTERFACE_AVATARS].SetAvatar(img_data, "image/jpeg", - reply_handler=set_self_avatar_cb, - error_handler=lambda e: self._log_error_cb("setting avatar", e)) - - def emit_joined_activity(self, activity_id, room): - self._joined_activities.append((activity_id, room)) - self._set_self_activities() - - def _join_activity_get_channel_cb(self, activity_id, callback, err_cb, - handles): - if not self._activities.has_key(activity_id): - self._activities[activity_id] = handles[0] - - if (activity_id, handles[0]) in self._joined_activities: - e = RuntimeError("Already joined activity %s" % activity_id) - _logger.debug('%s', e) - err_cb(e) - return - - self._conn[CONN_INTERFACE].RequestChannel(CHANNEL_TYPE_TEXT, - HANDLE_TYPE_ROOM, handles[0], True, - reply_handler=lambda path: callback(handles[0], path), - error_handler=err_cb) - - def join_activity(self, activity_id, callback, err_cb): - """Share activity with the network, or join an activity on the - network (or locally) - - activity_id -- unique ID for the activity - callback -- callback to be called when the join succeeds or fails, - with arguments: - activity room handle: int or long - channel: object path - err_cb -- callback to be called on failure, with one Exception argument - - Asks the Telepathy server to create a "conference" channel - for the activity or return a handle to an already created - conference channel for the activity. - """ - handle = self._activities.get(activity_id) - if not handle: - # FIXME: figure out why the server can't figure this out itself - room_jid = activity_id + "@conference." + self._account["server"] - self._conn[CONN_INTERFACE].RequestHandles(HANDLE_TYPE_ROOM, - [room_jid], - reply_handler=lambda *args: self._join_activity_get_channel_cb( - activity_id, callback, err_cb, *args), - error_handler=err_cb) - else: - self._join_activity_get_channel_cb(activity_id, callback, err_cb, - [handle]) - - def _ignore_success_cb(self): - """Ignore an event (null-operation)""" - - def _log_error_cb(self, msg, err): - """Log a message (error) at debug level with prefix msg""" - _logger.debug("Error %s: %s", msg, err) - - def _set_self_olpc_properties(self): - """Set color and key on our Telepathy server identity""" - props = {} - props['color'] = self._owner.props.color - props['key'] = dbus.ByteArray(self._owner.props.key) - addr = self._owner.props.ip4_address - if not addr: - props['ip4-address'] = "" - else: - props['ip4-address'] = addr - self._conn[CONN_INTERFACE_BUDDY_INFO].SetProperties(props, - reply_handler=self._ignore_success_cb, - error_handler=lambda e: self._log_error_cb("setting properties", e)) - - def _set_self_alias(self): - """Forwarded to SetActivities on AliasInfo channel""" - alias = self._owner.props.nick - self_handle = self._conn[CONN_INTERFACE].GetSelfHandle() - self._conn[CONN_INTERFACE_ALIASING].SetAliases({self_handle : alias}, - reply_handler=self._ignore_success_cb, - error_handler=lambda e: self._log_error_cb("setting alias", e)) - return False - - def _set_self_activities(self): - """Forward set of joined activities to network - - uses SetActivities on BuddyInfo channel - """ - self._conn[CONN_INTERFACE_BUDDY_INFO].SetActivities( - self._joined_activities, - reply_handler=self._ignore_success_cb, - error_handler=lambda e: self._log_error_cb("setting activities", e)) - - def _set_self_current_activity(self): - """Forward our current activity (or "") to network - - uses SetCurrentActivity on BuddyInfo channel - """ - cur_activity = self._owner.props.current_activity - cur_activity_handle = 0 - if not cur_activity: - cur_activity = "" - else: - cur_activity_handle = self._get_handle_for_activity(cur_activity) - if not cur_activity_handle: - # dont advertise a current activity that's not shared - cur_activity = "" - - _logger.debug("Setting current activity to '%s' (handle %s)", - cur_activity, cur_activity_handle) - self._conn[CONN_INTERFACE_BUDDY_INFO].SetCurrentActivity(cur_activity, - cur_activity_handle, - reply_handler=self._ignore_success_cb, - error_handler=lambda e: self._log_error_cb("setting current activity", e)) - - def _get_handle_for_activity(self, activity_id): - """Retrieve current handle for given activity or None""" - for (act, handle) in self._joined_activities: - if activity_id == act: - return handle - return None - - def _reconnect_cb(self): - """Attempt to reconnect to the server""" - self.start() - return False - - def _handle_connection_status_change(self, status, reason): - if status == self._conn_status: - return - - if status == CONNECTION_STATUS_CONNECTING: - self._conn_status = status - _logger.debug("status: connecting...") - elif status == CONNECTION_STATUS_CONNECTED: - if self._connected_cb(): - _logger.debug("status: connected") - self._conn_status = status - else: - self.cleanup() - _logger.debug("status: was connected, but an error occurred") - elif status == CONNECTION_STATUS_DISCONNECTED: - self.cleanup() - _logger.debug("status: disconnected (reason %r)", reason) - if reason == CONNECTION_STATUS_REASON_AUTHENTICATION_FAILED: - # FIXME: handle connection failure; retry later? - pass - else: - # If disconnected, but still have a network connection, retry - # If disconnected and no network connection, do nothing here - # and let the IP4AddressMonitor address-changed signal handle - # reconnection - if self._ip4am.props.address and not self._reconnect_id: - self._reconnect_id = gobject.timeout_add(_RECONNECT_TIMEOUT, - self._reconnect_cb) - - self.emit('status', self._conn_status, int(reason)) - return False - - def _status_changed_cb(self, status, reason): - """Handle notification of connection-status change - - status -- CONNECTION_STATUS_* - reason -- integer code describing the reason... - """ - _logger.debug("::: connection status changed to %s", status) - self._handle_connection_status_change(status, reason) - - def start(self): - """Start up the Telepathy networking connections - - if we are already connected, query for the initial contact - information. - - if we are already connecting, do nothing - - otherwise initiate a connection and transfer control to - _connect_reply_cb or _connect_error_cb - """ - _logger.debug("Starting up...") - - if self._reconnect_id > 0: - gobject.source_remove(self._reconnect_id) - self._reconnect_id = 0 - - # Only init connection if we have a valid IP address - if self._ip4am.props.address: - _logger.debug("::: Have IP4 address %s, will connect", - self._ip4am.props.address) - self._init_connection() - else: - _logger.debug("::: No IP4 address, postponing connection") - - def cleanup(self): - """If we still have a connection, disconnect it""" - if self._conn: - try: - self._conn[CONN_INTERFACE].Disconnect() - except: - pass - self._conn = None - self._conn_status = CONNECTION_STATUS_DISCONNECTED - - for handle in self._online_contacts.keys(): - self._contact_offline(handle) - self._online_contacts = {} - self._joined_activities = [] - self._activities = {} - - if self._reconnect_id > 0: - gobject.source_remove(self._reconnect_id) - self._reconnect_id = 0 - - def _contact_offline(self, handle): - """Handle contact going offline (send message, update set)""" - if not self._online_contacts.has_key(handle): - return - if self._online_contacts[handle]: - self.emit("contact-offline", handle) - del self._online_contacts[handle] - - def _contact_online_activities_cb(self, handle, activities): - """Handle contact's activity list update""" - self._buddy_activities_changed_cb(handle, activities) - - def _contact_online_activities_error_cb(self, handle, err): - """Handle contact's activity list being unavailable""" - _logger.debug("Handle %s - Error getting activities: %s", - handle, err) - # Don't drop the buddy if we can't get their activities, for now - #self._contact_offline(handle) - - def _contact_online_aliases_cb(self, handle, props, aliases): - """Handle contact's alias being received (do further queries)""" - if not self._conn or not aliases or not len(aliases): - _logger.debug("Handle %s - No aliases", handle) - self._contact_offline(handle) - return - - props['nick'] = aliases[0] - - jid = self._conn[CONN_INTERFACE].InspectHandles(HANDLE_TYPE_CONTACT, - [handle])[0] - self._online_contacts[handle] = jid - objid = self.identify_contacts(None, [handle])[handle] - - self.emit("contact-online", objid, handle, props) - - self._conn[CONN_INTERFACE_BUDDY_INFO].GetActivities(handle, - reply_handler=lambda *args: self._contact_online_activities_cb( - handle, *args), - error_handler=lambda e: self._contact_online_activities_error_cb( - handle, e)) - - def _contact_online_aliases_error_cb(self, handle, props, retry, err): - """Handle failure to retrieve given user's alias/information""" - if retry: - _logger.debug("Handle %s - Error getting nickname (will retry):" - "%s", handle, err) - self._conn[CONN_INTERFACE_ALIASING].RequestAliases([handle], - reply_handler=lambda *args: self._contact_online_aliases_cb( - handle, props, *args), - error_handler=lambda e: self._contact_online_aliases_error_cb( - handle, props, False, e)) - else: - _logger.debug("Handle %s - Error getting nickname: %s", - handle, err) - self._contact_offline(handle) - - def _contact_online_properties_cb(self, handle, props): - """Handle failure to retrieve given user's alias/information""" - if not props.has_key('key'): - _logger.debug("Handle %s - invalid key.", handle) - self._contact_offline(handle) - return - if not props.has_key('color'): - _logger.debug("Handle %s - invalid color.", handle) - self._contact_offline(handle) - return - - self._conn[CONN_INTERFACE_ALIASING].RequestAliases([handle], - reply_handler=lambda *args: self._contact_online_aliases_cb( - handle, props, *args), - error_handler=lambda e: self._contact_online_aliases_error_cb( - handle, props, True, e)) - - def _contact_online_request_properties(self, handle, tries): - self._conn[CONN_INTERFACE_BUDDY_INFO].GetProperties(handle, - byte_arrays=True, - reply_handler=lambda *args: self._contact_online_properties_cb( - handle, *args), - error_handler=lambda e: self._contact_online_properties_error_cb( - handle, tries, e)) - return False - - def _contact_online_properties_error_cb(self, handle, tries, err): - """Handle error retrieving property-set for a user (handle)""" - if tries <= 3: - _logger.debug("Handle %s - Error getting properties (will retry):" - " %s", handle, err) - tries += 1 - gobject.timeout_add(1000, self._contact_online_request_properties, - handle, tries) - else: - _logger.debug("Handle %s - Error getting properties: %s", - handle, err) - self._contact_offline(handle) - - def _contact_online(self, handle): - """Handle a contact coming online""" - if (handle not in self._subscribe_members and - handle not in self._subscribe_local_pending and - handle not in self._subscribe_remote_pending): - # it's probably a channel-specific handle - can't create a Buddy - # object for those yet - return - - self._online_contacts[handle] = None - if handle == self._conn[CONN_INTERFACE].GetSelfHandle(): - jid = self._conn[CONN_INTERFACE].InspectHandles( - HANDLE_TYPE_CONTACT, [handle])[0] - self._online_contacts[handle] = jid - # ignore network events for Owner property changes since those - # are handled locally - return - - self._contact_online_request_properties(handle, 1) - - def _subscribe_members_changed_cb(self, message, added, removed, - local_pending, remote_pending, - actor, reason): - - added = set(added) - removed = set(removed) - local_pending = set(local_pending) - remote_pending = set(remote_pending) - - affected = added|removed - affected |= local_pending - affected |= remote_pending - - self._subscribe_members -= affected - self._subscribe_members |= added - self._subscribe_local_pending -= affected - self._subscribe_local_pending |= local_pending - self._subscribe_remote_pending -= affected - self._subscribe_remote_pending |= remote_pending - - def _publish_members_changed_cb(self, added, removed, local_pending, - remote_pending, actor, reason): - - if local_pending: - # accept all requested subscriptions - self._publish_channel[CHANNEL_INTERFACE_GROUP].AddMembers( - local_pending, '') - - # subscribe to people who've subscribed to us, if necessary - added = list(set(added) - self._subscribe_members - - self._subscribe_remote_pending) - if added: - self._subscribe_channel[CHANNEL_INTERFACE_GROUP].AddMembers( - added, '') - - def _presence_update_cb(self, presence): - """Send update for online/offline status of presence""" - for handle in presence: - timestamp, statuses = presence[handle] - online = handle in self._online_contacts - for status, params in statuses.items(): - if not online and status == "offline": - # weren't online in the first place... - continue - jid = self._conn[CONN_INTERFACE].InspectHandles( - HANDLE_TYPE_CONTACT, [handle])[0] - olstr = "ONLINE" - if not online: olstr = "OFFLINE" - _logger.debug("Handle %s (%s) was %s, status now '%s'.", - handle, jid, olstr, status) - if not online and status in ["available", "away", "brb", - "busy", "dnd", "xa"]: - self._contact_online(handle) - elif status in ["offline", "invisible"]: - self._contact_offline(handle) - - def _request_avatar_cb(self, handle, new_avatar_token, avatar, mime_type): - jid = self._online_contacts[handle] - if not jid: - logging.debug("Handle %s not valid yet..." % handle) - return - icon = ''.join(map(chr, avatar)) - self._icon_cache.store_icon(self._conn.object_path, jid, - new_avatar_token, icon) - self.emit("avatar-updated", handle, icon) - - def _avatar_updated_cb(self, handle, new_avatar_token): - """Handle update of given user (handle)'s avatar""" - if handle == self._conn[CONN_INTERFACE].GetSelfHandle(): - # ignore network events for Owner property changes since those - # are handled locally - return - - if not self._online_contacts.has_key(handle): - _logger.debug("Handle %s unknown.", handle) - return - - jid = self._online_contacts[handle] - if not jid: - _logger.debug("Handle %s not valid yet...", handle) - return - - icon = self._icon_cache.get_icon(self._conn.object_path, jid, - new_avatar_token) - if not icon: - # cache miss - self._conn[CONN_INTERFACE_AVATARS].RequestAvatar(handle, - reply_handler=lambda *args: self._request_avatar_cb(handle, - new_avatar_token, *args), - error_handler=lambda e: self._log_error_cb( - "getting avatar", e)) - else: - self.emit("avatar-updated", handle, icon) - - def _alias_changed_cb(self, aliases): - """Handle update of aliases for all users""" - for handle, alias in aliases: - prop = {'nick': alias} - #print "Buddy %s alias changed to %s" % (handle, alias) - if (self._online_contacts.has_key(handle) and - self._online_contacts[handle]): - self._buddy_properties_changed_cb(handle, prop) - - def _buddy_properties_changed_cb(self, handle, properties): - """Handle update of given user (handle)'s properties""" - if handle == self._conn[CONN_INTERFACE].GetSelfHandle(): - # ignore network events for Owner property changes since those - # are handled locally - return - if (self._online_contacts.has_key(handle) and - self._online_contacts[handle]): - self.emit("buddy-properties-changed", handle, properties) - - def _buddy_activities_changed_cb(self, handle, activities): - """Handle update of given user (handle)'s activities""" - if handle == self._conn[CONN_INTERFACE].GetSelfHandle(): - # ignore network events for Owner activity changes since those - # are handled locally - return - if (not self._online_contacts.has_key(handle) or - not self._online_contacts[handle]): - return - - activities_dict = {} - for act_id, act_handle in activities: - self._activities[act_id] = act_handle - activities_dict[act_id] = act_handle - self.emit("buddy-activities-changed", handle, activities_dict) - - def _buddy_current_activity_changed_cb(self, handle, activity, channel): - """Handle update of given user (handle)'s current activity""" - - if handle == self._conn[CONN_INTERFACE].GetSelfHandle(): - # ignore network events for Owner current activity changes since - # those are handled locally - return - if (not self._online_contacts.has_key(handle) or - not self._online_contacts[handle]): - return - - if not len(activity) or not util.validate_activity_id(activity): - activity = None - prop = {'current-activity': activity} - _logger.debug("Handle %s: current activity now %s", handle, activity) - self._buddy_properties_changed_cb(handle, prop) - - def _new_channel_cb(self, object_path, channel_type, handle_type, handle, - suppress_handler): - """Handle creation of a new channel - """ - if (handle_type == HANDLE_TYPE_ROOM and - channel_type == CHANNEL_TYPE_TEXT): - def ready(channel): - - for act_id, act_handle in self._activities.iteritems(): - if handle == act_handle: - break - else: - return - - def got_all_members(current, local_pending, remote_pending): - if local_pending: - for act_id, act_handle in self._activities.iteritems(): - if handle == act_handle: - self.emit('activity-invitation', act_id, handle) - def got_all_members_err(e): - logger.debug('Unable to get channel members for %s:', - object_path, exc_info=1) - - group = channel[CHANNEL_INTERFACE_GROUP] - group.GetAllMembers(reply_handler=got_all_members, - error_handler=got_all_members_err) - - # we throw away the channel as soon as ready() finishes - Channel(self._conn.service_name, object_path, - ready_handler=ready) - - elif (handle_type == HANDLE_TYPE_CONTACT and - channel_type in (CHANNEL_TYPE_TEXT, - CHANNEL_TYPE_STREAMED_MEDIA)): - self.emit("private-invitation", object_path) - - def _activity_properties_changed_cb(self, room, properties): - """Handle update of properties for a "room" (activity handle)""" - for act_id, act_handle in self._activities.items(): - if room == act_handle: - self.emit("activity-properties-changed", act_id, room, properties) - return + TelepathyPlugin._connected_cb(self) def _server_is_trusted(self, hostname): """Return True if the server with the given hostname is trusted to @@ -998,7 +144,7 @@ class ServerPlugin(gobject.GObject): """ return (hostname == 'olpc.collabora.co.uk') - def identify_contacts(self, tp_chan, handles): + def identify_contacts(self, tp_chan, handles, identifiers=None): """Work out the "best" unique identifier we can for the given handles, in the context of the given channel (which may be None), using only 'fast' connection manager API (that does not involve network @@ -1041,6 +187,7 @@ class ServerPlugin(gobject.GObject): group = tp_chan[CHANNEL_INTERFACE_GROUP] if (group.GetGroupFlags() & CHANNEL_GROUP_FLAG_CHANNEL_SPECIFIC_HANDLES): + identifiers = None owners = group.GetHandleOwners(handles) for i, owner in enumerate(owners): if owner == 0: @@ -1048,11 +195,12 @@ class ServerPlugin(gobject.GObject): else: group = None - jids = self._conn[CONN_INTERFACE].InspectHandles(HANDLE_TYPE_CONTACT, - owners) + if identifiers is None: + identifiers = self._conn[CONN_INTERFACE].InspectHandles( + HANDLE_TYPE_CONTACT, owners) ret = {} - for handle, jid in zip(handles, jids): + for handle, jid in izip(handles, identifiers): # special-case the Owner - we always know who we are if (handle == self.self_handle or (group is not None and handle == group.GetSelfHandle())): @@ -1076,3 +224,38 @@ class ServerPlugin(gobject.GObject): ret[handle] = 'xmpp/' + psutils.escape_identifier(jid) return ret + + def _connected_cb(self): + TelepathyPlugin._connected_cb(self) + + publish_handles, local_pending, remote_pending = \ + self._publish_channel[CHANNEL_INTERFACE_GROUP].GetAllMembers() + + if local_pending: + # accept pending subscriptions + # FIXME: do this async + publish[CHANNEL_INTERFACE_GROUP].AddMembers(local_pending, '') + + # request subscriptions from people subscribed to us if we're not + # subscribed to them + not_subscribed = set(publish_handles) + not_subscribed -= self._subscribe_members + subscribe[CHANNEL_INTERFACE_GROUP].AddMembers(not_subscribed, '') + + def _publish_members_changed_cb(self, message, added, removed, + local_pending, remote_pending, + actor, reason): + TelepathyPlugin._publish_members_changed_cb() + + if local_pending: + # accept all requested subscriptions + self._publish_channel[CHANNEL_INTERFACE_GROUP].AddMembers( + local_pending, '') + + # subscribe to people who've subscribed to us, if necessary + if self._subscribe_channel is not None: + added = list(set(added) - self._subscribe_members + - self._subscribe_remote_pending) + if added: + self._subscribe_channel[CHANNEL_INTERFACE_GROUP].AddMembers( + added, '') diff --git a/src/telepathy_plugin.py b/src/telepathy_plugin.py new file mode 100644 index 0000000..7d84536 --- /dev/null +++ b/src/telepathy_plugin.py @@ -0,0 +1,460 @@ +"""Base class for Telepathy plugins.""" + +# Copyright (C) 2007, Red Hat, Inc. +# Copyright (C) 2007, Collabora Ltd. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + +import logging +from itertools import izip + +import gobject + +from telepathy.client import (Channel, Connection) +from telepathy.constants import (CONNECTION_STATUS_DISCONNECTED, + CONNECTION_STATUS_CONNECTING, CONNECTION_STATUS_CONNECTED, + CONNECTION_STATUS_REASON_AUTHENTICATION_FAILED, + CONNECTION_STATUS_REASON_NONE_SPECIFIED, + HANDLE_TYPE_CONTACT, HANDLE_TYPE_ROOM, HANDLE_TYPE_LIST) +from telepathy.interfaces import (CONN_INTERFACE, CHANNEL_TYPE_TEXT, + CHANNEL_TYPE_STREAMED_MEDIA, CHANNEL_INTERFACE_GROUP, + CONN_INTERFACE_PRESENCE, CONN_INTERFACE_AVATARS, + CONN_INTERFACE_ALIASING, CHANNEL_TYPE_CONTACT_LIST, + CONN_MGR_INTERFACE) + + +CONN_INTERFACE_BUDDY_INFO = 'org.laptop.Telepathy.BuddyInfo' +CONN_INTERFACE_ACTIVITY_PROPERTIES = 'org.laptop.Telepathy.ActivityProperties' + + +_logger = logging.getLogger('s-p-s.telepathy_plugin') + + +class TelepathyPlugin(gobject.GObject): + __gsignals__ = { + 'contacts-online': + # Contacts in the subscribe list have come online. + # args: + # contact identification (based on key ID or JID): list of str + # contact handle: list of int or long + # contact identifier (JID): list of str or unicode + (gobject.SIGNAL_RUN_FIRST, None, [object, object, object]), + 'contacts-offline': + # Contacts in the subscribe list have gone offline. + # args: iterable over contact handles + (gobject.SIGNAL_RUN_FIRST, None, [object]), + 'status': + # Connection status changed. + # args: status, reason as for Telepathy StatusChanged + (gobject.SIGNAL_RUN_FIRST, None, [int, int]), + 'activity-invitation': + # We were invited to join an activity + # args: + # activity room handle: int or long + (gobject.SIGNAL_RUN_FIRST, None, [object]), + 'private-invitation': + # We were invited to join a chat or a media call + # args: + # channel object path + (gobject.SIGNAL_RUN_FIRST, None, [object]), + } + + _RECONNECT_TIMEOUT = 5000 + _TP_CONN_MANAGER = 'gabble' + _PROTOCOL = 'jabber' + + def __init__(self, registry, owner): + """Initialize the ServerPlugin instance + + :Parameters: + `registry` : telepathy.client.ManagerRegistry + From the PresenceService, used to find the connection + manager details + `owner` : buddy.GenericOwner + The Buddy representing the owner of this XO (normally a + buddy.ShellOwner instance) + """ + gobject.GObject.__init__(self) + + #: The connection, a `telepathy.client.Connection` + self._conn = None + + #: List of dbus-python SignalMatch objects representing signal match + #: rules associated with the connection, so we don't leak the match + #: rules when disconnected. + self._matches = [] + + #: The manager registry, a `telepathy.client.ManagerRegistry` + self._registry = registry + + #: set of contact handles: those for whom we've emitted contacts-online + self._online_contacts = set() + + #: The owner, a `buddy.GenericOwner` + self._owner = owner + #: The owner's handle on this connection + self.self_handle = None + #: The owner's identifier (e.g. JID) on this connection + self.self_identifier = None + + #: The connection's status + self._conn_status = CONNECTION_STATUS_DISCONNECTED + + #: GLib signal ID for reconnections + self._reconnect_id = 0 + + #: Parameters for the connection manager + self._account = self._get_account_info() + + #: The ``subscribe`` channel: a `telepathy.client.Channel` or None + self._subscribe_channel = None + #: The members of the ``subscribe`` channel + self._subscribe_members = set() + #: The local-pending members of the ``subscribe`` channel + self._subscribe_local_pending = set() + #: The remote-pending members of the ``subscribe`` channel + self._subscribe_remote_pending = set() + + #: The ``publish`` channel: a `telepathy.client.Channel` or None + self._publish_channel = None + + @property + def status(self): + """Return the Telepathy connection status.""" + return self._conn_status + + def get_connection(self): + """Retrieve our telepathy.client.Connection object""" + return self._conn + + def _get_account_info(self): + """Retrieve connection manager parameters for this account + """ + raise NotImplementedError + + def start(self): + raise NotImplementedError + + def suggest_room_for_activity(self, activity_id): + """Suggest a room to use to share the given activity. + """ + return activity_id + + def identify_contacts(self, tp_chan, handles, identifiers=None): + raise NotImplementedError + + def _reconnect_cb(self): + """Attempt to reconnect to the server""" + self.start() + return False + + def _init_connection(self): + """Set up our connection + + if there is no existing connection + (_find_existing_connection returns None) + produce a new connection with our protocol for our + account. + + if there is an existing connection, reuse it by + registering for various of events on it. + """ + conn = self._find_existing_connection() + if not conn: + conn = self._make_new_connection() + + m = conn[CONN_INTERFACE].connect_to_signal('StatusChanged', + self._handle_connection_status_change) + self._matches.append(m) + m = conn[CONN_INTERFACE].connect_to_signal('NewChannel', + self._new_channel_cb) + self._matches.append(m) + + # hack + conn._valid_interfaces.add(CONN_INTERFACE_PRESENCE) + conn._valid_interfaces.add(CONN_INTERFACE_BUDDY_INFO) + conn._valid_interfaces.add(CONN_INTERFACE_ACTIVITY_PROPERTIES) + conn._valid_interfaces.add(CONN_INTERFACE_AVATARS) + conn._valid_interfaces.add(CONN_INTERFACE_ALIASING) + + m = conn[CONN_INTERFACE_PRESENCE].connect_to_signal('PresenceUpdate', + self._presence_update_cb) + self._matches.append(m) + + self._conn = conn + status = self._conn[CONN_INTERFACE].GetStatus() + + self._owner.set_properties_before_connect(self) + + if status == CONNECTION_STATUS_DISCONNECTED: + def connect_reply(): + _logger.debug('%r: Connect() succeeded', self) + def connect_error(e): + _logger.debug('%r: Connect() failed: %s', self, e) + if not self._reconnect_id: + self._reconnect_id = gobject.timeout_add(self._RECONNECT_TIMEOUT, + self._reconnect_cb) + + self._conn[CONN_INTERFACE].Connect(reply_handler=connect_reply, + error_handler=connect_error) + + self._handle_connection_status_change(status, + CONNECTION_STATUS_REASON_NONE_SPECIFIED) + + def _find_existing_connection(self): + raise NotImplementedError + + def _make_new_connection(self): + acct = self._account.copy() + + # Create a new connection + mgr = self._registry.GetManager(self._TP_CONN_MANAGER) + name, path = mgr[CONN_MGR_INTERFACE].RequestConnection( + self._PROTOCOL, acct) + conn = Connection(name, path) + del acct + return conn + + def _handle_connection_status_change(self, status, reason): + if status == self._conn_status: + return + + if status == CONNECTION_STATUS_CONNECTING: + self._conn_status = status + _logger.debug("%r: connecting...", self) + elif status == CONNECTION_STATUS_CONNECTED: + self._conn_status = status + _logger.debug("%r: connected", self) + self._connected_cb() + elif status == CONNECTION_STATUS_DISCONNECTED: + self.stop() + _logger.debug("%r: disconnected (reason %r)", self, reason) + if reason == CONNECTION_STATUS_REASON_AUTHENTICATION_FAILED: + # FIXME: handle connection failure; retry later? + pass + else: + # If disconnected, but still have a network connection, retry + # If disconnected and no network connection, do nothing here + # and let the IP4AddressMonitor address-changed signal handle + # reconnection + if self._could_connect() and not self._reconnect_id: + self._reconnect_id = gobject.timeout_add(self._RECONNECT_TIMEOUT, + self._reconnect_cb) + + self.emit('status', self._conn_status, int(reason)) + + def _could_connect(self): + return True + + def stop(self): + """If we still have a connection, disconnect it""" + + matches = self._matches + self._matches = [] + for match in matches: + match.remove() + + if self._conn: + try: + self._conn[CONN_INTERFACE].Disconnect() + except: + pass + self._conn = None + self._conn_status = CONNECTION_STATUS_DISCONNECTED + + if self._online_contacts: + self._contacts_offline(self._online_contacts) + + if self._reconnect_id > 0: + gobject.source_remove(self._reconnect_id) + self._reconnect_id = 0 + + def cleanup(self): + self.stop() + + def _contacts_offline(self, handles): + """Handle contacts going offline (send message, update set)""" + self._online_contacts -= handles + _logger.debug('%r: Contacts now offline: %r', self, handles) + self.emit("contacts-offline", handles) + + def _contacts_online(self, handles): + """Handle contacts coming online""" + relevant = [] + + for handle in handles: + if handle == self.self_handle: + # ignore network events for Owner property changes since those + # are handled locally + pass + elif (handle in self._subscribe_members or + handle in self._subscribe_local_pending or + handle in self._subscribe_remote_pending): + relevant.append(handle) + # else it's probably a channel-specific handle - can't create a + # Buddy object for those yet + + if not relevant: + return + + jids = self._conn[CONN_INTERFACE].InspectHandles( + HANDLE_TYPE_CONTACT, relevant) + + handle_to_objid = self.identify_contacts(None, relevant, jids) + objids = [] + for handle in relevant: + objids.append(handle_to_objid[handle]) + + self._online_contacts |= frozenset(relevant) + _logger.debug('%r: Contacts now online:', self) + for handle, objid in izip(relevant, objids): + _logger.debug(' %u .../%s', handle, objid) + self.emit('contacts-online', objids, relevant, jids) + + def _subscribe_members_changed_cb(self, message, added, removed, + local_pending, remote_pending, + actor, reason): + + added = set(added) + removed = set(removed) + local_pending = set(local_pending) + remote_pending = set(remote_pending) + + affected = added|removed + affected |= local_pending + affected |= remote_pending + + self._subscribe_members -= affected + self._subscribe_members |= added + self._subscribe_local_pending -= affected + self._subscribe_local_pending |= local_pending + self._subscribe_remote_pending -= affected + self._subscribe_remote_pending |= remote_pending + + def _publish_members_changed_cb(self, message, added, removed, + local_pending, remote_pending, actor, reason): + pass + + def _presence_update_cb(self, presence): + """Send update for online/offline status of presence""" + + now_online = set() + now_offline = set(presence.iterkeys()) + + for handle in presence: + timestamp, statuses = presence[handle] + for status, params in statuses.items(): + # FIXME: use correct logic involving the GetStatuses method + if status in ["available", "away", "brb", "busy", "dnd", "xa"]: + now_online.add(handle) + now_offline.discard(handle) + + now_online -= self._online_contacts + now_offline &= self._online_contacts + + if now_online: + self._contacts_online(now_online) + if now_offline: + self._contacts_offline(now_offline) + + def _new_channel_cb(self, object_path, channel_type, handle_type, handle, + suppress_handler): + """Handle creation of a new channel + """ + if (handle_type == HANDLE_TYPE_ROOM and + channel_type == CHANNEL_TYPE_TEXT): + def ready(channel): + def got_all_members(current, local_pending, remote_pending): + if local_pending: + self.emit('activity-invitation', handle) + def got_all_members_err(e): + _logger.debug('Unable to get channel members for %s:', + object_path, exc_info=1) + + group = channel[CHANNEL_INTERFACE_GROUP] + group.GetAllMembers(reply_handler=got_all_members, + error_handler=got_all_members_err) + + # we throw away the channel as soon as ready() finishes + Channel(self._conn.service_name, object_path, + ready_handler=ready) + + elif (handle_type == HANDLE_TYPE_CONTACT and + channel_type in (CHANNEL_TYPE_TEXT, + CHANNEL_TYPE_STREAMED_MEDIA)): + self.emit("private-invitation", object_path) + + def _connected_cb(self): + """Callback on successful connection to a server + """ + # FIXME: cope with CMs that lack some of the interfaces + # FIXME: cope with CMs with no 'publish' or 'subscribe' + + # request both handles at the same time to reduce round-trips + pub_handle, sub_handle = self._conn[CONN_INTERFACE].RequestHandles( + HANDLE_TYPE_LIST, ['publish', 'subscribe']) + + # the group of contacts who may receive your presence + publish = self._conn.request_channel(CHANNEL_TYPE_CONTACT_LIST, + HANDLE_TYPE_LIST, pub_handle, True) + self._publish_channel = publish + m = publish[CHANNEL_INTERFACE_GROUP].connect_to_signal( + 'MembersChanged', self._publish_members_changed_cb) + self._matches.append(m) + + # the group of contacts for whom you wish to receive presence + subscribe = self._conn.request_channel(CHANNEL_TYPE_CONTACT_LIST, + HANDLE_TYPE_LIST, sub_handle, True) + self._subscribe_channel = subscribe + m = subscribe[CHANNEL_INTERFACE_GROUP].connect_to_signal( + 'MembersChanged', self._subscribe_members_changed_cb) + self._matches.append(m) + subscribe_handles, subscribe_lp, subscribe_rp = \ + subscribe[CHANNEL_INTERFACE_GROUP].GetAllMembers() + self._subscribe_members = set(subscribe_handles) + self._subscribe_local_pending = set(subscribe_lp) + self._subscribe_remote_pending = set(subscribe_rp) + + # FIXME: do this async? + self.self_handle = self._conn[CONN_INTERFACE].GetSelfHandle() + self.self_identifier = self._conn[CONN_INTERFACE].InspectHandles( + HANDLE_TYPE_CONTACT, [self.self_handle])[0] + + # Request presence for everyone we're subscribed to + self._conn[CONN_INTERFACE_PRESENCE].RequestPresence(subscribe_handles) + + def start(self): + """Start up the Telepathy networking connections + + if we are already connected, query for the initial contact + information. + + if we are already connecting, do nothing + + otherwise initiate a connection and transfer control to + _connect_reply_cb or _connect_error_cb + """ + + _logger.debug("%r: Starting up...", self) + + if self._reconnect_id > 0: + gobject.source_remove(self._reconnect_id) + self._reconnect_id = 0 + + # Only init connection if we have a valid IP address + if self._could_connect(): + self._init_connection() + else: + _logger.debug('%r: Postponing connection', self) |