Delegate the notifications handling to the backend

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

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()

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)

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:]

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