Close connections cleanly.

alaisi · alaisi · commit 8dfa5f32cb78 · 2015-11-26T21:53:12.000+02:00
diff --git a/src/main/java/com/github/pgasync/impl/PgConnection.java b/src/main/java/com/github/pgasync/impl/PgConnection.java
@@ -90,9 +90,7 @@ public Observable<String> listen(String channel) {
 
     @Override
     public Observable<Void> close() {
-        return Observable.concat(stream.send(Terminate.INSTANCE), stream.close())
-                .first()
-                .map(__ -> null);
+        return stream.close();
     }
 
     private Observable<Message> sendQuery(String sql, Object[] params) {
diff --git a/src/main/java/com/github/pgasync/impl/PgConnectionPool.java b/src/main/java/com/github/pgasync/impl/PgConnectionPool.java
@@ -24,9 +24,11 @@
 import java.net.InetSocketAddress;
 import java.util.LinkedList;
 import java.util.Queue;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.logging.Logger;
 import javax.annotation.concurrent.GuardedBy;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -109,6 +111,7 @@ public Observable<String> listen(String channel) {
     @Override
     public Observable<Void> close() {
         return Observable.create(subscriber -> {
+
             lock.lock();
             try {
                 closed = true;
@@ -132,7 +135,11 @@ public Observable<Void> close() {
                             continue;
                         }
                         currentSize--;
-                        connection.close().toBlocking().single();
+                        CountDownLatch wait = new CountDownLatch(1);
+                        connection.close().subscribe(__ -> wait.countDown(), __ -> wait.countDown());
+                        if(!wait.await(5, SECONDS)) {
+                            Logger.getLogger(getClass().getName()).warning("Closing connection timed out");
+                        }
                     }
                 } catch (InterruptedException e) { /* ignore */ }
 
diff --git a/src/main/java/com/github/pgasync/impl/netty/NettyPgConnectionPool.java b/src/main/java/com/github/pgasync/impl/netty/NettyPgConnectionPool.java
@@ -62,7 +62,7 @@ public Observable<Void> close() {
 
     private <T> GenericFutureListener<Future<T>> close(Subscriber<?> subscriber, Throwable exception) {
         return f -> {
-            if (exception == null && !f.isSuccess()) {
+            if (exception == null && f.isSuccess()) {
                 subscriber.onNext(null);
                 subscriber.onCompleted();
                 return;
diff --git a/src/test/java/com/github/pgasync/impl/DatabaseRule.java b/src/test/java/com/github/pgasync/impl/DatabaseRule.java
@@ -12,6 +12,7 @@
 import java.util.Map.Entry;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -40,7 +41,14 @@ protected void before() {
     @Override
     protected void after() {
         if(pool != null) {
-            pool.close();
+            CountDownLatch latch = new CountDownLatch(1);
+            pool.close().subscribe(__ -> latch.countDown(), ex -> {
+                ex.printStackTrace();
+                latch.countDown();
+            });
+            try {
+                latch.await(30, TimeUnit.SECONDS);
+            } catch (InterruptedException e) { /* ignore */ }
         }
     }
 
diff --git a/src/test/java/com/github/pgasync/impl/ListenNotifyTest.java b/src/test/java/com/github/pgasync/impl/ListenNotifyTest.java
@@ -44,6 +44,6 @@ public void shouldReceiveNotificationsOnListenedChannel() throws Exception {
 
     @AfterClass
     public static void close() {
-        pool.close();
+        pool.close().toBlocking().single();
     }
 }
diff --git a/src/test/java/com/github/pgasync/impl/PerformanceTest.java b/src/test/java/com/github/pgasync/impl/PerformanceTest.java
@@ -90,7 +90,7 @@ public PerformanceTest(int poolSize, int numThreads, boolean pipeline) {
     @After
     public void close() {
         threadPool.shutdownNow();
-        dbPool.close();
+        dbPool.close().toBlocking().single();
     }
 
     @Test(timeout = 1000)
diff --git a/src/test/java/com/github/pgasync/impl/PipelineTest.java b/src/test/java/com/github/pgasync/impl/PipelineTest.java
@@ -52,7 +52,7 @@ public void closeConnection() {
             pool.release(c);
         }
         if (pool != null) {
-            pool.close();
+            pool.close().toBlocking().single();
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -90,9 +90,7 @@ public Observable<String> listen(String channel) {`
`90`	`90`
`91`	`91`	`@Override`
`92`	`92`	`public Observable<Void> close() {`
`93`		`- return Observable.concat(stream.send(Terminate.INSTANCE), stream.close())`
`94`		`- .first()`
`95`		`- .map(__ -> null);`
	`93`	`+ return stream.close();`
`96`	`94`	`}`
`97`	`95`
`98`	`96`	`private Observable<Message> sendQuery(String sql, Object[] params) {`
Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,7 @@ public Observable<Void> close() {`
`62`	`62`
`63`	`63`	`private <T> GenericFutureListener<Future<T>> close(Subscriber<?> subscriber, Throwable exception) {`
`64`	`64`	`return f -> {`
`65`		`- if (exception == null && !f.isSuccess()) {`
	`65`	`+ if (exception == null && f.isSuccess()) {`
`66`	`66`	`subscriber.onNext(null);`
`67`	`67`	`subscriber.onCompleted();`
`68`	`68`	`return;`
Original file line number	Diff line number	Diff line change
`@@ -44,6 +44,6 @@ public void shouldReceiveNotificationsOnListenedChannel() throws Exception {`
`44`	`44`
`45`	`45`	`@AfterClass`
`46`	`46`	`public static void close() {`
`47`		`- pool.close();`
	`47`	`+ pool.close().toBlocking().single();`
`48`	`48`	`}`
`49`	`49`	`}`
Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ public PerformanceTest(int poolSize, int numThreads, boolean pipeline) {`
`90`	`90`	`@After`
`91`	`91`	`public void close() {`
`92`	`92`	`threadPool.shutdownNow();`
`93`		`- dbPool.close();`
	`93`	`+ dbPool.close().toBlocking().single();`
`94`	`94`	`}`
`95`	`95`
`96`	`96`	`@Test(timeout = 1000)`
Original file line number	Diff line number	Diff line change
`@@ -52,7 +52,7 @@ public void closeConnection() {`
`52`	`52`	`pool.release(c);`
`53`	`53`	`}`
`54`	`54`	`if (pool != null) {`
`55`		`- pool.close();`
	`55`	`+ pool.close().toBlocking().single();`
`56`	`56`	`}`
`57`	`57`	`}`
`58`	`58`