postgresql-jdbc.git / summary / log / commit / refs
commit 15e3206a13540fbbcfcf6aad9de80fedf9a72270
Author: Oliver Jowett <oliver@opencloud.com>
Date: Wed Apr 20 00:10:58 2005 +0000
Support asynchronous notifications by looking for pending messages when
getNotifications() is called. Doesn't always pick up on them (it depends
if the underlying socket input stream implements available()) but it's
better than the status quo.
Andras Kadinger <bandit@surfnonstop.com>, with editing & testcase from
Oliver.
org/postgresql/PGConnection.java | 4 +--
org/postgresql/core/PGStream.java | 15 +++++++++-
org/postgresql/core/ProtocolConnection.java | 4 +--
org/postgresql/core/v2/ProtocolConnectionImpl.java | 6 ++--
org/postgresql/core/v2/QueryExecutorImpl.java | 29 +++++++++++++++++++-
org/postgresql/core/v3/ProtocolConnectionImpl.java | 6 ++--
org/postgresql/core/v3/QueryExecutorImpl.java | 30 +++++++++++++++++++-
org/postgresql/jdbc2/AbstractJdbc2Connection.java | 4 +--
org/postgresql/test/jdbc2/NotifyTest.java | 32 ++++++++++++++++++++--
9 files changed, 113 insertions(+), 17 deletions(-)
diff --git a/org/postgresql/PGConnection.java b/org/postgresql/PGConnection.java
index efbc2085..b6ed96b1 100644
--- a/org/postgresql/PGConnection.java
+++ b/org/postgresql/PGConnection.java
@@ -3,7 +3,7 @@
* Copyright (c) 2003-2005, PostgreSQL Global Development Group
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/PGConnection.java,v 1.12 2005/01/11 08:25:43 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/PGConnection.java,v 1.13 2005/01/17 09:51:40 jurka Exp $
*
*-------------------------------------------------------------------------
*/
@@ -27,7 +27,7 @@ public interface PGConnection
* Returns null if there have been no notifications.
* @since 7.3
*/
- public PGNotification[] getNotifications();
+ public PGNotification[] getNotifications() throws SQLException;
/**
* This returns the LargeObject API for the current connection.
diff --git a/org/postgresql/core/PGStream.java b/org/postgresql/core/PGStream.java
index f68d5324..013bcf3f 100644
--- a/org/postgresql/core/PGStream.java
+++ b/org/postgresql/core/PGStream.java
@@ -3,7 +3,7 @@
* Copyright (c) 2003-2005, PostgreSQL Global Development Group
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/core/PGStream.java,v 1.13 2005/01/11 08:25:43 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/core/PGStream.java,v 1.14 2005/02/22 22:35:54 oliver Exp $
*
*-------------------------------------------------------------------------
*/
@@ -71,6 +71,19 @@ public class PGStream
return connection;
}
+ /**
+ * Check for pending backend messages without blocking.
+ * Might return false when there actually are messages
+ * waiting, depending on the characteristics of the
+ * underlying socket. This is used to detect asynchronous
+ * notifies from the backend, when available.
+ *
+ * @return true if there is a pending backend message
+ */
+ public boolean hasMessagePending() throws IOException {
+ return connection.getInputStream().available() > 0;
+ }
+
/**
* Switch this stream to using a new socket. Any existing socket
* is <em>not</em> closed; it's assumed that we are changing to
diff --git a/org/postgresql/core/ProtocolConnection.java b/org/postgresql/core/ProtocolConnection.java
index c2eb22ee..bd6e6d4a 100644
--- a/org/postgresql/core/ProtocolConnection.java
+++ b/org/postgresql/core/ProtocolConnection.java
@@ -4,7 +4,7 @@
* Copyright (c) 2004, Open Cloud Limited.
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/core/ProtocolConnection.java,v 1.3 2004/11/09 08:44:43 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/core/ProtocolConnection.java,v 1.4 2005/01/11 08:25:43 jurka Exp $
*
*-------------------------------------------------------------------------
*/
@@ -82,7 +82,7 @@ public interface ProtocolConnection {
* @return an array of notifications; if there are no notifications, an empty
* array is returned.
*/
- PGNotification[] getNotifications();
+ PGNotification[] getNotifications() throws SQLException;
/**
* Retrieve and clear the chain of warnings accumulated on this connection.
diff --git a/org/postgresql/core/v2/ProtocolConnectionImpl.java b/org/postgresql/core/v2/ProtocolConnectionImpl.java
index 43cbe307..6806b98c 100644
--- a/org/postgresql/core/v2/ProtocolConnectionImpl.java
+++ b/org/postgresql/core/v2/ProtocolConnectionImpl.java
@@ -4,7 +4,7 @@
* Copyright (c) 2004, Open Cloud Limited.
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/core/v2/ProtocolConnectionImpl.java,v 1.4 2004/11/09 08:45:31 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/core/v2/ProtocolConnectionImpl.java,v 1.5 2005/01/11 08:25:43 jurka Exp $
*
*-------------------------------------------------------------------------
*/
@@ -56,8 +56,8 @@ class ProtocolConnectionImpl implements ProtocolConnection {
return transactionState;
}
- public synchronized PGNotification[]
- getNotifications() {
+ public synchronized PGNotification[] getNotifications() throws SQLException {
+ executor.processNotifies();
PGNotification[] array = (PGNotification[])notifications.toArray(new PGNotification[notifications.size()]);
notifications.clear();
return array;
diff --git a/org/postgresql/core/v2/QueryExecutorImpl.java b/org/postgresql/core/v2/QueryExecutorImpl.java
index 0aac3f92..726ac338 100644
--- a/org/postgresql/core/v2/QueryExecutorImpl.java
+++ b/org/postgresql/core/v2/QueryExecutorImpl.java
@@ -4,7 +4,7 @@
* Copyright (c) 2004, Open Cloud Limited.
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/core/v2/QueryExecutorImpl.java,v 1.11 2005/02/10 19:53:17 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/core/v2/QueryExecutorImpl.java,v 1.12 2005/02/15 08:56:25 jurka Exp $
*
*-------------------------------------------------------------------------
*/
@@ -148,6 +148,33 @@ public class QueryExecutorImpl implements QueryExecutor {
pgStream.flush();
}
+ public synchronized void processNotifies() throws SQLException {
+ // Asynchronous notifies only arrive when we are not in a transaction
+ if (protoConnection.getTransactionState() != ProtocolConnection.TRANSACTION_IDLE)
+ return;
+
+ try {
+ while (pgStream.hasMessagePending()) {
+ int c = pgStream.ReceiveChar();
+ switch (c) {
+ case 'A': // Asynchronous Notify
+ receiveAsyncNotify();
+ break;
+ case 'E': // Error Message
+ throw receiveErrorMessage();
+ // break;
+ case 'N': // Error Notification
+ protoConnection.addWarning(receiveNotification());
+ break;
+ default:
+ throw new PSQLException(GT.tr("Unknown Response Type {0}.", new Character((char) c)), PSQLState.CONNECTION_FAILURE);
+ }
+ }
+ } catch (IOException ioe) {
+ throw new PSQLException(GT.tr("An I/O error occured while sending to the backend."), PSQLState.CONNECTION_FAILURE, ioe);
+ }
+ }
+
private byte[] receiveFastpathResult() throws IOException, SQLException {
SQLException error = null;
boolean endQuery = false;
diff --git a/org/postgresql/core/v3/ProtocolConnectionImpl.java b/org/postgresql/core/v3/ProtocolConnectionImpl.java
index 7f3a27c2..b5443da8 100644
--- a/org/postgresql/core/v3/ProtocolConnectionImpl.java
+++ b/org/postgresql/core/v3/ProtocolConnectionImpl.java
@@ -4,7 +4,7 @@
* Copyright (c) 2004, Open Cloud Limited.
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/core/v3/ProtocolConnectionImpl.java,v 1.5 2004/12/11 03:31:52 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/core/v3/ProtocolConnectionImpl.java,v 1.6 2005/01/11 08:25:44 jurka Exp $
*
*-------------------------------------------------------------------------
*/
@@ -57,8 +57,8 @@ class ProtocolConnectionImpl implements ProtocolConnection {
return transactionState;
}
- public synchronized PGNotification[]
- getNotifications() {
+ public synchronized PGNotification[] getNotifications() throws SQLException {
+ executor.processNotifies();
PGNotification[] array = (PGNotification[])notifications.toArray(new PGNotification[notifications.size()]);
notifications.clear();
return array;
diff --git a/org/postgresql/core/v3/QueryExecutorImpl.java b/org/postgresql/core/v3/QueryExecutorImpl.java
index dce00fe7..393d5164 100644
--- a/org/postgresql/core/v3/QueryExecutorImpl.java
+++ b/org/postgresql/core/v3/QueryExecutorImpl.java
@@ -4,7 +4,7 @@
* Copyright (c) 2004, Open Cloud Limited.
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/core/v3/QueryExecutorImpl.java,v 1.21 2005/02/01 07:27:54 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/core/v3/QueryExecutorImpl.java,v 1.22 2005/04/10 16:44:13 jurka Exp $
*
*-------------------------------------------------------------------------
*/
@@ -524,6 +524,34 @@ public class QueryExecutorImpl implements QueryExecutor {
pgStream.flush();
}
+ public synchronized void processNotifies() throws SQLException {
+ // Asynchronous notifies only arrive when we are not in a transaction
+ if (protoConnection.getTransactionState() != ProtocolConnection.TRANSACTION_IDLE)
+ return;
+
+ try {
+ while (pgStream.hasMessagePending()) {
+ int c = pgStream.ReceiveChar();
+ switch (c) {
+ case 'A': // Asynchronous Notify
+ receiveAsyncNotify();
+ break;
+ case 'E': // Error Response (response to pretty much everything; backend then skips until Sync)
+ throw receiveErrorResponse();
+ // break;
+ case 'N': // Notice Response (warnings / info)
+ SQLWarning warning = receiveNoticeResponse();
+ protoConnection.addWarning(warning);
+ break;
+ default:
+ throw new PSQLException(GT.tr("Unknown Response Type {0}.", new Character((char) c)), PSQLState.CONNECTION_FAILURE);
+ }
+ }
+ } catch (IOException ioe) {
+ throw new PSQLException(GT.tr("An I/O error occured while sending to the backend."), PSQLState.CONNECTION_FAILURE, ioe);
+ }
+ }
+
private byte[] receiveFastpathResult() throws IOException, SQLException {
boolean endQuery = false;
SQLException error = null;
diff --git a/org/postgresql/jdbc2/AbstractJdbc2Connection.java b/org/postgresql/jdbc2/AbstractJdbc2Connection.java
index b49d514f..a6e200cf 100644
--- a/org/postgresql/jdbc2/AbstractJdbc2Connection.java
+++ b/org/postgresql/jdbc2/AbstractJdbc2Connection.java
@@ -3,7 +3,7 @@
* Copyright (c) 2004-2005, PostgreSQL Global Development Group
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/jdbc2/AbstractJdbc2Connection.java,v 1.27 2005/02/15 08:56:25 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/jdbc2/AbstractJdbc2Connection.java,v 1.28 2005/04/10 21:54:16 jurka Exp $
*
*-------------------------------------------------------------------------
*/
@@ -954,7 +954,7 @@ public abstract class AbstractJdbc2Connection implements BaseConnection
protoConnection.sendQueryCancel();
}
- public PGNotification[] getNotifications()
+ public PGNotification[] getNotifications() throws SQLException
{
// Backwards-compatibility hand-holding.
PGNotification[] notifications = protoConnection.getNotifications();
diff --git a/org/postgresql/test/jdbc2/NotifyTest.java b/org/postgresql/test/jdbc2/NotifyTest.java
index 7130df11..40cbb4d2 100644
--- a/org/postgresql/test/jdbc2/NotifyTest.java
+++ b/org/postgresql/test/jdbc2/NotifyTest.java
@@ -3,7 +3,7 @@
* Copyright (c) 2004-2005, PostgreSQL Global Development Group
*
* IDENTIFICATION
-* $PostgreSQL: pgjdbc/org/postgresql/test/jdbc2/NotifyTest.java,v 1.4 2004/11/09 08:54:42 jurka Exp $
+* $PostgreSQL: pgjdbc/org/postgresql/test/jdbc2/NotifyTest.java,v 1.5 2005/01/11 08:25:48 jurka Exp $
*
*-------------------------------------------------------------------------
*/
@@ -17,7 +17,6 @@ import org.postgresql.PGNotification;
public class NotifyTest extends TestCase
{
-
private Connection conn;
public NotifyTest(String name)
@@ -49,4 +48,33 @@ public class NotifyTest extends TestCase
stmt.close();
}
+
+ public void testAsyncNotify() throws SQLException
+ {
+ Statement stmt = conn.createStatement();
+ stmt.executeUpdate("LISTEN mynotification");
+
+ // Notify on a separate connection to get an async notify on the first.
+ Connection conn2 = TestUtil.openDB();
+ try {
+ Statement stmt2 = conn2.createStatement();
+ stmt2.executeUpdate("NOTIFY mynotification");
+ stmt2.close();
+ } finally {
+ conn2.close();
+ }
+
+ // Wait a bit to let the notify come through..
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException ie) {}
+
+ PGNotification notifications[] = ((org.postgresql.PGConnection)conn).getNotifications();
+ assertNotNull(notifications);
+ assertEquals(1, notifications.length);
+ assertEquals("mynotification", notifications[0].getName());
+ assertEquals("", notifications[0].getParameter());
+
+ stmt.close();
+ }
}
[parent: f4ab943ef571]