Fix postgres notification support in the QPSQLDriver.

This patch fixes a critical bug in the qsqlpsql driver where
notifications aren't delivered when received. Any blocking libpq
function(specifically PQexec) will read all the incoming data
from the socket, including any pending notifications. This would
cause the socket notifier to never be fired for incoming
notifications that are already queued inside libpq. The qsqldriver
test case was skipping the postgres notification test because of
this bug, now its enabled and passing. In order to fix this
bug I made a wrapper function for PQexec in QPSQLDriverPrivate
that calls _q_handleNotification via QMetaObject::callMethod
QueuedConnection in order to deliver pending notifications
when control returns to the event loop. I also added a flag
to ensure only one call is made each time the event loop is
entered.

This backport omits the changes to the NotificationSource version
of the "notification" signal that were included in the original
commit, since NotificationSource itself has not yet been introduced
into Qt4.

(Backport of qtbase aea684506945a12312fc05fb3bb4f549da93f7f5)

Change-Id: I19f5297094ae7ae46bfb0717e4fca744d69f7b92
Reviewed-by: Mark Brand <mabrand@mabrand.nl>

2 files changed, 51 insertions, 41 deletions

diff --git a/src/sql/drivers/psql/qsql_psql.cpp b/src/sql/drivers/psql/qsql_psql.cpp
index acab36472c..66cd431f93 100644
--- a/src/sql/drivers/psql/qsql_psql.cpp
+++ b/src/sql/drivers/psql/qsql_psql.cpp
@@ -120,14 +120,21 @@ inline void qPQfreemem(void *buffer)
 class QPSQLDriverPrivate
 {
 public:
-    QPSQLDriverPrivate() : connection(0), isUtf8(false), pro(QPSQLDriver::Version6), sn(0) {}
+    QPSQLDriverPrivate(QPSQLDriver *qq) : q(qq), connection(0), isUtf8(false), pro(QPSQLDriver::Version6), sn(0), pendingNotifyCheck(false) {}
+    QPSQLDriver *q;
     PGconn *connection;
     bool isUtf8;
     QPSQLDriver::Protocol pro;
     QSocketNotifier *sn;
     QStringList seid;
+    mutable bool pendingNotifyCheck;
 
     void appendTables(QStringList &tl, QSqlQuery &t, QChar type);
+    PGresult * exec(const char * stmt) const;
+    PGresult * exec(const QString & stmt) const;
+    QPSQLDriver::Protocol getPSQLVersion();
+    bool setEncodingUtf8();
+    void setDatestyle();
 };
 
 void QPSQLDriverPrivate::appendTables(QStringList &tl, QSqlQuery &t, QChar type)
@@ -154,6 +161,21 @@ void QPSQLDriverPrivate::appendTables(QStringList &tl, QSqlQuery &t, QChar type)
     }
 }
 
+PGresult * QPSQLDriverPrivate::exec(const char * stmt) const
+{
+    PGresult *result = PQexec(connection, stmt);
+    if (seid.size() && !pendingNotifyCheck) {
+        pendingNotifyCheck = true;
+        QMetaObject::invokeMethod(q, "_q_handleNotification", Qt::QueuedConnection, Q_ARG(int,0));
+    }
+    return result;
+}
+
+PGresult * QPSQLDriverPrivate::exec(const QString & stmt) const
+{
+    return exec(isUtf8 ? stmt.toUtf8().constData() : stmt.toLocal8Bit().constData());
+}
+
 class QPSQLResultPrivate
 {
 public:
@@ -248,9 +270,7 @@ static QVariant::Type qDecodePSQLType(int t)
 static void qDeallocatePreparedStmt(QPSQLResultPrivate *d)
 {
     const QString stmt = QLatin1String("DEALLOCATE ") + d->preparedStmtId;
-    PGresult *result = PQexec(d->driver->connection,
-                              d->driver->isUtf8 ? stmt.toUtf8().constData()
-                                                : stmt.toLocal8Bit().constData());
+    PGresult *result = d->driver->exec(stmt);
 
     if (PQresultStatus(result) != PGRES_COMMAND_OK)
         qWarning("Unable to free statement: %s", PQerrorMessage(d->driver->connection));
@@ -428,9 +448,7 @@ bool QPSQLResult::reset (const QString& query)
         return false;
     if (!driver()->isOpen() || driver()->isOpenError())
         return false;
-    d->result = PQexec(d->driver->connection,
-                       d->driver->isUtf8 ? query.toUtf8().constData()
-                                         : query.toLocal8Bit().constData());
+    d->result = d->driver->exec(query);
     return d->processResults();
 }
 
@@ -561,9 +579,7 @@ bool QPSQLResult::prepare(const QString &query)
     const QString stmtId = qMakePreparedStmtId();
     const QString stmt = QString::fromLatin1("PREPARE %1 AS ").arg(stmtId).append(qReplacePlaceholderMarkers(query));
 
-    PGresult *result = PQexec(d->driver->connection,
-                              d->driver->isUtf8 ? stmt.toUtf8().constData()
-                                                : stmt.toLocal8Bit().constData());
+    PGresult *result = d->driver->exec(stmt);
 
     if (PQresultStatus(result) != PGRES_COMMAND_OK) {
         setLastError(qMakeError(QCoreApplication::translate("QPSQLResult",
@@ -592,26 +608,24 @@ bool QPSQLResult::exec()
     else
         stmt = QString::fromLatin1("EXECUTE %1 (%2)").arg(d->preparedStmtId).arg(params);
 
-    d->result = PQexec(d->driver->connection,
-                       d->driver->isUtf8 ? stmt.toUtf8().constData()
-                                         : stmt.toLocal8Bit().constData());
+    d->result = d->driver->exec(stmt);
 
     return d->processResults();
 }
 
 ///////////////////////////////////////////////////////////////////
 
-static bool setEncodingUtf8(PGconn* connection)
+bool QPSQLDriverPrivate::setEncodingUtf8()
 {
-    PGresult* result = PQexec(connection, "SET CLIENT_ENCODING TO 'UNICODE'");
+    PGresult* result = exec("SET CLIENT_ENCODING TO 'UNICODE'");
     int status = PQresultStatus(result);
     PQclear(result);
     return status == PGRES_COMMAND_OK;
 }
 
-static void setDatestyle(PGconn* connection)
+void QPSQLDriverPrivate::setDatestyle()
 {
-    PGresult* result = PQexec(connection, "SET DATESTYLE TO 'ISO'");
+    PGresult* result = exec("SET DATESTYLE TO 'ISO'");
     int status =  PQresultStatus(result);
     if (status != PGRES_COMMAND_OK)
         qWarning("%s", PQerrorMessage(connection));
@@ -662,10 +676,10 @@ static QPSQLDriver::Protocol qMakePSQLVersion(int vMaj, int vMin)
     return QPSQLDriver::VersionUnknown;
 }
 
-static QPSQLDriver::Protocol getPSQLVersion(PGconn* connection)
+QPSQLDriver::Protocol QPSQLDriverPrivate::getPSQLVersion()
 {
     QPSQLDriver::Protocol serverVersion = QPSQLDriver::Version6;
-    PGresult* result = PQexec(connection, "select version()");
+    PGresult* result = exec("select version()");
     int status = PQresultStatus(result);
     if (status == PGRES_COMMAND_OK || status == PGRES_TUPLES_OK) {
         QString val = QString::fromAscii(PQgetvalue(result, 0, 0));
@@ -688,7 +702,7 @@ static QPSQLDriver::Protocol getPSQLVersion(PGconn* connection)
                 //Client version before QPSQLDriver::Version9 only supports escape mode for bytea type,
                 //but bytea format is set to hex by default in PSQL 9 and above. So need to force the
                 //server use the old escape mode when connects to the new server with old client library.
-                result = PQexec(connection, "SET bytea_output=escape; ");
+                result = exec("SET bytea_output=escape; ");
                 status = PQresultStatus(result);
             } else if (serverVersion == QPSQLDriver::VersionUnknown) {
                 serverVersion = clientVersion;
@@ -723,7 +737,7 @@ QPSQLDriver::QPSQLDriver(PGconn *conn, QObject *parent)
     init();
     d->connection = conn;
     if (conn) {
-        d->pro = getPSQLVersion(d->connection);
+        d->pro = d->getPSQLVersion();
         setOpen(true);
         setOpenError(false);
     }
@@ -731,7 +745,7 @@ QPSQLDriver::QPSQLDriver(PGconn *conn, QObject *parent)
 
 void QPSQLDriver::init()
 {
-    d = new QPSQLDriverPrivate();
+    d = new QPSQLDriverPrivate(this);
 }
 
 QPSQLDriver::~QPSQLDriver()
@@ -823,9 +837,9 @@ bool QPSQLDriver::open(const QString & db,
         return false;
     }
 
-    d->pro = getPSQLVersion(d->connection);
-    d->isUtf8 = setEncodingUtf8(d->connection);
-    setDatestyle(d->connection);
+    d->pro = d->getPSQLVersion();
+    d->isUtf8 = d->setEncodingUtf8();
+    d->setDatestyle();
 
     setOpen(true);
     setOpenError(false);
@@ -862,7 +876,7 @@ bool QPSQLDriver::beginTransaction()
         qWarning("QPSQLDriver::beginTransaction: Database not open");
         return false;
     }
-    PGresult* res = PQexec(d->connection, "BEGIN");
+    PGresult* res = d->exec("BEGIN");
     if (!res || PQresultStatus(res) != PGRES_COMMAND_OK) {
         PQclear(res);
         setLastError(qMakeError(tr("Could not begin transaction"),
@@ -879,7 +893,7 @@ bool QPSQLDriver::commitTransaction()
         qWarning("QPSQLDriver::commitTransaction: Database not open");
         return false;
     }
-    PGresult* res = PQexec(d->connection, "COMMIT");
+    PGresult* res = d->exec("COMMIT");
 
     bool transaction_failed = false;
 
@@ -912,7 +926,7 @@ bool QPSQLDriver::rollbackTransaction()
         qWarning("QPSQLDriver::rollbackTransaction: Database not open");
         return false;
     }
-    PGresult* res = PQexec(d->connection, "ROLLBACK");
+    PGresult* res = d->exec("ROLLBACK");
     if (!res || PQresultStatus(res) != PGRES_COMMAND_OK) {
         setLastError(qMakeError(tr("Could not rollback transaction"),
                                 QSqlError::TransactionError, d));
@@ -1295,11 +1309,11 @@ bool QPSQLDriver::subscribeToNotificationImplementation(const QString &name)
 
     int socket = PQsocket(d->connection);
     if (socket) {
+        // Add the name to the list of subscriptions here so that QSQLDriverPrivate::exec knows
+        // to check for notifications immediately after executing the LISTEN
+        d->seid << name;
         QString query = QLatin1String("LISTEN ") + escapeIdentifier(name, QSqlDriver::TableName);
-        if (PQresultStatus(PQexec(d->connection,
-                                  d->isUtf8 ? query.toUtf8().constData()
-                                            : query.toLocal8Bit().constData())
-                          ) != PGRES_COMMAND_OK) {
+        if (PQresultStatus(d->exec(query)) != PGRES_COMMAND_OK) {
             setLastError(qMakeError(tr("Unable to subscribe"), QSqlError::StatementError, d));
             return false;
         }
@@ -1308,9 +1322,11 @@ bool QPSQLDriver::subscribeToNotificationImplementation(const QString &name)
             d->sn = new QSocketNotifier(socket, QSocketNotifier::Read);
             connect(d->sn, SIGNAL(activated(int)), this, SLOT(_q_handleNotification(int)));
         }
+    } else {
+        qWarning("QPSQLDriver::subscribeToNotificationImplementation: PQsocket didn't return a valid socket to listen on");
+        return false;
     }
 
-    d->seid << name;
     return true;
 }
 
@@ -1328,10 +1344,7 @@ bool QPSQLDriver::unsubscribeFromNotificationImplementation(const QString &name)
     }
 
     QString query = QLatin1String("UNLISTEN ") + escapeIdentifier(name, QSqlDriver::TableName);
-    if (PQresultStatus(PQexec(d->connection,
-                              d->isUtf8 ? query.toUtf8().constData()
-                                        : query.toLocal8Bit().constData())
-                      ) != PGRES_COMMAND_OK) {
+    if (PQresultStatus(d->exec(query)) != PGRES_COMMAND_OK) {
         setLastError(qMakeError(tr("Unable to unsubscribe"), QSqlError::StatementError, d));
         return false;
     }
@@ -1354,6 +1367,7 @@ QStringList QPSQLDriver::subscribedToNotificationsImplementation() const
 
 void QPSQLDriver::_q_handleNotification(int)
 {
+    d->pendingNotifyCheck = false;
     PQconsumeInput(d->connection);
 
     PGnotify *notify = 0;
diff --git a/tests/auto/qsqldatabase/tst_qsqldatabase.cpp b/tests/auto/qsqldatabase/tst_qsqldatabase.cpp
index fda678b173..bbd08bdf02 100644
--- a/tests/auto/qsqldatabase/tst_qsqldatabase.cpp
+++ b/tests/auto/qsqldatabase/tst_qsqldatabase.cpp
@@ -2415,10 +2415,6 @@ void tst_QSqlDatabase::eventNotificationPSQL()
     QSqlDatabase db = QSqlDatabase::database(dbName);
     CHECK_DATABASE(db);
 
-#if defined(Q_OS_LINUX)
-    QSKIP( "Event support doesn't work on linux", SkipAll );
-#endif
-
     QSqlQuery query(db);
     QString procedureName = qTableName("posteventProc", __FILE__);