rendered paste bodyIndex: coherence/extern/telepathy/client.py===================================================================--- coherence/extern/telepathy/client.py (revision 1360)+++ coherence/extern/telepathy/client.py (working copy)@@ -158,7 +158,7 @@ state)) channel_iface = tube[CHANNEL_INTERFACE] channel_iface.connect_to_signal("Closed",- lambda: self.tube_closed_cb(tube))+ lambda: self.tube_closed(tube)) def got_tube(self, tube): props = tube.props@@ -203,7 +203,7 @@ if state == TUBE_CHANNEL_STATE_OPEN: self.tube_opened(tube) - def tube_closed_cb (self, tube):+ def tube_closed(self, tube): tube_path = tube.object_path self.info("tube %r closed", tube_path) self._tube_conns[tube_path].close()Index: coherence/extern/telepathy/tube.py===================================================================--- coherence/extern/telepathy/tube.py (revision 1360)+++ coherence/extern/telepathy/tube.py (working copy)@@ -5,15 +5,12 @@ from coherence.extern.telepathy.client import Client -class TubePublisher(Client):- logCategory = "tube_publisher"+class TubePublisherMixin(object): - def __init__(self, manager, protocol, account, muc_id, tubes_to_offer):- Client.__init__(self, manager, protocol, account, muc_id)+ def __init__(self, tubes_to_offer): self._tubes_to_offer = tubes_to_offer def muc_joined(self):- super(TubePublisher, self).muc_joined() self.info("muc joined. Offering the tubes") for interface in self._tubes_to_offer.keys():@@ -24,30 +21,40 @@ CHANNEL_TYPE_DBUS_TUBE + ".ServiceName": interface}) def got_tube(self, tube):- super(TubePublisher, self).got_tube(tube)+ super(TubePublisherMixin, self).got_tube(tube) initiator_handle = tube.props[CHANNEL_INTERFACE + ".InitiatorHandle"] if initiator_handle == self.self_handle:- self.info("offering my tube located at %r", tube.object_path)- service_name = tube.props[CHANNEL_TYPE_DBUS_TUBE + ".ServiceName"]- params = self._tubes_to_offer[service_name]- address = tube[CHANNEL_TYPE_DBUS_TUBE].Offer(params,- SOCKET_ACCESS_CONTROL_CREDENTIALS)- tube.local_address = address- self.info("local tube address: %r", address)+ self.finish_tube_offer(tube) -class TubeConsumer(Client):+ def finish_tube_offer(self, tube):+ self.info("offering my tube located at %r", tube.object_path)+ service_name = tube.props[CHANNEL_TYPE_DBUS_TUBE + ".ServiceName"]+ params = self._tubes_to_offer[service_name]+ address = tube[CHANNEL_TYPE_DBUS_TUBE].Offer(params,+ SOCKET_ACCESS_CONTROL_CREDENTIALS)+ tube.local_address = address+ self.info("local tube address: %r", address)++class TubePublisher(TubePublisherMixin, Client):+ logCategory = "tube_publisher"++ def __init__(self, manager, protocol, account, muc_id, tubes_to_offer):+ TubePublisherMixin.__init__(self, tubes_to_offer)+ Client.__init__(self, manager, protocol, account, muc_id)+++class TubeConsumerMixin(object): logCategory = "tube_consumer" - def __init__(self, manager, protocol,- account, muc_id, found_peer_callback=None,- disapeared_peer_callback=None, existing_client=False):- Client.__init__(self, manager, protocol,- account, muc_id, existing_client=existing_client)+ def __init__(self, found_peer_callback=None, disapeared_peer_callback=None): self.found_peer_callback = found_peer_callback self.disapeared_peer_callback = disapeared_peer_callback def got_tube(self, tube):- super(TubeConsumer, self).got_tube(tube)+ super(TubeConsumerMixin, self).got_tube(tube)+ self.accept_tube(tube)++ def accept_tube(self, tube): if self.pre_accept_tube(tube): self.info("accepting tube %r", tube.object_path) tube_iface = tube[CHANNEL_TYPE_DBUS_TUBE]@@ -58,6 +65,16 @@ def pre_accept_tube(self, tube): return True - def tube_closed_cb(self, tube):+ def tube_closed(self, tube): self.disapeared_peer_callback(tube)- super(TubeConsumer, self).tube_closed_cb(tube)+ super(TubeConsumerMixin, self).tube_closed(tube)++class TubeConsumer(TubeConsumerMixin, Client):+ logCategory = "tube_consumer"++ def __init__(self, manager, protocol,+ account, muc_id, found_peer_callback=None,+ disapeared_peer_callback=None):+ TubeConsumerMixin.__init__(self, found_peer_callback=found_peer_callback,+ disapeared_peer_callback=disapeared_peer_callback)+ Client.__init__(self, manager, protocol, account, muc_id)Index: coherence/extern/telepathy/mirabeau_tube_consumer.py===================================================================--- coherence/extern/telepathy/mirabeau_tube_consumer.py (revision 1360)+++ coherence/extern/telepathy/mirabeau_tube_consumer.py (working copy)@@ -1,20 +1,17 @@ from telepathy.interfaces import CHANNEL_TYPE_DBUS_TUBE, CONN_INTERFACE, \ CHANNEL_INTERFACE -from coherence.extern.telepathy import tube+from coherence.extern.telepathy import client, tube from coherence.dbus_constants import BUS_NAME, OBJECT_PATH, DEVICE_IFACE, SERVICE_IFACE from coherence import dbus_service -class MirabeauTubeConsumer(tube.TubeConsumer):- logCategory = "mirabeau_tube_consumer"+class MirabeauTubeConsumerMixin(tube.TubeConsumerMixin): - def __init__(self, manager, protocol, account, muc_id,- found_peer_callback=None, disapeared_peer_callback=None,- got_devices_callback=None, existing_client=False):- super(MirabeauTubeConsumer, self).__init__(manager, protocol, account,- muc_id, existing_client=existing_client,- found_peer_callback=found_peer_callback,- disapeared_peer_callback=disapeared_peer_callback)+ def __init__(self, found_peer_callback=None, disapeared_peer_callback=None,+ got_devices_callback=None):+ tube.TubeConsumerMixin.__init__(self,+ found_peer_callback=found_peer_callback,+ disapeared_peer_callback=disapeared_peer_callback) self.got_devices_callback = got_devices_callback self.info("MirabeauTubeConsumer created") self._coherence_tubes = {}@@ -25,10 +22,8 @@ ## return initiator in self.roster return True - def tube_opened(self, tube):- tube_conn = super(MirabeauTubeConsumer, self).tube_opened(tube)+ def post_tube_accept(self, tube, tube_conn, initiator_handle): service = tube.props[CHANNEL_TYPE_DBUS_TUBE + ".ServiceName"]- initiator_handle = tube.props[CHANNEL_INTERFACE + ".InitiatorHandle"] if service == BUS_NAME: tube.remote_object = dbus_service.DBusPontoon(None, tube_conn)@@ -46,14 +41,17 @@ if len(self._coherence_tubes[initiator_handle]) == 3: self.announce(initiator_handle)- return tube_conn + def tube_closed(self, tube):+ self.disapeared_peer_callback(tube)+ super(MirabeauTubeConsumerMixin, self).tube_closed(tube)+ def announce(self, initiator_handle): service_name = BUS_NAME pontoon_tube = self._coherence_tubes[initiator_handle][service_name] def cb(participants, removed):- if participants:+ if participants and initiator_handle in participants: initiator_bus_name = participants[initiator_handle] self.info("bus name %r for service %r", initiator_bus_name, service_name)@@ -69,24 +67,48 @@ device_tube = tubes[DEVICE_IFACE].remote_object.tube service_tube = tubes[SERVICE_IFACE].remote_object.tube self.info("using pontoon tube at %r", tubes[BUS_NAME].object_path)++ def got_devices(pontoon_devices):+ self.info("%r devices registered in remote pontoon", len(pontoon_devices))+ for device_dict in pontoon_devices:+ device_path = device_dict["path"]+ self.info("getting object at %r from %r", device_path,+ initiator_bus_name)+ proxy = device_tube.get_object(initiator_bus_name, device_path)+ infos = proxy.get_info(dbus_interface=DEVICE_IFACE)+ service_proxies = []+ for service_path in device_dict["services"]:+ service_proxy = service_tube.get_object(initiator_bus_name,+ service_path)+ service_proxies.append(service_proxy)+ proxy.services = service_proxies+ devices.append(proxy)+ self.got_devices_callback(devices)++ def got_error(exception):+ print ">>>", exception+ pontoon = pontoon_tube.get_object(initiator_bus_name, OBJECT_PATH)- pontoon_devices = pontoon.get_devices()- self.info("%r devices registered in remote pontoon", len(pontoon_devices))- for device_dict in pontoon_devices:- device_path = device_dict["path"]- self.info("getting object at %r from %r", device_path,- initiator_bus_name)- proxy = device_tube.get_object(initiator_bus_name, device_path)- try:- infos = proxy.get_info(dbus_interface=DEVICE_IFACE)- except Exception, exc:- self.warning(exc)- continue- service_proxies = []- for service_path in device_dict["services"]:- service_proxy = service_tube.get_object(initiator_bus_name,- service_path)- service_proxies.append(service_proxy)- proxy.services = service_proxies- devices.append(proxy)- self.got_devices_callback(devices)+ pontoon.get_devices_async(reply_handler=got_devices,+ error_handler=got_error)++class MirabeauTubeConsumer(MirabeauTubeConsumerMixin, client.Client):+ logCategory = "mirabeau_tube_consumer"++ def __init__(self, manager, protocol, account, muc_id,+ found_peer_callback=None, disapeared_peer_callback=None,+ got_devices_callback=None):+ MirabeauTubeConsumerMixin.__init__(self,+ found_peer_callback=found_peer_callback,+ disapeared_peer_callback=disapeared_peer_callback,+ got_devices_callback=got_devices_callback)+ client.Client.__init__(self, manager, protocol, account, muc_id)++ def got_tube(self, tube):+ client.Client.got_tube(self, tube)+ self.accept_tube(tube)++ def tube_opened(self, tube):+ tube_conn = super(MirabeauTubePublisherConsumer, self).tube_opened(tube)+ self.post_tube_accept(tube, tube_conn)+ return tube_connIndex: coherence/extern/telepathy/mirabeau_tube_publisher.py===================================================================--- coherence/extern/telepathy/mirabeau_tube_publisher.py (revision 1360)+++ coherence/extern/telepathy/mirabeau_tube_publisher.py (working copy)@@ -11,111 +11,20 @@ CHANNEL_INTERFACE from telepathy.constants import TUBE_CHANNEL_STATE_LOCAL_PENDING -from coherence.extern.telepathy import client+from coherence.extern.telepathy import client, tube, mirabeau_tube_consumer from coherence.dbus_constants import BUS_NAME, OBJECT_PATH, DEVICE_IFACE, SERVICE_IFACE from coherence import dbus_service -class MirabeauTubePublisherConsumer(client.Client):- logCategory = "mirabeau_tube_publisher"-- def __init__(self, manager, protocol, account, muc_id, tubes_to_offer,- application, allowed_devices, found_peer_callback=None,- disapeared_peer_callback=None, got_devices_callback=None):- super(MirabeauTubePublisherConsumer, self).__init__(manager, protocol,- account, muc_id)- self._tubes_to_offer = tubes_to_offer- self.found_peer_callback = found_peer_callback- self.disapeared_peer_callback=disapeared_peer_callback+class MirabeauTubePublisherMixin(tube.TubePublisherMixin):+ def __init__(self, tubes_to_offer, application, allowed_devices):+ tube.TubePublisherMixin.__init__(self, tubes_to_offer) self.coherence = application self.allowed_devices = allowed_devices- self.got_devices_callback = got_devices_callback self.coherence_tube = None self.device_tube = None self.service_tube = None self.announce_done = False - self._coherence_tubes = {}--- def muc_joined(self):- super(MirabeauTubePublisherConsumer, self).muc_joined()- self.info("muc joined. Offering the tubes")-- for interface in self._tubes_to_offer.keys():- self.conn[CONNECTION_INTERFACE_REQUESTS].CreateChannel({- CHANNEL_INTERFACE + ".ChannelType": CHANNEL_TYPE_DBUS_TUBE,- CHANNEL_INTERFACE + ".TargetHandleType": CONNECTION_HANDLE_TYPE_ROOM,- CHANNEL_INTERFACE + ".TargetID": self.muc_id,- CHANNEL_TYPE_DBUS_TUBE + ".ServiceName": interface})-- def got_tube(self, tube):- super(MirabeauTubePublisherConsumer, self).got_tube(tube)- initiator_handle = tube.props[CHANNEL_INTERFACE + ".InitiatorHandle"]- if initiator_handle == self.self_handle:- self.info("offering my tube located at %r", tube.object_path)- service_name = tube.props[CHANNEL_TYPE_DBUS_TUBE + ".ServiceName"]- params = self._tubes_to_offer[service_name]- address = tube[CHANNEL_TYPE_DBUS_TUBE].Offer(params,- SOCKET_ACCESS_CONTROL_CREDENTIALS)- tube.local_address = address- self.info("local tube address: %r", address)- elif self.pre_accept_tube(tube):- self.info("accepting tube %r", tube.object_path)- tube_iface = tube[CHANNEL_TYPE_DBUS_TUBE]- tube.local_address = tube_iface.Accept(SOCKET_ACCESS_CONTROL_CREDENTIALS)- else:- self.warning("tube %r not allowed", tube)-- def pre_accept_tube(self, tube):- return True-- def tube_opened(self, tube):- tube_conn = super(MirabeauTubePublisherConsumer, self).tube_opened(tube)- service = tube.props[CHANNEL_TYPE_DBUS_TUBE + ".ServiceName"]- initiator_handle = tube.props[CHANNEL_INTERFACE + ".InitiatorHandle"]- if initiator_handle == self.self_handle:- if service == BUS_NAME:- self.coherence.dbus.add_to_connection(tube_conn, OBJECT_PATH)- self.coherence_tube = tube_conn- elif service == DEVICE_IFACE:- self.device_tube = tube_conn- elif service == SERVICE_IFACE:- self.service_tube = tube_conn-- if not self.announce_done and None not in (self.coherence_tube,- self.device_tube,- self.service_tube):- self.announce_done = True- for device in self.coherence.dbus.devices.values():- self._register_device(device)- self.coherence.dbus.bus.add_signal_receiver(self._media_server_found,- "UPnP_ControlPoint_MediaServer_detected")- self.coherence.dbus.bus.add_signal_receiver(self._media_server_removed,- "UPnP_ControlPoint_MediaServer_removed")- else:- if service == BUS_NAME:- tube.remote_object = dbus_service.DBusPontoon(None, tube_conn)- elif service == DEVICE_IFACE:- tube.remote_object = dbus_service.DBusDevice(None, tube_conn)- elif service == SERVICE_IFACE:- tube.remote_object = dbus_service.DBusService(None, None, tube_conn)- else:- self.info("tube %r is not coming from Coherence", service)- return tube_conn-- if initiator_handle not in self._coherence_tubes:- self._coherence_tubes[initiator_handle] = {}- self._coherence_tubes[initiator_handle][service] = tube-- if len(self._coherence_tubes[initiator_handle]) == 3:- self.announce(initiator_handle)-- return tube_conn-- def tube_closed_cb(self, tube):- self.disapeared_peer_callback(tube)- super(MirabeauTubePublisherConsumer, self).tube_closed_cb(tube)- def _media_server_found(self, infos, udn): uuid = udn[5:] for device in self.coherence.dbus.devices.values():@@ -148,48 +57,56 @@ service.remove_from_connection(self.service_tube, service.path) return - def announce(self, initiator_handle):- service_name = BUS_NAME- pontoon_tube = self._coherence_tubes[initiator_handle][service_name]+ def post_tube_offer(self, tube, tube_conn):+ service = tube.props[CHANNEL_TYPE_DBUS_TUBE + ".ServiceName"]+ if service == BUS_NAME:+ self.coherence.dbus.add_to_connection(tube_conn, OBJECT_PATH)+ self.coherence_tube = tube_conn+ elif service == DEVICE_IFACE:+ self.device_tube = tube_conn+ elif service == SERVICE_IFACE:+ self.service_tube = tube_conn - def cb(participants, removed):- if participants and initiator_handle in participants:- initiator_bus_name = participants[initiator_handle]- self.info("bus name %r for service %r", initiator_bus_name,- service_name)- if initiator_bus_name is not None:- self.found_devices(initiator_handle, initiator_bus_name)+ if not self.announce_done and None not in (self.coherence_tube,+ self.device_tube,+ self.service_tube):+ self.announce_done = True+ for device in self.coherence.dbus.devices.values():+ self._register_device(device)+ self.coherence.dbus.bus.add_signal_receiver(self._media_server_found,+ "UPnP_ControlPoint_MediaServer_detected")+ self.coherence.dbus.bus.add_signal_receiver(self._media_server_removed,+ "UPnP_ControlPoint_MediaServer_removed") - pontoon_tube.remote_object.tube.watch_participants(cb)+class MirabeauTubePublisherConsumer(MirabeauTubePublisherMixin,+ mirabeau_tube_consumer.MirabeauTubeConsumerMixin,+ client.Client):+ logCategory = "mirabeau_tube_publisher" - def found_devices(self, initiator_handle, initiator_bus_name):- devices = []- tubes = self._coherence_tubes[initiator_handle]- pontoon_tube = tubes[BUS_NAME].remote_object.tube- device_tube = tubes[DEVICE_IFACE].remote_object.tube- service_tube = tubes[SERVICE_IFACE].remote_object.tube- self.info("using pontoon tube at %r", tubes[BUS_NAME].object_path)+ def __init__(self, manager, protocol, account, muc_id, tubes_to_offer,+ application, allowed_devices, found_peer_callback=None,+ disapeared_peer_callback=None, got_devices_callback=None):+ MirabeauTubePublisherMixin.__init__(self, tubes_to_offer, application,+ allowed_devices)+ mirabeau_tube_consumer.MirabeauTubeConsumerMixin.__init__(self,+ found_peer_callback=found_peer_callback,+ disapeared_peer_callback=disapeared_peer_callback,+ got_devices_callback=got_devices_callback)+ client.Client.__init__(self, manager, protocol, account, muc_id) - def got_devices(pontoon_devices):- self.info("%r devices registered in remote pontoon", len(pontoon_devices))- for device_dict in pontoon_devices:- device_path = device_dict["path"]- self.info("getting object at %r from %r", device_path,- initiator_bus_name)- proxy = device_tube.get_object(initiator_bus_name, device_path)- infos = proxy.get_info(dbus_interface=DEVICE_IFACE)- service_proxies = []- for service_path in device_dict["services"]:- service_proxy = service_tube.get_object(initiator_bus_name,- service_path)- service_proxies.append(service_proxy)- proxy.services = service_proxies- devices.append(proxy)- self.got_devices_callback(devices)+ def got_tube(self, tube):+ client.Client.got_tube(self, tube)+ initiator_handle = tube.props[CHANNEL_INTERFACE + ".InitiatorHandle"]+ if initiator_handle == self.self_handle:+ self.finish_tube_offer(tube)+ else:+ self.accept_tube(tube) - def got_error(exception):- print ">>>", exception-- pontoon = pontoon_tube.get_object(initiator_bus_name, OBJECT_PATH)- pontoon.get_devices_async(reply_handler=got_devices,- error_handler=got_error)+ def tube_opened(self, tube):+ tube_conn = super(MirabeauTubePublisherConsumer, self).tube_opened(tube)+ initiator_handle = tube.props[CHANNEL_INTERFACE + ".InitiatorHandle"]+ if initiator_handle == self.self_handle:+ self.post_tube_offer(tube, tube_conn)+ else:+ self.post_tube_accept(tube, tube_conn, initiator_handle)+ return tube_conn