diff --git a/CHANGELOG b/CHANGELOG index 3cd16eed77..3cd0889eca 100644 --- a/CHANGELOG +++ b/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 diff --git a/doc/ref/bus.rst b/doc/ref/bus.rst new file mode 100644 index 0000000000..c30d40a874 --- /dev/null +++ b/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 +`//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 ` + +.. 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. diff --git a/doc/topics/bus.rst b/doc/topics/bus.rst new file mode 100644 index 0000000000..12012778da --- /dev/null +++ b/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) diff --git a/doc/topics/configuration.rst b/doc/topics/configuration.rst index 1ea9bcecf0..bde4cb0be0 100644 --- a/doc/topics/configuration.rst +++ b/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 diff --git a/trytond/backend/postgresql/database.py b/trytond/backend/postgresql/database.py index 710566edb9..92dfca180a 100644 --- a/trytond/backend/postgresql/database.py +++ b/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) diff --git a/trytond/bus.py b/trytond/bus.py new file mode 100644 index 0000000000..d05f326d87 --- /dev/null +++ b/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('//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) diff --git a/trytond/config.py b/trytond/config.py index 0c7e820787..c3cefeb5fb 100644 --- a/trytond/config.py +++ b/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() diff --git a/trytond/tests/test_bus.py b/trytond/tests/test_bus.py new file mode 100644 index 0000000000..8a1dec6124 --- /dev/null +++ b/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_ diff --git a/trytond/transaction.py b/trytond/transaction.py index 6d393499e8..c6756e3f51 100644 --- a/trytond/transaction.py +++ b/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() diff --git a/trytond/wsgi.py b/trytond/wsgi.py index e92c2ef4d9..b0a32213af 100644 --- a/trytond/wsgi.py +++ b/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