Repository: pgjdbc/pgjdbc
Pull request #778 (2017-03-11 10:12:11)

feat: improve waiting for notifications by providing a timeout option
Improve waiting for notifications using the LISTEN/NOTIFY framework by adding an optional timeout to the PGConnection.getNotifications() method. The change does not have any effects for users that don't use this functionality and the API remain fully backwards compatible.

PS: Please squash commits when merging, the text of this pull request should give a nice commit message.

codecov-io: # [Codecov](https://codecov.io/gh/pgjdbc/pgjdbc/pull/778?src=pr&el=h1) Report
> Merging [#778](https://codecov.io/gh/pgjdbc/pgjdbc/pull/778?src=pr&el=desc) into [master](https://codecov.io/gh/pgjdbc/pgjdbc/commit/9c80adc24776ae02e0aff419e1cddf26a28eeff0?src=pr&el=desc) will **decrease** coverage by `0.11%`.
> The diff coverage is `19.35%`.



```diff
@@             Coverage Diff              @@
##             master     #778      +/-   ##
============================================
- Coverage     65.25%   65.13%   -0.12%     
- Complexity     3496     3497       +1     
============================================
  Files           164      164              
  Lines         15125    15152      +27     
  Branches       2450     2458       +8     
============================================
  Hits           9870     9870              
- Misses         4078     4099      +21     
- Partials       1177     1183       +6
```

------

[Continue to review full report at Codecov](https://codecov.io/gh/pgjdbc/pgjdbc/pull/778?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pgjdbc/pgjdbc/pull/778?src=pr&el=footer). Last update [9c80adc...da49f92](https://codecov.io/gh/pgjdbc/pgjdbc/compare/9c80adc24776ae02e0aff419e1cddf26a28eeff0...da49f926eb6f8a36e67e6168eed6f8bc920d2687?src=pr&el=footer&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
dmigowski: OK, I can do that.

Von: Vladimir Sitnikov [mailto:notifications@github.com]
Gesendet: Samstag, 11. März 2017 13:39
An: pgjdbc/pgjdbc <pgjdbc@noreply.github.com>
Cc: Daniel Migowski <dmigowski@ikoffice.de>; Author <author@noreply.github.com>
Betreff: Re: [pgjdbc/pgjdbc] feat: improve waiting for notifications by providing a timeout option (#778)


@vlsi requested changes on this pull request.

Would you please add test cases to cover the new feature?

________________________________

In pgjdbc/src/main/java/org/postgresql/PGConnection.java<https://github.com/pgjdbc/pgjdbc/pull/778#discussion_r105532507>:

> @@ -32,6 +32,20 @@

   PGNotification[] getNotifications() throws SQLException;



   /**

+   * This method returns any notifications that have been received since the last call to this

+   * method. Returns null if there have been no notifications. A timeout can be speficied so the

speficied -> specified

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub<https://github.com/pgjdbc/pgjdbc/pull/778#pullrequestreview-26421764>, or mute the thread<https://github.com/notifications/unsubscribe-auth/ADpQwwaCpSyvAMH3mZHhFDCweMJPleHxks5rkpXXgaJpZM4MaKZD>.

davecramer: @dmigowski any chance you can roll this https://github.com/pgjdbc/pgjdbc/pull/579 in at the same time ?
dmigowski: At first I thought about to implement something like NotificationListeners also, but to implement this in a useful way there should be a Thread listening on the Socket all the time, so notifications come along truly asynchronous at the time they are sent. This thread would add incoming messages (except notifications and maybe warnings) to a queue where other Threads read them from, e.g. the QueryExecutor running statements could read its responses from that queue then.

The implementation in #579 is not very useful in my eyes because you still have to issue commands to the server or at least call getNotifications() from time to time to have the notifications delivered. If timely delivery is not a concern you can then dispatch them to your listeners manually, so nothing gained. Does anyone see a nice usecase for these notifications?

Because my first approach with having a distinct reader thread seemed to heavy, I came up with the solution of a timeouting getNotifications(), but this is of course only useful when you use your connection for nothing else but listening for notifies and reacting to them.

More thoughts: If one uses my getNotifications() with timeout on a distinct connection to the database handled by a distinct thread, one could easily write client code that dispatches these notifications.

Conclusion: When you still find the “async notification feature” useful, I would add them here, but as I said, they are not even async in the current implementation. The feature simply describes the dispatching of them on a listener basis.


Von: Dave Cramer [mailto:notifications@github.com]
Gesendet: Samstag, 11. März 2017 22:04
An: pgjdbc/pgjdbc <pgjdbc@noreply.github.com>
Cc: Daniel Migowski <dmigowski@ikoffice.de>; Mention <mention@noreply.github.com>
Betreff: Re: [pgjdbc/pgjdbc] feat: improve waiting for notifications by providing a timeout option (#778)


@dmigowski<https://github.com/dmigowski> any chance you can roll this #579<https://github.com/pgjdbc/pgjdbc/pull/579> in at the same time ?

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub<https://github.com/pgjdbc/pgjdbc/pull/778#issuecomment-285899971>, or mute the thread<https://github.com/notifications/unsubscribe-auth/ADpQwwDDy8fK5m-7cN9KIcAeSKqrfzRKks5rkwxRgaJpZM4MaKZD>.

dmigowski: @vlsi: I added the unit tests, also fixing a bug I found in the mean time. The code coverage is sadly -0.01% now, because i added some code I cannot get to run in the tests (the case of a Warning sent by the server while waiting for notifications). I would fix that if you can tell me how to simulate that in my tests.
dmigowski: Also would like to have my patch committed now. I could create a separate one for #579 if that is still wished for.  
davecramer: looks good to me
dmigowski: @vlsi: What is your opinion about this commit now that I added unit tests? What about my discussion regarding your #579?
dmigowski: @vlsi: Can I do something for this change to be accepted? I added unit tests as requested. Please don't be insulted because I did't want to incorporate your #579 this time. Thanks in advance. 
davecramer: @dmigowski #579 is mine, he's not insulted by it ;) I'll try to get to this Monday. Thanks!
dmigowski: Thank you very much! If this gets pulled in, I will prepare my next pull request which increases the drivers performance by a few percent points.

Von: Dave Cramer [mailto:notifications@github.com]
Gesendet: Sonntag, 19. März 2017 12:12
An: pgjdbc/pgjdbc <pgjdbc@noreply.github.com>
Cc: Daniel Migowski <dmigowski@ikoffice.de>; Mention <mention@noreply.github.com>
Betreff: Re: [pgjdbc/pgjdbc] feat: improve waiting for notifications by providing a timeout option (#778)


@dmigowski<https://github.com/dmigowski> #579<https://github.com/pgjdbc/pgjdbc/pull/579> is mine, he's not insulted by it ;) I'll try to get to this Monday. Thanks!

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub<https://github.com/pgjdbc/pgjdbc/pull/778#issuecomment-287609285>, or mute the thread<https://github.com/notifications/unsubscribe-auth/ADpQw8GMRThoDkPF9GJkR4tcz0ojxasqks5rnQ1kgaJpZM4MaKZD>.

vlsi: @dmigowski , there is a couple of things that bothers me
1) `setSocketTimeout(0)`. It looks like it would interfere with existing `socketTimeout` connection option.

2) Can you clarify why did you put all the tests into a single test method? Having separate test methods helps debugging (it is easier to execute/debug a test method rather than a part of a big test method)
dmigowski: Hello Vladimir,

I didn’t know that socket timeout was already used on other places, but you are right, I should read the current value and reset it, just in case it has been changed by other means than the options passed as connection parameters.

I will also split the different tests into separate methods.

Regards,
Daniel Migowski

Von: Vladimir Sitnikov [mailto:notifications@github.com]
Gesendet: Montag, 20. März 2017 07:53
An: pgjdbc/pgjdbc <pgjdbc@noreply.github.com>
Cc: Daniel Migowski <dmigowski@ikoffice.de>; Mention <mention@noreply.github.com>
Betreff: Re: [pgjdbc/pgjdbc] feat: improve waiting for notifications by providing a timeout option (#778)


@dmigowski<https://github.com/dmigowski> , there is a couple of things that bothers me

1.      setSocketTimeout(0). It looks like it would interfere with existing socketTimeout connection option.

2.      Can you clarify why did you put all the tests into a single test method? Having separate test methods helps debugging (it is easier to execute/debug a test method rather than a part of a big test method)

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub<https://github.com/pgjdbc/pgjdbc/pull/778#issuecomment-287688069>, or mute the thread<https://github.com/notifications/unsubscribe-auth/ADpQw6Vq2tEbjVNNBJlUJE2WNx9Hc46bks5rniJigaJpZM4MaKZD>.

dmigowski: @vlsi: I broke up the test into pieces now and revert the timeout to the previous value after returing from getNotifications().  
dmigowski: There is a failure in the replication test suite now that I didn't touch. Could someone please investigate? I don't think it is my fault.

Tests run: 56, Failures: 1, Errors: 0, Skipped: 3, Time elapsed: 7.57 sec <<< FAILURE! - in org.postgresql.replication.ReplicationTestSuite

testLastReceiveLSNCorrectOnView(org.postgresql.replication.LogicalReplicationStatusTest)  Time elapsed: 0.063 sec  <<< FAILURE!

java.lang.AssertionError: Replication stream by execute forceUpdateStatus should send to view actual received position that allow monitoring lag

Expected: <LSN{0/16D3390}>
     but: was <LSN{0/16D3457}>
	at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
	at org.postgresql.replication.LogicalReplicationStatusTest.testLastReceiveLSNCorrectOnView(LogicalReplicationStatusTest.java:162)


dmigowski: @vlsi: Is there still something to do? @davecramer: Can you pull this in now?

Patch:
diff --git a/docs/documentation/head/listennotify.md b/docs/documentation/head/listennotify.md
index b5863ba3..0352a0b3 100644
--- a/docs/documentation/head/listennotify.md
+++ b/docs/documentation/head/listennotify.md
@@ -24,6 +24,8 @@ any outstanding notifications.
 
 > A key limitation of the JDBC driver is that it cannot receive asynchronous
 notifications and must poll the backend to check if any notifications were issued.
+I timeout can be given to the poll function, but then the execution of statements
+from other threads will block.  
 
 <a name="listen-notify-example"></a>
 **Example 9.2. Receiving Notifications**
@@ -72,20 +74,16 @@ class Listener extends Thread
 
 	public void run()
 	{
-		while (true)
+		try
 		{
-			try
+			while (true)
 			{
-				// issue a dummy query to contact the backend
-				// and receive any pending notifications.
-
-				Statement stmt = conn.createStatement();
-				ResultSet rs = stmt.executeQuery("SELECT 1");
-				rs.close();
-				stmt.close();
-
 				org.postgresql.PGNotification notifications[] = pgconn.getNotifications();
 				
+				// If this thread is the only one that uses the connection, a timeout can be used to 
+				// receive notifications immediately:
+				// org.postgresql.PGNotification notifications[] = pgconn.getNotifications(10000);
+				
 				if (notifications != null)
 				{
 					for (int i=0; i&lt;notifications.length; i++)
@@ -97,14 +95,14 @@ class Listener extends Thread
 				
 				Thread.sleep(500);
 			}
-			catch (SQLException sqle)
-			{
-				sqle.printStackTrace();
-			}
-			catch (InterruptedException ie)
-			{
-				ie.printStackTrace();
-			}
+		}
+		catch (SQLException sqle)
+		{
+			sqle.printStackTrace();
+		}
+		catch (InterruptedException ie)
+		{
+			ie.printStackTrace();
 		}
 	}
 }
diff --git a/pgjdbc/src/main/java/org/postgresql/PGConnection.java b/pgjdbc/src/main/java/org/postgresql/PGConnection.java
index 7a6b6d8e..3182ebae 100644
--- a/pgjdbc/src/main/java/org/postgresql/PGConnection.java
+++ b/pgjdbc/src/main/java/org/postgresql/PGConnection.java
@@ -31,6 +31,20 @@ public interface PGConnection {
    */
   PGNotification[] getNotifications() throws SQLException;
 
+  /**
+   * This method returns any notifications that have been received since the last call to this
+   * method. Returns null if there have been no notifications. A timeout can be specified so the
+   * driver waits for notifications.
+   *
+   * @param timeoutMillis when 0, blocks forever. when &gt; 0, blocks up to the specified number of millies
+   *        or until at least one notification has been received. If more than one notification is
+   *        about to be received, these will be returned in one batch.
+   * @return notifications that have been received
+   * @throws SQLException if something wrong happens
+   * @since 43
+   */
+  PGNotification[] getNotifications(int timeoutMillis) throws SQLException;
+
   /**
    * This returns the COPY API for the current connection.
    *
diff --git a/pgjdbc/src/main/java/org/postgresql/core/QueryExecutor.java b/pgjdbc/src/main/java/org/postgresql/core/QueryExecutor.java
index fc061346..db58a091 100644
--- a/pgjdbc/src/main/java/org/postgresql/core/QueryExecutor.java
+++ b/pgjdbc/src/main/java/org/postgresql/core/QueryExecutor.java
@@ -217,6 +217,16 @@ public interface QueryExecutor extends TypeTransferModeRegistry {
    */
   void processNotifies() throws SQLException;
 
+  /**
+   * Prior to attempting to retrieve notifications, we need to pull any recently received
+   * notifications off of the network buffers. The notification retrieval in ProtocolConnection
+   * cannot do this as it is prone to deadlock, so the higher level caller must be responsible which
+   * requires exposing this method. This variant supports blocking for the given time in millis.
+   *
+   * @throws SQLException if and error occurs while fetching notifications
+   */
+  void processNotifies(int timeoutMillis) throws SQLException;
+
   //
   // Fastpath interface.
   //
diff --git a/pgjdbc/src/main/java/org/postgresql/core/QueryExecutorBase.java b/pgjdbc/src/main/java/org/postgresql/core/QueryExecutorBase.java
index f8708187..732d3a29 100644
--- a/pgjdbc/src/main/java/org/postgresql/core/QueryExecutorBase.java
+++ b/pgjdbc/src/main/java/org/postgresql/core/QueryExecutorBase.java
@@ -375,4 +375,8 @@ public abstract class QueryExecutorBase implements QueryExecutor {
   public void setFlushCacheOnDeallocate(boolean flushCacheOnDeallocate) {
     this.flushCacheOnDeallocate = flushCacheOnDeallocate;
   }
+
+  protected boolean hasNotifications() {
+    return notifications.size() > 0;
+  }
 }
diff --git a/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java b/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
index b4c827da..25400f6f 100644
--- a/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
+++ b/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java
@@ -45,6 +45,9 @@ import java.io.IOException;
 import java.lang.ref.PhantomReference;
 import java.lang.ref.Reference;
 import java.lang.ref.ReferenceQueue;
+import java.net.Socket;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
 import java.sql.SQLException;
 import java.sql.SQLWarning;
 import java.sql.Statement;
@@ -644,35 +647,96 @@ public class QueryExecutorImpl extends QueryExecutorBase {
     pgStream.flush();
   }
 
+  // Just for API compatibility with previous versions.
   public synchronized void processNotifies() throws SQLException {
+    processNotifies(-1);
+  }
+
+  /**
+   * @param timeoutMillis when &gt; 0, block for this time
+   *                      when =0, block forever
+   *                      when &lt; 0, don't block
+   */
+  public synchronized void processNotifies(int timeoutMillis) throws SQLException {
     waitOnLock();
     // Asynchronous notifies only arrive when we are not in a transaction
     if (getTransactionState() != TransactionState.IDLE) {
       return;
     }
 
+    if (hasNotifications()) {
+      // No need to timeout when there are already notifications. We just check for more in this case.
+      timeoutMillis = -1;
+    }
+
+    boolean useTimeout = timeoutMillis > 0;
+    long startTime = 0;
+    int oldTimeout = 0;
+    if (useTimeout) {
+      startTime = System.currentTimeMillis();
+      try {
+        oldTimeout = pgStream.getSocket().getSoTimeout();
+      } catch (SocketException e) {
+        throw new PSQLException(GT.tr("An error occurred while trying to get the socket "
+          + "timeout."), PSQLState.CONNECTION_FAILURE, e);
+      }
+    }
+
     try {
-      while (pgStream.hasMessagePending()) {
+      while (pgStream.hasMessagePending() || timeoutMillis >= 0 ) {
+        if (useTimeout && timeoutMillis >= 0) {
+          setSocketTimeout(timeoutMillis);
+        }
         int c = pgStream.receiveChar();
+        if (useTimeout && timeoutMillis >= 0) {
+          setSocketTimeout(0); // Don't timeout after first char
+        }
         switch (c) {
           case 'A': // Asynchronous Notify
             receiveAsyncNotify();
-            break;
+            timeoutMillis = -1;
+            continue;
           case 'E':
             // Error Response (response to pretty much everything; backend then skips until Sync)
             throw receiveErrorResponse();
           case 'N': // Notice Response (warnings / info)
             SQLWarning warning = receiveNoticeResponse();
             addWarning(warning);
+            if (useTimeout) {
+              long newTimeMillis = System.currentTimeMillis();
+              timeoutMillis += startTime - newTimeMillis; // Overflows after 49 days, ignore that
+              startTime = newTimeMillis;
+              if (timeoutMillis == 0) {
+                timeoutMillis = -1; // Don't accidentially wait forever
+              }
+            }
             break;
           default:
             throw new PSQLException(GT.tr("Unknown Response Type {0}.", (char) c),
                 PSQLState.CONNECTION_FAILURE);
         }
       }
+    } catch (SocketTimeoutException ioe) {
+      // No notifications this time...
     } catch (IOException ioe) {
       throw new PSQLException(GT.tr("An I/O error occurred while sending to the backend."),
           PSQLState.CONNECTION_FAILURE, ioe);
+    } finally {
+      if (useTimeout) {
+        setSocketTimeout(oldTimeout);
+      }
+    }
+  }
+
+  private void setSocketTimeout(int millis) throws PSQLException {
+    try {
+      Socket s = pgStream.getSocket();
+      if (!s.isClosed()) { // Is this check required?
+        pgStream.getSocket().setSoTimeout(millis);
+      }
+    } catch (SocketException e) {
+      throw new PSQLException(GT.tr("An error occurred while trying to reset the socket timeout."),
+        PSQLState.CONNECTION_FAILURE, e);
     }
   }
 
diff --git a/pgjdbc/src/main/java/org/postgresql/jdbc/PgConnection.java b/pgjdbc/src/main/java/org/postgresql/jdbc/PgConnection.java
index 4c0951a3..ac12ce88 100644
--- a/pgjdbc/src/main/java/org/postgresql/jdbc/PgConnection.java
+++ b/pgjdbc/src/main/java/org/postgresql/jdbc/PgConnection.java
@@ -973,8 +973,13 @@ public class PgConnection implements BaseConnection {
 
   @Override
   public PGNotification[] getNotifications() throws SQLException {
+    return getNotifications(-1);
+  }
+
+  @Override
+  public PGNotification[] getNotifications(int timeoutMillis) throws SQLException {
     checkClosed();
-    getQueryExecutor().processNotifies();
+    getQueryExecutor().processNotifies(timeoutMillis);
     // Backwards-compatibility hand-holding.
     PGNotification[] notifications = queryExecutor.getNotifications();
     return (notifications.length == 0 ? null : notifications);


Test patch:
diff --git a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/NotifyTest.java b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/NotifyTest.java
index 7c0daa4d..d4d7206b 100644
--- a/pgjdbc/src/test/java/org/postgresql/test/jdbc2/NotifyTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/test/jdbc2/NotifyTest.java
@@ -7,12 +7,14 @@ package org.postgresql.test.jdbc2;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 
 import org.postgresql.PGNotification;
 import org.postgresql.core.ServerVersion;
 import org.postgresql.test.TestUtil;
 
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -33,7 +35,7 @@ public class NotifyTest {
     TestUtil.closeDB(conn);
   }
 
-  @Test
+  @Test(timeout = 60000)
   public void testNotify() throws SQLException {
     Statement stmt = conn.createStatement();
     stmt.executeUpdate("LISTEN mynotification");
@@ -48,7 +50,7 @@ public class NotifyTest {
     stmt.close();
   }
 
-  @Test
+  @Test(timeout = 60000)
   public void testNotifyArgument() throws Exception {
     if (!TestUtil.haveMinimumServerVersion(conn, ServerVersion.v9_0) || TestUtil.isProtocolVersion(conn, 2)) {
       return;
@@ -67,28 +69,26 @@ public class NotifyTest {
     stmt.close();
   }
 
-  @Test
+  @Test(timeout = 60000)
   public void testAsyncNotify() throws Exception {
     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();
-    }
+    connectAndNotify("mynotification");
 
-    // Wait a bit to let the notify come through..
+    // Wait a bit to let the notify come through... Changed this so the test takes ~2 seconds
+    // less to run and is still as effective.
+    PGNotification notifications[] = null;
     try {
-      Thread.sleep(2000);
+      int retries = 20;
+      while (retries-- > 0
+        && (notifications = ((org.postgresql.PGConnection) conn).getNotifications()) == null ) {
+        Thread.sleep(100);
+      }
     } catch (InterruptedException ie) {
     }
 
-    PGNotification notifications[] = ((org.postgresql.PGConnection) conn).getNotifications();
     assertNotNull(notifications);
     assertEquals(1, notifications.length);
     assertEquals("mynotification", notifications[0].getName());
@@ -96,4 +96,160 @@ public class NotifyTest {
 
     stmt.close();
   }
+
+  /**
+   * To test timeouts we have to send the notification from another thread, because we
+   * listener is blocking.
+   */
+  @Test(timeout = 60000)
+  public void testAsyncNotifyWithTimeout() throws Exception {
+    Statement stmt = conn.createStatement();
+    stmt.executeUpdate("LISTEN mynotification");
+
+    // Here we let the getNotifications() timeout.
+    long startMillis = System.currentTimeMillis();
+    PGNotification notifications[] = ((org.postgresql.PGConnection) conn).getNotifications(500);
+    long endMillis = System.currentTimeMillis();
+    long runtime = endMillis - startMillis;
+    assertNull("There have been notifications, althought none have been expected.",notifications);
+    Assert.assertTrue("We didn't wait long enough! runtime=" + runtime, runtime > 450);
+
+    stmt.close();
+  }
+
+  @Test(timeout = 60000)
+  public void testAsyncNotifyWithTimeoutAndMessagesAvailableWhenStartingListening() throws Exception {
+    Statement stmt = conn.createStatement();
+    stmt.executeUpdate("LISTEN mynotification");
+
+    // Now we check the case where notifications are already available while we are starting to
+    // listen for notifications
+    connectAndNotify("mynotification");
+
+    PGNotification notifications[] = ((org.postgresql.PGConnection) conn).getNotifications(10000);
+    assertNotNull(notifications);
+    assertEquals(1, notifications.length);
+    assertEquals("mynotification", notifications[0].getName());
+    assertEquals("", notifications[0].getParameter());
+
+    stmt.close();
+  }
+
+  @Test(timeout = 60000)
+  public void testAsyncNotifyWithEndlessTimeoutAndMessagesAvailableWhenStartingListening() throws Exception {
+    Statement stmt = conn.createStatement();
+    stmt.executeUpdate("LISTEN mynotification");
+
+    // Now we check the case where notifications are already available while we are waiting forever
+    connectAndNotify("mynotification");
+
+    PGNotification notifications[] = ((org.postgresql.PGConnection) conn).getNotifications(0);
+    assertNotNull(notifications);
+    assertEquals(1, notifications.length);
+    assertEquals("mynotification", notifications[0].getName());
+    assertEquals("", notifications[0].getParameter());
+
+    stmt.close();
+  }
+
+  @Test(timeout = 60000)
+  public void testAsyncNotifyWithTimeoutAndMessagesSendAfter() throws Exception {
+    Statement stmt = conn.createStatement();
+    stmt.executeUpdate("LISTEN mynotification");
+
+    // Now we check the case where notifications are send after we have started to listen for
+    // notifications
+    new Thread( new Runnable() {
+      public void run() {
+        try {
+          Thread.sleep(200);
+        } catch (InterruptedException ie) {
+        }
+        connectAndNotify("mynotification");
+      }
+    }).start();
+
+    PGNotification notifications[] = ((org.postgresql.PGConnection) conn).getNotifications(10000);
+    assertNotNull(notifications);
+    assertEquals(1, notifications.length);
+    assertEquals("mynotification", notifications[0].getName());
+    assertEquals("", notifications[0].getParameter());
+
+    stmt.close();
+  }
+
+  @Test(timeout = 60000)
+  public void testAsyncNotifyWithEndlessTimeoutAndMessagesSendAfter() throws Exception {
+    Statement stmt = conn.createStatement();
+    stmt.executeUpdate("LISTEN mynotification");
+
+    // Now we check the case where notifications are send after we have started to listen for
+    // notifications forever
+    new Thread( new Runnable() {
+      public void run() {
+        try {
+          Thread.sleep(200);
+        } catch (InterruptedException ie) {
+        }
+        connectAndNotify("mynotification");
+      }
+    }).start();
+
+    PGNotification notifications[] = ((org.postgresql.PGConnection) conn).getNotifications(0);
+    assertNotNull(notifications);
+    assertEquals(1, notifications.length);
+    assertEquals("mynotification", notifications[0].getName());
+    assertEquals("", notifications[0].getParameter());
+
+    stmt.close();
+  }
+
+  @Test(timeout = 60000)
+  public void testAsyncNotifyWithTimeoutAndSocketThatBecomesClosed() throws Exception {
+    Statement stmt = conn.createStatement();
+    stmt.executeUpdate("LISTEN mynotification");
+
+    // Here we check what happens when the connection gets closed from another thread. This
+    // should be able, and this test ensures that no synchronized statements will stop the
+    // connection from becoming closed.
+    new Thread( new Runnable() {
+      public void run() {
+        try {
+          Thread.sleep(500);
+        } catch (InterruptedException ie) {
+        }
+        try {
+          conn.close();
+        } catch (SQLException e) {
+        }
+      }
+    }).start();
+
+    try {
+      ((org.postgresql.PGConnection) conn).getNotifications(40000);
+      Assert.fail("The getNotifications(...) call didn't return when the socket closed.");
+    } catch (SQLException e) {
+      // We expected thta
+    }
+
+    stmt.close();
+  }
+
+  private static void connectAndNotify(String channel) {
+    Connection conn2 = null;
+    try {
+      conn2 = TestUtil.openDB();
+      Statement stmt2 = conn2.createStatement();
+      stmt2.executeUpdate("NOTIFY " + channel);
+      stmt2.close();
+    } catch (Exception e) {
+      throw new RuntimeException("Couldn't notify '" + channel + "'.",e);
+    } finally {
+      try {
+        conn2.close();
+      } catch (SQLException e) {
+      }
+    }
+  }
+
 }