diff --git a/trytond/trytond/backend/database.py b/trytond/trytond/backend/database.py index 9f5e7a3539..e9b1e15ffe 100644 --- a/trytond/trytond/backend/database.py +++ b/trytond/trytond/backend/database.py @@ -73,6 +73,12 @@ class DatabaseInterface(object): def estimated_count(self, connection, table): raise NotImplementedError + def notify(self, connection, channel, payload): + raise NotImplementedError + + def get_notifications(self, connection): + raise NotImplementedError + @classmethod def lock(cls, connection, table): raise NotImplementedError diff --git a/trytond/trytond/backend/postgresql/database.py b/trytond/trytond/backend/postgresql/database.py index ab65481852..50856ea4d2 100644 --- a/trytond/trytond/backend/postgresql/database.py +++ b/trytond/trytond/backend/postgresql/database.py @@ -492,6 +492,14 @@ class Database(DatabaseInterface): cursor.execute(*from_item.select(Count(Literal('*')))) return cursor.fetchone()[0] + def notify(self, connection, channel, payload): + cursor = connection.cursor() + cursor.execute('NOTIFY "%s", %%s' % channel, (payload,)) + + def get_notifications(self, connection): + connection.poll() + return connection.notifies + @classmethod def lock(cls, connection, table): cursor = connection.cursor() diff --git a/trytond/trytond/bus.py b/trytond/trytond/bus.py index 32c606e4e5..39c83f464a 100644 --- a/trytond/trytond/bus.py +++ b/trytond/trytond/bus.py @@ -154,9 +154,9 @@ class LongPollingBus: selector.register(conn, selectors.EVENT_READ) while cls._queues[pid, database]['timeout'] > now: selector.select(timeout=config.getint('bus', 'select_timeout')) - conn.poll() - while conn.notifies: - notification = conn.notifies.pop() + notifications = db.get_notifications(conn) + while notifications: + notification = notifications.pop() payload = json.loads( notification.payload, object_hook=JSONDecoder()) @@ -200,13 +200,13 @@ class LongPollingBus: logger.info("database backend does 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,)) + transaction.database.notify( + transaction.connection, cls._channel, payload) logger.debug( "publish %r to '%s' on '%s'", message, channel, transaction.database.name) diff --git a/trytond/trytond/cache.py b/trytond/trytond/cache.py index 978a7b6592..7d99ba0331 100644 --- a/trytond/trytond/cache.py +++ b/trytond/trytond/cache.py @@ -308,9 +308,8 @@ class MemoryCache(BaseCache): # The count computed as # 8000 (max notify size) / 64 (max name data len) for sub_reset in grouped_slice(reset, 125): - cursor.execute( - 'NOTIFY "%s", %%s' % cls._channel, - (json.dumps(list(sub_reset), separators=(',', ':')),)) + database.notify(transaction.connection, cls._channel, + json.dumps(list(sub_reset), separators=(',', ':'))) else: connection = database.get_connection( readonly=False, autocommit=True) @@ -429,9 +428,9 @@ class MemoryCache(BaseCache): while cls._local.listeners.get(dbname) == current_thread: selector.select( timeout=config.getint('cache', 'select_timeout')) - conn.poll() - while conn.notifies: - notification = conn.notifies.pop() + notifications = database.get_notifications(conn) + while notifications: + notification = notifications.pop() payload = notification.payload if payload and payload.startswith(REFRESH_POOL_MSG): remote_id = payload[len(REFRESH_POOL_MSG) + 1:] diff --git a/trytond/trytond/ir/queue_.py b/trytond/trytond/ir/queue_.py index 5dfd64c1d3..90738c6d6c 100644 --- a/trytond/trytond/ir/queue_.py +++ b/trytond/trytond/ir/queue_.py @@ -61,7 +61,6 @@ class Queue(ModelSQL): def push(cls, name, data, scheduled_at=None, expected_at=None): transaction = Transaction() database = transaction.database - cursor = transaction.connection.cursor() with without_check_access(): record, = cls.create([{ 'name': name, @@ -70,7 +69,7 @@ class Queue(ModelSQL): 'expected_at': expected_at, }]) if database.has_channel(): - cursor.execute('NOTIFY "%s"', (cls.__name__,)) + database.notify(transaction.connection, cls.__name__, '') if not config.getboolean('queue', 'worker', default=False): transaction.tasks.append(record.id) return record.id