Restore subscription API

Author: dantti

README.md

@@ -241,26 +241,26 @@ delete cancelator; // cancels the in-flight query; co_await resumes with an erro
 
 ### Notifications (Postgres only)
 
-Subscribe to a channel and connect to the driver signal to receive payloads. If the connection drops, re-subscribe after reconnecting:
+Subscribe with a per-channel callback. If the connection drops, re-subscribe after reconnecting:
 
 ```c++
 ADatabase db;
 QObject receiver;
 
-connect(db.driver(), &ADriver::notificationReceived, &receiver,
-        [](const ADatabaseNotification &notification) {
-            qDebug() << "DB notification:" << notification.self << notification.name
-                     << notification.payload;
-        });
-
 connect(db.driver(), &ADriver::stateChanged, &receiver,
         [&db](ADatabase::State state, const QString &status) {
             qDebug() << "state changed" << state << status << db.isOpen();
 
             if (state == ADatabase::Disconnected) {
                 qDebug() << "state disconnected";
             } else if (state == ADatabase::Connected) {
-                db.subscribeToNotification(u"my_awesome_notification"_s, &receiver);
+                db.subscribeToNotification(
+                    u"my_awesome_notification"_s,
+                    &receiver,
+                    [](const ADatabaseNotification &notification) {
+                        qDebug() << "DB notification:" << notification.self
+                                 << notification.name << notification.payload;
+                    });
             }
         });
 

src/ADriverOdbc.cpp

@@ -1165,7 +1165,8 @@ int ADriverOdbc::queueSize() const
 
 void ADriverOdbc::subscribeToNotification(const std::shared_ptr<ADriver> &,
                                           const QString &,
-                                          QObject *)
+                                          QObject *,
+                                          ANotificationFn)
 {
 }
 

src/ADriverOdbc.hpp

@@ -176,7 +176,8 @@ class ADriverOdbc final : public ADriver
 
     void subscribeToNotification(const std::shared_ptr<ADriver> &db,
                                  const QString &name,
-                                 QObject *receiver) override;
+                                 QObject *receiver,
+                                 ANotificationFn cb) override;
     QStringList subscribedToNotifications() const override;
     void unsubscribeFromNotification(const std::shared_ptr<ADriver> &db,
                                      const QString &name) override;

src/ADriverSqlite.cpp

@@ -339,11 +339,13 @@ int ADriverSqlite::queueSize() const
 
 void ADriverSqlite::subscribeToNotification(const std::shared_ptr<ADriver> &db,
                                             const QString &name,
-                                            QObject *receiver)
+                                            QObject *receiver,
+                                            ANotificationFn cb)
 {
     Q_UNUSED(db);
     Q_UNUSED(name);
     Q_UNUSED(receiver);
+    Q_UNUSED(cb);
 }
 
 QStringList ADriverSqlite::subscribedToNotifications() const

src/ADriverSqlite.hpp

@@ -179,7 +179,8 @@ class ADriverSqlite final : public ADriver
 
     void subscribeToNotification(const std::shared_ptr<ADriver> &db,
                                  const QString &name,
-                                 QObject *receiver) override;
+                                 QObject *receiver,
+                                 ANotificationFn cb) override;
     QStringList subscribedToNotifications() const override;
     void unsubscribeFromNotification(const std::shared_ptr<ADriver> &db,
                                      const QString &name) override;

src/adatabase.cpp

@@ -227,10 +227,12 @@ bool ADatabase::pipelineSync()
     return d->pipelineSync();
 }
 
-void ADatabase::subscribeToNotification(const QString &channel, QObject *receiver)
+void ADatabase::subscribeToNotification(const QString &channel,
+                                        QObject *receiver,
+                                        ANotificationFn cb)
 {
     Q_ASSERT(d);
-    d->subscribeToNotification(d, channel, receiver);
+    d->subscribeToNotification(d, channel, receiver, std::move(cb));
 }
 
 QStringList ADatabase::subscribedToNotifications() const

src/adatabase.h

@@ -8,6 +8,7 @@
 #include <asqliterals.h>
 #include <chrono>
 #include <cstddef>
+#include <functional>
 #include <memory>
 
 #include <QObject>
@@ -29,6 +30,8 @@ class ADatabaseNotification
     bool self;
 };
 
+using ANotificationFn = std::function<void(const ADatabaseNotification &notification)>;
+
 template <typename T>
 class ACoroExpected;
 
@@ -335,9 +338,10 @@ class ASQL_EXPORT ADatabase
      * back, in which case it will not be effective.
      *
      * \param channel name of the channel
-     * \param receiver optional lifetime guard for the LISTEN operation
+     * \param receiver optional lifetime guard; destroyed receivers drop the subscription
+     * \param cb called when a notification arrives on this channel
      */
-    void subscribeToNotification(const QString &channel, QObject *receiver = nullptr);
+    void subscribeToNotification(const QString &channel, QObject *receiver, ANotificationFn cb);
 
     /**
      * @brief subscribedToNotifications

src/adriver.cpp

@@ -221,11 +221,13 @@ int ADriver::queueSize() const
 
 void ADriver::subscribeToNotification(const std::shared_ptr<ADriver> &db,
                                       const QString &name,
-                                      QObject *receiver)
+                                      QObject *receiver,
+                                      ANotificationFn cb)
 {
     Q_UNUSED(db);
     Q_UNUSED(name);
     Q_UNUSED(receiver);
+    Q_UNUSED(cb);
 }
 
 QStringList ADriver::subscribedToNotifications() const

src/adriver.h

@@ -82,7 +82,8 @@ class ASQL_EXPORT ADriver : public QObject
 
     virtual void subscribeToNotification(const std::shared_ptr<ADriver> &driver,
                                          const QString &name,
-                                         QObject *receiver);
+                                         QObject *receiver,
+                                         ANotificationFn cb);
     virtual QStringList subscribedToNotifications() const;
     virtual void unsubscribeFromNotification(const std::shared_ptr<ADriver> &driver,
                                              const QString &name);

src/adrivermysql.cpp

@@ -1065,11 +1065,13 @@ int ADriverMysql::queueSize() const
 
 void ADriverMysql::subscribeToNotification(const std::shared_ptr<ADriver> &db,
                                            const QString &name,
-                                           QObject *receiver)
+                                           QObject *receiver,
+                                           ANotificationFn cb)
 {
     Q_UNUSED(db)
     Q_UNUSED(name)
     Q_UNUSED(receiver)
+    Q_UNUSED(cb)
     // MySQL does not support server-side notifications
 }