Re-establish LISTEN subscription on error.

alaisi · alaisi · commit 5e1258509e4e · 2015-09-30T22:40:57.000+03:00
diff --git a/src/main/java/com/github/pgasync/impl/PgConnectionPool.java b/src/main/java/com/github/pgasync/impl/PgConnectionPool.java
@@ -23,9 +23,7 @@
 
 import java.net.InetSocketAddress;
 import java.util.LinkedList;
-import java.util.Map;
 import java.util.Queue;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -43,8 +41,6 @@ public abstract class PgConnectionPool implements ConnectionPool {
     final Queue<Connection> connections = new LinkedList<>();
     final Object lock = new Object[0];
 
-    final Map<String,Connection> listeners = new ConcurrentHashMap<>();
-
     final InetSocketAddress address;
     final String username;
     final String password;
@@ -98,8 +94,9 @@ public Observable<String> listen(String channel) {
         return getConnection()
                 .lift(subscriber -> Subscribers.create(
                         connection -> connection.listen(channel)
-                                .doOnSubscribe(() -> release(connection))
-                                .subscribe(subscriber),
+                                        .doOnSubscribe(() -> release(connection))
+                                        .onErrorResumeNext(exception -> listen(channel))
+                                        .subscribe(subscriber),
                         subscriber::onError));
     }
 
