iframe-proxy | Sunbelt Computer Software

148 lines (123 loc) · 4.91 KB
# coding: utf-8
This module contains the implementation of :class:`~can.Notifier`.
import threading
import logging
import time
import asyncio
logger = logging.getLogger("can.Notifier")
class Notifier:
    def __init__(self, bus, listeners, timeout=1.0, loop=None):
        """Manages the distribution of :class:`can.Message` instances to listeners.
        Supports multiple buses and listeners.
        .. Note::
            Remember to call `stop()` after all messages are received as
            many listeners carry out flush operations to persist data.
        :param can.BusABC bus: A :ref:`bus` or a list of buses to listen to.
        :param list listeners: An iterable of :class:`~can.Listener`
        :param float timeout: An optional maximum number of seconds to wait for any message.
        :param asyncio.AbstractEventLoop loop:
            An :mod:`asyncio` event loop to schedule listeners in.
        """
        self.listeners = listeners
        self.bus = bus
        self.timeout = timeout
        self._loop = loop
        #: Exception raised in thread
        self.exception = None
        self._running = True
        self._lock = threading.Lock()
        self._readers = []
        buses = self.bus if isinstance(self.bus, list) else [self.bus]
        for bus in buses:
            self.add_bus(bus)
    def add_bus(self, bus):
        """Add a bus for notification.
        :param can.BusABC bus:
            CAN bus instance.
        """
        if self._loop is not None and hasattr(bus, "fileno") and bus.fileno() >= 0:
            # Use file descriptor to watch for messages
            reader = bus.fileno()
            self._loop.add_reader(reader, self._on_message_available, bus)
        else:
            reader = threading.Thread(
                target=self._rx_thread,
                args=(bus,),
                name='can.notifier for bus "{}"'.format(bus.channel_info),
            reader.daemon = True
            reader.start()
        self._readers.append(reader)
    def stop(self, timeout=5):
        """Stop notifying Listeners when new :class:`~can.Message` objects arrive
        and call :meth:`~can.Listener.stop` on each Listener.
        :param float timeout:
            Max time in seconds to wait for receive threads to finish.
            Should be longer than timeout given at instantiation.
        """
        self._running = False
        end_time = time.time() + timeout
        for reader in self._readers:
            if isinstance(reader, threading.Thread):
                now = time.time()
                if now < end_time:
                    reader.join(end_time - now)
            else:
                # reader is a file descriptor
                self._loop.remove_reader(reader)
        for listener in self.listeners:
            if hasattr(listener, "stop"):
                listener.stop()
    def _rx_thread(self, bus):
        msg = None
        try:
            while self._running:
                if msg is not None:
                    with self._lock:
                        if self._loop is not None:
                            self._loop.call_soon_threadsafe(
                                self._on_message_received, msg
                            self._on_message_received(msg)
                msg = bus.recv(self.timeout)
        except Exception as exc:
            self.exception = exc
            if self._loop is not None:
                self._loop.call_soon_threadsafe(self._on_error, exc)
            else:
                self._on_error(exc)
            raise
    def _on_message_available(self, bus):
        msg = bus.recv(0)
        if msg is not None:
            self._on_message_received(msg)
    def _on_message_received(self, msg):
        for callback in self.listeners:
            res = callback(msg)
            if self._loop is not None and asyncio.iscoroutine(res):
                # Schedule coroutine
                self._loop.create_task(res)
    def _on_error(self, exc):
        for listener in self.listeners:
            if hasattr(listener, "on_error"):
                listener.on_error(exc)
    def add_listener(self, listener):
        """Add new Listener to the notification list.
        If it is already present, it will be called two times
        each time a message arrives.
        :param can.Listener listener: Listener to be added to
                         the list to be notified
        """
        self.listeners.append(listener)
    def remove_listener(self, listener):
        """Remove a listener from the notification list. This method
        trows an exception if the given listener is not part of the
        stored listeners.
        :param can.Listener listener: Listener to be removed from
                         the list to be notified
        :raises ValueError: if `listener` was never added to this notifier
        """
        self.listeners.remove(listener)
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

notifier.py

Latest commit

History

notifier.py

File metadata and controls
