Add bus management

issue7537
review66341002

CHANGELOG

@@ -1,3 +1,4 @@
+* Add bus system
 * Allow to set any default configuration value from environment
 * Clear existing session when password is changed
 * Manage session with max_age and timeout

doc/ref/bus.rst

@@ -0,0 +1,103 @@
+.. _ref-bus:
+.. module:: trytond.bus
+
+===
+Bus
+===
+
+.. class:: Bus
+
+The Tryton server listens on `POST` requests on the routes matching
+`/<database_name>/bus` and replies with JSON dictionary containing:
+
+channel
+    The channel on which this message has been received.
+
+message
+    A dictionary that is the message the client must handle. The specification
+    of the message depends of its type.
+
+    All messages should at least content a unique identifier in the key
+    `message_id` and their type in the key of the same name.
+
+Client sending their request on the route must be authenticated. The request
+must submit a JSON dictionary containing:
+
+last_message
+    A value identifying the last message received by the client. This value
+    can be `null`.
+
+channels
+    A list of strings denoting the channels the client is listening to.
+
+The :class:Bus exposes two methods that are used by the framework: `publish`
+and `subscribe`.
+
+.. classmethod:: Bus.publish(channel, message)
+
+    Send a message to a specific channel.
+
+    Currently implemented messages are:
+
+        * :ref:`Notifications <bus_notification_spec>`
+
+.. classmethod:: Bus.subscribe(database, channels[, last_message])
+
+    Subscribe a user client to some `channels` of messages.
+
+    The `message_id` parameter defines the last message ID received by the
+    client. It defaults to None when not provided.
+
+The default implementation provides an helper method to construct the response:
+
+.. classmethod:: Bus.create_response(channel, message)
+
+    Creates a dictionary suitable as a response from a message and a timestamp.
+
+    `channel` is the channel on which the message has been received.
+
+    `message` is the content of the message sent to the client.
+
+Please note that the implementation relies on the fact that the order of the
+messages received is consistent accros different trytond instances allowing to
+dispatch the request to any trytond server running.
+
+Notification
+------------
+
+Tryton provides a shortcut to send a notification with the `notify` method.
+
+.. method:: notify(title[, body[, priority[, user[, client]]]])
+
+    Send a text message to a user's client to be displayed using a
+    notification popup. The meaning of `title`, `body` and `priority` is
+    defined in :ref:`bus_notification_spec`.
+
+    If `user` is not set, the current :ref:`Transaction.user` is used.
+    Otherwise `user` is the user ID to notify.
+
+    If `client` is not set then every client of the user will receive the
+    message. If `client` and `user` are not set, the system will send the
+    notification to the current user client. Otherwise the notification will
+    be sent to the client whose ID matches `client`.
+
+.. _bus_notification_spec:
+
+Notification message
+~~~~~~~~~~~~~~~~~~~~
+
+Notification messages are composed of four parts:
+
+kind
+    The string 'notification'.
+
+title
+    A string. A one-line summary of the message.
+
+body
+    A string. The short informative message for the user. It can span multiple
+    lines but no markup is allowed.
+
+priority
+    An integer between 0 (low priority) to 3 (urgent). The notification priority
+    on the platform supporting it.

doc/topics/bus.rst

@@ -0,0 +1,28 @@
+.. _topics-notification:
+
+=====================
+Sending notifications
+=====================
+
+Tryton embeds a bus system allowing the system to send text messages to clients
+logged in the system.
+
+It allows the server to warn quickly the client user about some events using
+the :meth:`~notify` function. Sending the notifications is done in a
+transactional way and will occur at then end of the transaction.
+
+For example, we warn the user of low stock level when selecting a product::
+
+    from trytond.bus import notify
+
+    class SaleLine:
+        __name__ = 'sale.line'
+
+        def on_change_product(self):
+            super().on_change_product()
+
+            # compute the product current stock
+            stock = …
+
+            if stock < 0:
+                notify('Not enough stock', priority=3)

doc/topics/configuration.rst

@@ -372,6 +372,32 @@ The prefix to use with the `FileStore`.
 
 Default: `None`
 
+bus
+---
+
+long_polling_timeout
+~~~~~~~~~~~~~~~~~~~~
+
+The time in seconds to keep the connection to the client opened when using long
+polling for bus messages
+
+Default: `300`
+
+cache_timeout
+~~~~~~~~~~~~~
+
+The number of seconds a message should be kept by the queue before being
+discarded.
+
+Default: `300`
+
+select_timeout
+~~~~~~~~~~~~~~
+
+The timeout duration of the select call when listening on a channel.
+
+Default: `5`
+
 .. _JSON-RPC: http://en.wikipedia.org/wiki/JSON-RPC
 .. _XML-RPC: http://en.wikipedia.org/wiki/XML-RPC
 .. _RFC-3986: http://tools.ietf.org/html/rfc3986

trytond/backend/postgresql/database.py

@@ -41,7 +41,7 @@ __all__ = ['Database', 'DatabaseIntegrityError', 'DatabaseOperationalError']
 logger = logging.getLogger(__name__)
 
 os.environ['PGTZ'] = os.environ.get('TZ', '')
-_timeout = config.getint('database', 'timeout', default=30 * 60)
+_timeout = config.getint('database', 'timeout')
 _minconn = config.getint('database', 'minconn', default=1)
 _maxconn = config.getint('database', 'maxconn', default=64)
 

trytond/bus.py

@@ -0,0 +1,250 @@
+# This file is part of Tryton.  The COPYRIGHT file at the top level of
+# this repository contains the full copyright notices and license terms.
+
+import collections
+import json
+import logging
+import select
+import threading
+import time
+import uuid
+
+from werkzeug.wrappers import Response
+from werkzeug.exceptions import NotImplemented, BadRequest
+
+from trytond import backend
+from trytond.wsgi import app
+from trytond.transaction import Transaction
+from trytond.protocols.jsonrpc import JSONEncoder
+from trytond.config import config
+from trytond.tools import resolve
+
+
+logger = logging.getLogger(__name__)
+
+_db_timeout = config.getint('database', 'timeout')
+_cache_timeout = config.getint('bus', 'cache_timeout')
+_select_timeout = config.getint('bus', 'select_timeout')
+_long_polling_timeout = config.getint('bus', 'long_polling_timeout')
+
+
+class _MessageQueue:
+
+    Message = collections.namedtuple('Message', 'channel content timestamp')
+
+    def __init__(self, timeout):
+        super().__init__()
+        self._lock = threading.Lock()
+        self._timeout = timeout
+        self._messages = []
+
+    def append(self, channel, element):
+        self._messages.append(
+            self.Message(channel, element, time.time()))
+
+    def get_next(self, channels, from_id=None):
+        oldest = time.time() - self._timeout
+        to_delete_index = 0
+        found = False
+        first_message = None
+        message = self.Message(None, None, None)
+        for idx, item in enumerate(self._messages):
+            if item.timestamp < oldest:
+                to_delete_index = idx
+                continue
+            if item.channel not in channels:
+                continue
+            if not first_message:
+                first_message = item
+            if from_id is None or found:
+                message = item
+                break
+            found = item.content['message_id'] == from_id
+        else:
+            if first_message and not found:
+                message = first_message
+
+        with self._lock:
+            del self._messages[:to_delete_index]
+
+        return message.channel, message.content
+
+
+class LongPollingBus:
+
+    _channel = 'bus'
+    _queues_lock = threading.Lock()
+    _queues = collections.defaultdict(
+        lambda: {'timeout': None, 'events': collections.defaultdict(list)})
+    _messages = {}
+
+    @classmethod
+    def subscribe(cls, database, channels, last_message=None):
+        with cls._queues_lock:
+            start_listener = database not in cls._queues
+            cls._queues[database]['timeout'] = time.time() + _db_timeout
+            if start_listener:
+                listener = threading.Thread(
+                    target=cls._listen, args=(database,), daemon=True)
+                listener.start()
+
+        messages = cls._messages.get(database)
+        if messages:
+            channel, content = messages.get_next(channels, last_message)
+            if content:
+                return cls.create_response(channel, content)
+
+        event = threading.Event()
+        for channel in channels:
+            if channel in cls._queues[database]['events']:
+                event_channel = cls._queues[database]['events'][channel]
+            else:
+                with cls._queues_lock:
+                    event_channel = cls._queues[database]['events'][channel]
+            event_channel.append(event)
+
+        triggered = event.wait(_long_polling_timeout)
+        if not triggered:
+            response = cls.create_response(None, None)
+        else:
+            response = cls.create_response(
+                *cls._messages[database].get_next(channels, last_message))
+
+        with cls._queues_lock:
+            for channel in channels:
+                events = cls._queues[database]['events'][channel]
+                for e in events[:]:
+                    if e.is_set():
+                        events.remove(e)
+
+        return response
+
+    @classmethod
+    def create_response(cls, channel, message):
+        response_data = {
+            'message': message,
+            'channel': channel,
+            }
+        logger.debug('Bus: %s', response_data)
+        return response_data
+
+    @classmethod
+    def _listen(cls, database):
+        Database = backend.get('Database')
+        db = Database(database)
+        if not db.has_channel():
+            raise NotImplemented
+
+        logger.info("listening on channel '%s'", cls._channel)
+        conn = db.get_connection()
+        try:
+            cursor = conn.cursor()
+            cursor.execute('LISTEN "%s"' % cls._channel)
+            conn.commit()
+
+            cls._messages[database] = messages = _MessageQueue(_cache_timeout)
+
+            now = time.time()
+            while cls._queues[database]['timeout'] > now:
+                readable, _, _ = select.select([conn], [], [], _select_timeout)
+                if not readable:
+                    continue
+
+                conn.poll()
+                while conn.notifies:
+                    notification = conn.notifies.pop()
+                    payload = json.loads(notification.payload)
+                    channel = payload['channel']
+                    message = payload['message']
+                    messages.append(channel, message)
+
+                    with cls._queues_lock:
+                        events = \
+                            cls._queues[database]['events'][channel].copy()
+                        cls._queues[database]['events'][channel].clear()
+                    for event in events:
+                        event.set()
+                now = time.time()
+        except Exception:
+            logger.error('bus listener on "%s" crashed', database,
+                exc_info=True)
+
+            with cls._queues_lock:
+                del cls._queues[database]
+            raise
+        finally:
+            db.put_connection(conn)
+
+        with cls._queues_lock:
+            if cls._queues[database]['timeout'] <= now:
+                del cls._queues[database]
+            else:
+                # A query arrived between the end of the while and here
+                cls._listen(database)
+
+    @classmethod
+    def publish(cls, channel, message):
+        transaction = Transaction()
+        if not transaction.database.has_channel():
+            logger.debug('Database backend do not support channels')
+            return
+
+        cursor = transaction.connection.cursor()
+        message['message_id'] = str(uuid.uuid4())
+        payload = json.dumps({
+                'channel': channel,
+                'message': message,
+                }, cls=JSONEncoder, separators=(',', ':'))
+        cursor.execute('NOTIFY "%s", %%s' % cls._channel, (payload,))
+
+
+if config.get('bus', 'queue'):
+    Bus = resolve(config.get('bus', 'class'))
+else:
+    Bus = LongPollingBus
+
+
+@app.route('/<string:database_name>/bus', methods=['POST'])
+@app.auth_required
+def subscribe(request, database_name):
+    user = request.authorization.get('userid')
+    channels = request.parsed_data.get('channels', [])
+    if user is None:
+        raise BadRequest
+
+    channels = set(filter(lambda c: not c.startswith('user:'), channels))
+    channels.add('user:%s' % user)
+
+    last_message = request.parsed_data.get('last_message')
+
+    logger.debug(
+        "getting bus messages from %s@%s/%s for %s since %s",
+        request.authorization.username, request.remote_addr, request.path,
+        channels, last_message)
+    bus_response = Bus.subscribe(database_name, channels, last_message)
+    return Response(
+        json.dumps(bus_response, cls=JSONEncoder, separators=(',', ':')),
+        content_type='application/json')
+
+
+def notify(title, body=None, priority=1, user=None, client=None):
+    if user is None:
+        if client is None:
+            context_client = Transaction().context.get('client')
+            if context_client:
+                channel = 'client:%s' % context_client
+            else:
+                return
+        else:
+            channel = 'client:%s' % client
+    elif client is None:
+        channel = 'user:%s' % user
+    else:
+        channel = 'client:%s' % client
+
+    return Bus.publish({
+            'type': 'notification',
+            'title': title,
+            'body': body,
+            'priority': priority,
+            }, channel)

trytond/config.py

@@ -53,6 +53,7 @@ class TrytonConfigParser(configparser.RawConfigParser):
         self.set('database', 'list', 'True')
         self.set('database', 'retry', 5)
         self.set('database', 'language', 'en')
+        self.set('database', 'timeout', 30 * 60)
         self.add_section('request')
         self.set('request', 'max_size', 2 * 1024 * 1024)
         self.set('request', 'max_size_authenticated', 2 * 1024 * 1024 * 1024)
@@ -77,6 +78,10 @@ class TrytonConfigParser(configparser.RawConfigParser):
         self.set('password', 'length', 8)
         self.set('password', 'entropy', 0.75)
         self.set('password', 'reset_timeout', 24 * 60 * 60)
+        self.add_section('bus')
+        self.set('bus', 'long_polling_timeout', 5 * 60)
+        self.set('bus', 'cache_timeout', 5)
+        self.set('bus', 'select_timeout', 5)
         self.update_environ()
         self.update_etc()
 

trytond/tests/test_bus.py

@@ -0,0 +1,75 @@
+# This file is part of Tryton.  The COPYRIGHT file at the top level of
+# this repository contains the full copyright notices and license terms.
+import unittest
+from unittest.mock import patch
+
+from trytond.bus import _MessageQueue
+
+
+class MessageQueueTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self._timestamp = 0
+
+    def _time(self):
+        _timestamp = self._timestamp
+        self._timestamp += 1
+        return _timestamp
+
+    def test_get_next(self):
+        "Testing the basic functionality of get_next"
+        with patch('time.time', self._time):
+            mq = _MessageQueue(5)
+            for x in range(15):
+                mq.append('channel', {'message_id': x})
+            channel, content = mq.get_next({'channel'}, 11)
+
+        self.assertEqual(content, {'message_id': 12})
+
+    def test_get_next_channels(self):
+        "Testing get_next with multiple channels"
+        with patch('time.time', self._time):
+            mq = _MessageQueue(5)
+            for x in range(15):
+                mq.append('odd' if x % 2 else 'even', {'message_id': x})
+            channel, content = mq.get_next({'odd'}, 11)
+
+        self.assertEqual(content, {'message_id': 13})
+        self.assertEqual(channel, 'odd')
+
+    def test_get_next_timeout_expired(self):
+        "Testing get_next when requesting an outdated message"
+        with patch('time.time', self._time):
+            mq = _MessageQueue(5)
+            for x in range(15):
+                mq.append('channel', {'message_id': x})
+            channel, content = mq.get_next({'channel'}, 0)
+
+        self.assertEqual(content, {'message_id': 10})
+
+    def test_get_next_message_id_missing(self):
+        "Testing get_next when requesting a missing message"
+        with patch('time.time', self._time):
+            mq = _MessageQueue(5)
+            for x in range(15):
+                mq.append('channel', {'message_id': x})
+            channel, content = mq.get_next({'channel'}, -5)
+
+        self.assertEqual(content, {'message_id': 10})
+
+    def test_get_next_message_id_None(self):
+        "Testing get_next when not specifying a message"
+        with patch('time.time', self._time):
+            mq = _MessageQueue(5)
+            for x in range(15):
+                mq.append('channel', {'message_id': x})
+            channel, content = mq.get_next({'channel'})
+
+        self.assertEqual(content, {'message_id': 10})
+
+
+def suite():
+    suite_ = unittest.TestSuite()
+    suite_.addTests(unittest.TestLoader().loadTestsFromTestCase(
+            MessageQueueTestCase))
+    return suite_

trytond/transaction.py

@@ -127,7 +127,9 @@ class Transaction(object):
             if transactions.count(self) == 1:
                 try:
                     try:
-                        if commit and not self.readonly:
+                        # Transaction must be commited to send notifications
+                        if commit and (not self.readonly
+                                or self.database.has_channel()):
                             self.commit()
                         else:
                             self.rollback()

trytond/wsgi.py

@@ -152,3 +152,4 @@ num_proxies = config.getint('web', 'num_proxies')
 if num_proxies:
     app.wsgi_app = ProxyFix(app.wsgi_app, num_proxies=num_proxies)
 import trytond.protocols.dispatcher
+import trytond.bus