1797

pending completion

Build # 1797

Build Type

push

travis-ci-com

Committed by

pwt

Commit Message

<a href="https://github.com/SoCo/SoCo/commit/<a class=hub.com/SoCo/SoCo/commit/7f1048227ccba7ee3c90ebfb00ebc60b7561aa8d">7f1048227<a href="https://github.com/SoCo/SoCo/commit/7f1048227ccba7ee3c90ebfb00ebc60b7561aa8d">&quot;&gt;Revert &amp;quot;Fix event unsubscription logic (#841)&quot;

This reverts commit </a><a class="double-link" href="https://github.com/SoCo/SoCo/commit/<a class="double-link" href="https://github.com/SoCo/SoCo/commit/c9dcf765e0eabbefd3ca6cfd43bdaeb871c3fcbd">c9dcf765e</a>">c9dcf765e</a><a href="https://github.com/SoCo/SoCo/commit/7f1048227ccba7ee3c90ebfb00ebc60b7561aa8d">.

Run Details

436 of 1074 branches covered (40.6%)

4 of 4 new or added lines in 3 files covered. (100.0%)

2061 of 3836 relevant lines covered (53.73%)

4.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/soco/events_asyncio.py

"""Classes to handle Sonos UPnP Events and Subscriptions using asyncio.

The `Subscription` class from this module will be used in
:py:mod:`soco.services` if `config.EVENTS_MODULE` is set
to point to this module.

Example:

    Run this code, and change your volume, tracks etc::

        import logging

        logging.basicConfig()
        import soco
        import asyncio
        from pprint import pprint

        from soco import events_asyncio

        soco.config.EVENTS_MODULE = events_asyncio


        def print_event(event):
            try:
                pprint(event.variables)
            except Exception as e:
                print("There was an error in print_event:", e)


        def _get_device():
            device = soco.discover().pop().group.coordinator
            print(device.player_name)
            return device


        async def main():
            # pick a device at random and use it to get
            # the group coordinator
            loop = asyncio.get_event_loop()
            device = await loop.run_in_executor(None, _get_device)
            sub = await device.renderingControl.subscribe()
            sub2 = await device.avTransport.subscribe()
            sub.callback = print_event
            sub2.callback = print_event

            async def before_shutdown():
                await sub.unsubscribe()
                await sub2.unsubscribe()
                await events_asyncio.event_listener.async_stop()

            await asyncio.sleep(1)
            print("Renewing subscription..")
            await sub.renew()

            await asyncio.sleep(100)
            await before_shutdown()


        if __name__ == "__main__":
            asyncio.run(main())

"""

import logging
import socket
import sys
import time
import asyncio

try:
    from aiohttp import ClientSession, web
except ImportError as error:
    print(
        """ImportError: {}:
    Use of the SoCo events_asyncio module requires the 'aiohttp'
    package and its dependencies to be installed. aiohttp is not
    installed with SoCo by default due to potential issues installing
    the dependencies 'mutlidict' and 'yarl' on some platforms.
    See: https://github.com/SoCo/SoCo/issues/819""".format(
            error
        )
    )
    sys.exit(1)

# Event is imported for compatibility with events.py
# pylint: disable=unused-import
from .events_base import Event  # noqa: F401

from .events_base import (  # noqa: E402
    get_listen_ip,
    parse_event_xml,
    EventNotifyHandlerBase,
    EventListenerBase,
    SubscriptionBase,
    SubscriptionsMap,
)

from .exceptions import SoCoException  # noqa: E402

log = logging.getLogger(__name__)  # pylint: disable=C0103


class EventNotifyHandler(EventNotifyHandlerBase):
    """Handles HTTP ``NOTIFY`` Verbs sent to the listener server.
    Inherits from `soco.events_base.EventNotifyHandlerBase`.
    """

    def __init__(self):
        super().__init__()
        # The SubscriptionsMapAio instance created when this module is
        # imported. This is referenced by
        # soco.events_base.EventNotifyHandlerBase.
        self.subscriptions_map = subscriptions_map

    async def notify(self, request):
        """Serve a ``NOTIFY`` request by calling `handle_notification`
        with the headers and content.
        """
        content = await request.text()
        seq = request.headers["seq"]  # Event sequence number
        sid = request.headers["sid"]  # Event Subscription Identifier
        # find the relevant service from the sid
        # pylint: disable=no-member
        subscription = self.subscriptions_map.get_subscription(sid)
        # It might have been removed by another thread
        if subscription:
            timestamp = time.time()
            service = subscription.service
            self.log_event(seq, service.service_id, timestamp)
            log.debug("Event content: %s", content)
            if "x-sonos-http" in content:
                # parse_event_xml will generate I/O if
                # x-sonos-http is in the content
                variables = await asyncio.get_event_loop().run_in_executor(
                    None, parse_event_xml, content
                )
            else:
                variables = parse_event_xml(content)

            # Build the Event object
            event = Event(sid, seq, service, timestamp, variables)
            # pass the event details on to the service so it can update
            # its cache.
            # pylint: disable=protected-access
            service._update_cache_on_event(event)
            # Pass the event on for handling
            subscription.send_event(event)
        else:
            log.debug("No service registered for %s", sid)

        return web.Response(text="OK", status=200)

    # pylint: disable=no-self-use, missing-docstring
    def log_event(self, seq, service_id, timestamp):
        log.debug("Event %s received for %s service at %s", seq, service_id, timestamp)


class EventListener(EventListenerBase):  # pylint: disable=too-many-instance-attributes
    """The Event Listener.

    Runs an http server which is an endpoint for ``NOTIFY``
    requests from Sonos devices. Inherits from
    `soco.events_base.EventListenerBase`.
    """

    def __init__(self):
        super().__init__()
        self.sock = None
        self.ip_address = None
        self.port = None
        self.runner = None
        self.site = None
        self.session = None
        self.start_lock = None

    def start(self, any_zone):
        """A stub since the first subscribe calls async_start."""
        return

    def listen(self, ip_address):
        """A stub since since async_listen is used."""
        return

    async def async_start(self, any_zone):
        """Start the event listener listening on the local machine under the lock.

        Args:
            any_zone (SoCo): Any Sonos device on the network. It does not
                matter which device. It is used only to find a local IP
                address reachable by the Sonos net.

        """
        if not self.start_lock:
            self.start_lock = asyncio.Lock()
        async with self.start_lock:
            if self.is_running:
                return
            # Use configured IP address if there is one, else detect
            # automatically.
            ip_address = get_listen_ip(any_zone.ip_address)
            if not ip_address:
                log.exception("Could not start Event Listener: check network.")
                # Otherwise, no point trying to start server
                return
            port = await self.async_listen(ip_address)
            if not port:
                return
            self.address = (ip_address, port)
            self.session = ClientSession(raise_for_status=True)
            self.is_running = True
            log.debug("Event Listener started")

    async def async_listen(self, ip_address):
        """Start the event listener listening on the local machine at
        port 1400 (default). If this port is unavailable, the
        listener will attempt to listen on the next available port,
        within a range of 100.

        Make sure that your firewall allows connections to this port.

        This method is called by `soco.events_base.EventListenerBase.start`

        Handling of requests is delegated to an instance of the
        `EventNotifyHandler` class.

        Args:
            ip_address (str): The local network interface on which the server
                should start listening.
        Returns:
            int: The port on which the server is listening.

        Note:
            The port on which the event listener listens is configurable.
            See `config.EVENT_LISTENER_PORT`
        """
        for port_number in range(
            self.requested_port_number, self.requested_port_number + 100
        ):
            try:
                if port_number > self.requested_port_number:
                    log.debug("Trying next port (%d)", port_number)
                # pylint: disable=no-member
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.bind((ip_address, port_number))
                sock.listen(200)
                self.sock = sock
                self.port = port_number
                break
            # pylint: disable=invalid-name
            except socket.error as e:
                log.warning("Could not bind to %s:%s: %s", ip_address, port_number, e)
                continue

        if not self.port:
            return None

        self.ip_address = ip_address
        await self._async_start()
        return self.port

    async def _async_start(self):
        """Start the site."""
        handler = EventNotifyHandler()
        app = web.Application()
        app.add_routes([web.route("notify", "", handler.notify)])
        self.runner = web.AppRunner(app)
        await self.runner.setup()
        self.site = web.SockSite(self.runner, self.sock)
        await self.site.start()
        log.debug("Event listener running on %s", (self.ip_address, self.port))

    async def async_stop(self):
        """Stop the listener."""
        if self.site:
            await self.site.stop()
            self.site = None
        if self.runner:
            await self.runner.cleanup()
            self.runner = None
        if self.session:
            await self.session.close()
            self.session = None
        if self.sock:
            self.sock.close()
            self.sock = None
        self.port = None
        self.ip_address = None

    # pylint: disable=unused-argument
    def stop_listening(self, address):
        """Stop the listener."""
        asyncio.ensure_future(self.async_stop())


class Subscription(SubscriptionBase):
    """A class representing the subscription to a UPnP event.
    Inherits from `soco.events_base.SubscriptionBase`.
    """

    def __init__(self, service, callback=None):
        """
        Args:
            service (Service): The SoCo `Service` to which the subscription
                 should be made.
            event_queue (:class:`~queue.Queue`): A queue on which received
                events will be put. If not specified, a queue will be
                created and used.

        """
        super().__init__(service, None)
        #: :py:obj:`function`: callback function to be called whenever an
        #: `Event` is received. If it is set and is callable, the callback
        #: function will be called with the `Event` as the only parameter and
        #: the Subscription's event queue won't be used.
        self.callback = callback
        # The SubscriptionsMapAio instance created when this module is
        # imported. This is referenced by soco.events_base.SubscriptionBase.
        self.subscriptions_map = subscriptions_map
        # The EventListener instance created when this module is imported.
        # This is referenced by soco.events_base.SubscriptionBase.
        self.event_listener = event_listener
        # Used to keep track of the auto_renew loop
        self._auto_renew_task = None

    # pylint: disable=arguments-differ
    def subscribe(self, requested_timeout=None, auto_renew=False, strict=True):
        """Subscribe to the service.

        If requested_timeout is provided, a subscription valid for that number
        of seconds will be requested, but not guaranteed. Check
        `timeout` on return to find out what period of validity is
        actually allocated.

        This method calls `events_base.SubscriptionBase.subscribe`.

        Note:
            SoCo will try to unsubscribe any subscriptions which are still
            subscribed on program termination, but it is good practice for
            you to clean up by making sure that you call :meth:`unsubscribe`
            yourself.

        Args:
            requested_timeout(int, optional): The timeout to be requested.
            auto_renew (bool, optional): If `True`, renew the subscription
                automatically shortly before timeout. Default `False`.
            strict (bool, optional): If True and an Exception occurs during
                execution, the Exception will be raised or, if False, the
                Exception will be logged and the Subscription instance will be
                returned. Default `True`.

        Returns:
            `Subscription`: The Subscription instance.

        """
        self.subscriptions_map.subscribing()
        future = asyncio.Future()
        subscribe = super().subscribe

        async def _async_wrap_subscribe():
            try:
                if not self.event_listener.is_running:
                    await self.event_listener.async_start(self.service.soco)
                await subscribe(requested_timeout, auto_renew)
                future.set_result(self)
            except SoCoException as ex:
                future.set_exception(ex)
            except Exception as exc:  # pylint: disable=broad-except
                self._cancel_subscription(exc)
                if strict:
                    future.set_exception(exc)
                else:
                    self._log_exception(exc)
                    future.set_result(self)
            finally:
                self.subscriptions_map.finished_subscribing()

        asyncio.ensure_future(_async_wrap_subscribe())
        return future

    def _log_exception(self, exc):
        """Log an exception during subscription."""
        msg = (
            "An Exception occurred: {}.".format(exc)
            + " Subscription to {},".format(
                self.service.base_url + self.service.event_subscription_url
            )
            + " sid: {} has been cancelled".format(self.sid)
        )
        log.exception(msg)

    async def renew(
        self, requested_timeout=None, is_autorenew=False, strict=True
    ):  # pylint: disable=invalid-overridden-method
        """renew(requested_timeout=None)
        Renew the event subscription.
        You should not try to renew a subscription which has been
        unsubscribed, or once it has expired.

        This method calls `events_base.SubscriptionBase.renew`.

        Args:
            requested_timeout (int, optional): The period for which a renewal
                request should be made. If None (the default), use the timeout
                requested on subscription.
            is_autorenew (bool, optional): Whether this is an autorenewal.
                Default `False`.
            strict (bool, optional): If True and an Exception occurs during
                execution, the Exception will be raised or, if False, the
                Exception will be logged and the Subscription instance will be
                returned. Default `True`.

        Returns:
            `Subscription`: The Subscription instance.

        """
        try:
            return await super().renew(requested_timeout, is_autorenew)
        except Exception as exc:  # pylint: disable=broad-except
            self._cancel_subscription(exc)
            if self.auto_renew_fail is not None and hasattr(
                self.auto_renew_fail, "__call__"
            ):
                # pylint: disable=not-callable
                self.auto_renew_fail(exc)
            else:
                self._log_exception(exc)
            if strict:
                raise
            return self

    async def unsubscribe(
        self, strict=True
    ):  # pylint: disable=invalid-overridden-method
        """unsubscribe()
        Unsubscribe from the service's events.
        Once unsubscribed, a Subscription instance should not be reused

        This method calls `events_base.SubscriptionBase.unsubscribe`.

        Args:
            strict (bool, optional): If True and an Exception occurs during
                execution, the Exception will be raised or, if False, the
                Exception will be logged and the Subscription instance will be
                returned. Default `True`.

        Returns:
            `Subscription`: The Subscription instance.
        """
        try:
            unsub = super().unsubscribe()
            if unsub is None:
                return
            await unsub
        except Exception as exc:  # pylint: disable=broad-except
            if strict:
                raise
            self._log_exception(exc)
            return self

    def _auto_renew_start(self, interval):
        """Starts the auto_renew loop."""
        self._auto_renew_task = asyncio.get_event_loop().call_later(
            interval, self._auto_renew_run, interval
        )

    def _auto_renew_run(self, interval):
        asyncio.ensure_future(self.renew(is_autorenew=True, strict=False))
        self._auto_renew_start(interval)

    def _auto_renew_cancel(self):
        """Cancels the auto_renew loop"""
        if self._auto_renew_task:
            self._auto_renew_task.cancel()
            self._auto_renew_task = None

    # pylint: disable=no-self-use, too-many-branches, too-many-arguments
    def _request(self, method, url, headers, success):
        """Sends an HTTP request.

        Args:
            method (str): 'SUBSCRIBE' or 'UNSUBSCRIBE'.
            url (str): The full endpoint to which the request is being sent.
            headers (dict): A dict of headers, each key and each value being
                of type `str`.
            success (function): A function to be called if the
                request succeeds. The function will be called with a dict
                of response headers as its only parameter.

        """

        async def _async_make_request():
            response = await self.event_listener.session.request(
                method, url, headers=headers
            )
            if response.ok:
                success(response.headers)

        return _async_make_request()


class nullcontext:  # pylint: disable=invalid-name
    """Context manager that does no additional processing.

    Backport from python 3.7+ for older pythons.
    """

    def __init__(self, enter_result=None):
        self.enter_result = enter_result

    def __enter__(self):
        return self.enter_result

    def __exit__(self, *excinfo):
        pass


class SubscriptionsMapAio(SubscriptionsMap):
    """Maintains a mapping of sids to `soco.events_asyncio.Subscription`
    instances. Registers each subscription to be unsubscribed at exit.

    Inherits from `soco.events_base.SubscriptionsMap`.
    """

    def __init__(self):
        super().__init__()
        # A counter of calls to Subscription.subscribe
        # that have started but not completed. This is
        # to prevent the event listener from being stopped prematurely
        self._pending = 0
        self.subscriptions_lock = nullcontext()

    def register(self, subscription):
        """Register a subscription by updating local mapping of sid to
        subscription and registering it to be unsubscribed at exit.

        Args:
            subscription(`soco.events_asyncio.Subscription`): the subscription
                to be registered.

        """

        # Add the subscription to the local dict of subscriptions so it
        # can be looked up by sid
        self.subscriptions[subscription.sid] = subscription

    def subscribing(self):
        """Called when the `Subscription.subscribe` method
        commences execution.
        """
        # Increment the counter
        self._pending += 1

    def finished_subscribing(self):
        """Called when the `Subscription.subscribe` method
        completes execution.
        """
        # Decrement the counter
        self._pending -= 1

    @property
    def count(self):
        """
        `int`: The number of active or pending subscriptions.
        """
        return len(self.subscriptions) + self._pending


subscriptions_map = SubscriptionsMapAio()  # pylint: disable=C0103
event_listener = EventListener()  # pylint: disable=C0103

1	"""Classes to handle Sonos UPnP Events and Subscriptions using asyncio.
2
3	The `Subscription` class from this module will be used in
4	:py:mod:`soco.services` if `config.EVENTS_MODULE` is set
5	to point to this module.
6
7	Example:
8
9	Run this code, and change your volume, tracks etc::
10
11	import logging
12
13	logging.basicConfig()
14	import soco
15	import asyncio
16	from pprint import pprint
17
18	from soco import events_asyncio
19
20	soco.config.EVENTS_MODULE = events_asyncio
21
22
23	def print_event(event):
24	try:
25	pprint(event.variables)
26	except Exception as e:
27	print("There was an error in print_event:", e)
28
29
30	def _get_device():
31	device = soco.discover().pop().group.coordinator
32	print(device.player_name)
33	return device
34
35
36	async def main():
37	# pick a device at random and use it to get
38	# the group coordinator
39	loop = asyncio.get_event_loop()
40	device = await loop.run_in_executor(None, _get_device)
41	sub = await device.renderingControl.subscribe()
42	sub2 = await device.avTransport.subscribe()
43	sub.callback = print_event
44	sub2.callback = print_event
45
46	async def before_shutdown():
47	await sub.unsubscribe()
48	await sub2.unsubscribe()
49	await events_asyncio.event_listener.async_stop()
50
51	await asyncio.sleep(1)
52	print("Renewing subscription..")
53	await sub.renew()
54
55	await asyncio.sleep(100)
56	await before_shutdown()
57
58
59	if __name__ == "__main__":
60	asyncio.run(main())
61
62	"""
63
64	import logging	×
65	import socket	×
66	import sys	×
67	import time	×
68	import asyncio	×
69
70	try:	×
71	from aiohttp import ClientSession, web	×
72	except ImportError as error:	×
73	print(	×
74	"""ImportError: {}:
75	Use of the SoCo events_asyncio module requires the 'aiohttp'
76	package and its dependencies to be installed. aiohttp is not
77	installed with SoCo by default due to potential issues installing
78	the dependencies 'mutlidict' and 'yarl' on some platforms.
79	See: https://github.com/SoCo/SoCo/issues/819""".format(
80	error
81	)
82	)
83	sys.exit(1)	×
84
85	# Event is imported for compatibility with events.py
86	# pylint: disable=unused-import
87	from .events_base import Event # noqa: F401	×
88
89	from .events_base import ( # noqa: E402	×
90	get_listen_ip,
91	parse_event_xml,
92	EventNotifyHandlerBase,
93	EventListenerBase,
94	SubscriptionBase,
95	SubscriptionsMap,
96	)
97
98	from .exceptions import SoCoException # noqa: E402	×
99
100	log = logging.getLogger(__name__) # pylint: disable=C0103	×
101
102
103	class EventNotifyHandler(EventNotifyHandlerBase):	×
104	"""Handles HTTP ``NOTIFY`` Verbs sent to the listener server.
105	Inherits from `soco.events_base.EventNotifyHandlerBase`.
106	"""
107
108	def __init__(self):	×
109	super().__init__()	×
110	# The SubscriptionsMapAio instance created when this module is
111	# imported. This is referenced by
112	# soco.events_base.EventNotifyHandlerBase.
113	self.subscriptions_map = subscriptions_map	×
114
115	async def notify(self, request):	×
116	"""Serve a ``NOTIFY`` request by calling `handle_notification`
117	with the headers and content.
118	"""
119	content = await request.text()	×
120	seq = request.headers["seq"] # Event sequence number	×
121	sid = request.headers["sid"] # Event Subscription Identifier	×
122	# find the relevant service from the sid
123	# pylint: disable=no-member
124	subscription = self.subscriptions_map.get_subscription(sid)	×
125	# It might have been removed by another thread
126	if subscription:	×
127	timestamp = time.time()	×
128	service = subscription.service	×
129	self.log_event(seq, service.service_id, timestamp)	×
130	log.debug("Event content: %s", content)	×
131	if "x-sonos-http" in content:	×
132	# parse_event_xml will generate I/O if
133	# x-sonos-http is in the content
134	variables = await asyncio.get_event_loop().run_in_executor(	×
135	None, parse_event_xml, content
136	)
137	else:
138	variables = parse_event_xml(content)	×
139
140	# Build the Event object
141	event = Event(sid, seq, service, timestamp, variables)	×
142	# pass the event details on to the service so it can update
143	# its cache.
144	# pylint: disable=protected-access
145	service._update_cache_on_event(event)	×
146	# Pass the event on for handling
147	subscription.send_event(event)	×
148	else:
149	log.debug("No service registered for %s", sid)	×
150
151	return web.Response(text="OK", status=200)	×
152
153	# pylint: disable=no-self-use, missing-docstring
154	def log_event(self, seq, service_id, timestamp):	×
155	log.debug("Event %s received for %s service at %s", seq, service_id, timestamp)	×
156
157
158	class EventListener(EventListenerBase): # pylint: disable=too-many-instance-attributes	×
159	"""The Event Listener.
160
161	Runs an http server which is an endpoint for ``NOTIFY``
162	requests from Sonos devices. Inherits from
163	`soco.events_base.EventListenerBase`.
164	"""
165
166	def __init__(self):	×
167	super().__init__()	×
168	self.sock = None	×
169	self.ip_address = None	×
170	self.port = None	×
171	self.runner = None	×
172	self.site = None	×
173	self.session = None	×
174	self.start_lock = None	×
175
176	def start(self, any_zone):	×
177	"""A stub since the first subscribe calls async_start."""
178	return	×
179
180	def listen(self, ip_address):	×
181	"""A stub since since async_listen is used."""
182	return	×
183
184	async def async_start(self, any_zone):	×
185	"""Start the event listener listening on the local machine under the lock.
186
187	Args:
188	any_zone (SoCo): Any Sonos device on the network. It does not
189	matter which device. It is used only to find a local IP
190	address reachable by the Sonos net.
191
192	"""
193	if not self.start_lock:	×
194	self.start_lock = asyncio.Lock()	×
195	async with self.start_lock:	×
196	if self.is_running:	×
197	return	×
198	# Use configured IP address if there is one, else detect
199	# automatically.
200	ip_address = get_listen_ip(any_zone.ip_address)	×
201	if not ip_address:	×
202	log.exception("Could not start Event Listener: check network.")	×
203	# Otherwise, no point trying to start server
204	return	×
205	port = await self.async_listen(ip_address)	×
206	if not port:	×
207	return	×
208	self.address = (ip_address, port)	×
209	self.session = ClientSession(raise_for_status=True)	×
210	self.is_running = True	×
211	log.debug("Event Listener started")	×
212
213	async def async_listen(self, ip_address):	×
214	"""Start the event listener listening on the local machine at
215	port 1400 (default). If this port is unavailable, the
216	listener will attempt to listen on the next available port,
217	within a range of 100.
218
219	Make sure that your firewall allows connections to this port.
220
221	This method is called by `soco.events_base.EventListenerBase.start`
222
223	Handling of requests is delegated to an instance of the
224	`EventNotifyHandler` class.
225
226	Args:
227	ip_address (str): The local network interface on which the server
228	should start listening.
229	Returns:
230	int: The port on which the server is listening.
231
232	Note:
233	The port on which the event listener listens is configurable.
234	See `config.EVENT_LISTENER_PORT`
235	"""
236	for port_number in range(	×
237	self.requested_port_number, self.requested_port_number + 100
238	):
239	try:	×
240	if port_number > self.requested_port_number:	×
241	log.debug("Trying next port (%d)", port_number)	×
242	# pylint: disable=no-member
243	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)	×
244	sock.bind((ip_address, port_number))	×
245	sock.listen(200)	×
246	self.sock = sock	×
247	self.port = port_number	×
248	break	×
249	# pylint: disable=invalid-name
250	except socket.error as e:	×
251	log.warning("Could not bind to %s:%s: %s", ip_address, port_number, e)	×
252	continue	×
253
254	if not self.port:	×
255	return None	×
256
257	self.ip_address = ip_address	×
258	await self._async_start()	×
259	return self.port	×
260
261	async def _async_start(self):	×
262	"""Start the site."""
263	handler = EventNotifyHandler()	×
264	app = web.Application()	×
265	app.add_routes([web.route("notify", "", handler.notify)])	×
266	self.runner = web.AppRunner(app)	×
267	await self.runner.setup()	×
268	self.site = web.SockSite(self.runner, self.sock)	×
269	await self.site.start()	×
270	log.debug("Event listener running on %s", (self.ip_address, self.port))	×
271
272	async def async_stop(self):	×
273	"""Stop the listener."""
274	if self.site:	×
275	await self.site.stop()	×
276	self.site = None	×
277	if self.runner:	×
278	await self.runner.cleanup()	×
279	self.runner = None	×
280	if self.session:	×
281	await self.session.close()	×
282	self.session = None	×
283	if self.sock:	×
284	self.sock.close()	×
285	self.sock = None	×
286	self.port = None	×
287	self.ip_address = None	×
288
289	# pylint: disable=unused-argument
290	def stop_listening(self, address):	×
291	"""Stop the listener."""
292	asyncio.ensure_future(self.async_stop())	×
293
294
295	class Subscription(SubscriptionBase):	×
296	"""A class representing the subscription to a UPnP event.
297	Inherits from `soco.events_base.SubscriptionBase`.
298	"""
299
300	def __init__(self, service, callback=None):	×
301	"""
302	Args:
303	service (Service): The SoCo `Service` to which the subscription
304	should be made.
305	event_queue (:class:`~queue.Queue`): A queue on which received
306	events will be put. If not specified, a queue will be
307	created and used.
308
309	"""
310	super().__init__(service, None)	×
311	#: :py:obj:`function`: callback function to be called whenever an
312	#: `Event` is received. If it is set and is callable, the callback
313	#: function will be called with the `Event` as the only parameter and
314	#: the Subscription's event queue won't be used.
315	self.callback = callback	×
316	# The SubscriptionsMapAio instance created when this module is
317	# imported. This is referenced by soco.events_base.SubscriptionBase.
318	self.subscriptions_map = subscriptions_map	×
319	# The EventListener instance created when this module is imported.
320	# This is referenced by soco.events_base.SubscriptionBase.
321	self.event_listener = event_listener	×
322	# Used to keep track of the auto_renew loop
323	self._auto_renew_task = None	×
324
325	# pylint: disable=arguments-differ
326	def subscribe(self, requested_timeout=None, auto_renew=False, strict=True):	×
327	"""Subscribe to the service.
328
329	If requested_timeout is provided, a subscription valid for that number
330	of seconds will be requested, but not guaranteed. Check
331	`timeout` on return to find out what period of validity is
332	actually allocated.
333
334	This method calls `events_base.SubscriptionBase.subscribe`.
335
336	Note:
337	SoCo will try to unsubscribe any subscriptions which are still
338	subscribed on program termination, but it is good practice for
339	you to clean up by making sure that you call :meth:`unsubscribe`
340	yourself.
341
342	Args:
343	requested_timeout(int, optional): The timeout to be requested.
344	auto_renew (bool, optional): If `True`, renew the subscription
345	automatically shortly before timeout. Default `False`.
346	strict (bool, optional): If True and an Exception occurs during
347	execution, the Exception will be raised or, if False, the
348	Exception will be logged and the Subscription instance will be
349	returned. Default `True`.
350
351	Returns:
352	`Subscription`: The Subscription instance.
353
354	"""
355	self.subscriptions_map.subscribing()	×
356	future = asyncio.Future()	×
357	subscribe = super().subscribe	×
358
359	async def _async_wrap_subscribe():	×
360	try:	×
361	if not self.event_listener.is_running:	×
362	await self.event_listener.async_start(self.service.soco)	×
363	await subscribe(requested_timeout, auto_renew)	×
364	future.set_result(self)	×
365	except SoCoException as ex:	×
366	future.set_exception(ex)	×
367	except Exception as exc: # pylint: disable=broad-except	×
368	self._cancel_subscription(exc)	×
369	if strict:	×
370	future.set_exception(exc)	×
371	else:
372	self._log_exception(exc)	×
373	future.set_result(self)	×
374	finally:
375	self.subscriptions_map.finished_subscribing()	×
376
377	asyncio.ensure_future(_async_wrap_subscribe())	×
378	return future	×
379
380	def _log_exception(self, exc):	×
381	"""Log an exception during subscription."""
382	msg = (	×
383	"An Exception occurred: {}.".format(exc)
384	+ " Subscription to {},".format(
385	self.service.base_url + self.service.event_subscription_url
386	)
387	+ " sid: {} has been cancelled".format(self.sid)
388	)
389	log.exception(msg)	×
390
391	async def renew(	×
392	self, requested_timeout=None, is_autorenew=False, strict=True
393	): # pylint: disable=invalid-overridden-method
394	"""renew(requested_timeout=None)
395	Renew the event subscription.
396	You should not try to renew a subscription which has been
397	unsubscribed, or once it has expired.
398
399	This method calls `events_base.SubscriptionBase.renew`.
400
401	Args:
402	requested_timeout (int, optional): The period for which a renewal
403	request should be made. If None (the default), use the timeout
404	requested on subscription.
405	is_autorenew (bool, optional): Whether this is an autorenewal.
406	Default `False`.
407	strict (bool, optional): If True and an Exception occurs during
408	execution, the Exception will be raised or, if False, the
409	Exception will be logged and the Subscription instance will be
410	returned. Default `True`.
411
412	Returns:
413	`Subscription`: The Subscription instance.
414
415	"""
416	try:	×
417	return await super().renew(requested_timeout, is_autorenew)	×
418	except Exception as exc: # pylint: disable=broad-except	×
419	self._cancel_subscription(exc)	×
420	if self.auto_renew_fail is not None and hasattr(	×
421	self.auto_renew_fail, "__call__"
422	):
423	# pylint: disable=not-callable
424	self.auto_renew_fail(exc)	×
425	else:
426	self._log_exception(exc)	×
427	if strict:	×
428	raise	×
429	return self	×
430
431	async def unsubscribe(	×
432	self, strict=True
433	): # pylint: disable=invalid-overridden-method
434	"""unsubscribe()
435	Unsubscribe from the service's events.
436	Once unsubscribed, a Subscription instance should not be reused
437
438	This method calls `events_base.SubscriptionBase.unsubscribe`.
439
440	Args:
441	strict (bool, optional): If True and an Exception occurs during
442	execution, the Exception will be raised or, if False, the
443	Exception will be logged and the Subscription instance will be
444	returned. Default `True`.
445
446	Returns:
447	`Subscription`: The Subscription instance.
448	"""
449	try:	×
450	unsub = super().unsubscribe()	×
451	if unsub is None:	×
452	return	×
453	await unsub	×
454	except Exception as exc: # pylint: disable=broad-except	×
455	if strict:	×
456	raise	×
457	self._log_exception(exc)	×
458	return self	×
459
460	def _auto_renew_start(self, interval):	×
461	"""Starts the auto_renew loop."""
462	self._auto_renew_task = asyncio.get_event_loop().call_later(	×
463	interval, self._auto_renew_run, interval
464	)
465
466	def _auto_renew_run(self, interval):	×
467	asyncio.ensure_future(self.renew(is_autorenew=True, strict=False))	×
468	self._auto_renew_start(interval)	×
469
470	def _auto_renew_cancel(self):	×
471	"""Cancels the auto_renew loop"""
472	if self._auto_renew_task:	×
473	self._auto_renew_task.cancel()	×
474	self._auto_renew_task = None	×
475
476	# pylint: disable=no-self-use, too-many-branches, too-many-arguments
477	def _request(self, method, url, headers, success):	×
478	"""Sends an HTTP request.
479
480	Args:
481	method (str): 'SUBSCRIBE' or 'UNSUBSCRIBE'.
482	url (str): The full endpoint to which the request is being sent.
483	headers (dict): A dict of headers, each key and each value being
484	of type `str`.
485	success (function): A function to be called if the
486	request succeeds. The function will be called with a dict
487	of response headers as its only parameter.
488
489	"""
490
491	async def _async_make_request():	×
492	response = await self.event_listener.session.request(	×
493	method, url, headers=headers
494	)
495	if response.ok:	×
496	success(response.headers)	×
497
498	return _async_make_request()	×
499
500
501	class nullcontext: # pylint: disable=invalid-name	×
502	"""Context manager that does no additional processing.
503
504	Backport from python 3.7+ for older pythons.
505	"""
506
507	def __init__(self, enter_result=None):	×
508	self.enter_result = enter_result	×
509
510	def __enter__(self):	×
511	return self.enter_result	×
512
513	def __exit__(self, *excinfo):	×
514	pass	×
515
516
517	class SubscriptionsMapAio(SubscriptionsMap):	×
518	"""Maintains a mapping of sids to `soco.events_asyncio.Subscription`
519	instances. Registers each subscription to be unsubscribed at exit.
520
521	Inherits from `soco.events_base.SubscriptionsMap`.
522	"""
523
524	def __init__(self):	×
525	super().__init__()	×
526	# A counter of calls to Subscription.subscribe
527	# that have started but not completed. This is
528	# to prevent the event listener from being stopped prematurely
529	self._pending = 0	×
530	self.subscriptions_lock = nullcontext()	×
531
532	def register(self, subscription):	×
533	"""Register a subscription by updating local mapping of sid to
534	subscription and registering it to be unsubscribed at exit.
535
536	Args:
537	subscription(`soco.events_asyncio.Subscription`): the subscription
538	to be registered.
539
540	"""
541
542	# Add the subscription to the local dict of subscriptions so it
543	# can be looked up by sid
544	self.subscriptions[subscription.sid] = subscription	×
545
546	def subscribing(self):	×
547	"""Called when the `Subscription.subscribe` method
548	commences execution.
549	"""
550	# Increment the counter
551	self._pending += 1	×
552
553	def finished_subscribing(self):	×
554	"""Called when the `Subscription.subscribe` method
555	completes execution.
556	"""
557	# Decrement the counter
558	self._pending -= 1	×
559
560	@property	×
561	def count(self):
562	"""
563	`int`: The number of active or pending subscriptions.
564	"""
565	return len(self.subscriptions) + self._pending	×
566
567
568	subscriptions_map = SubscriptionsMapAio() # pylint: disable=C0103	×
569	event_listener = EventListener() # pylint: disable=C0103	×

SoCo / SoCo / 1797

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous