Index: 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_conn Index: 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