Merge pull request #615 from FortnoxAB/patch-4

Interval instead of timer for the idleConnectionCleanup so that it keep cleaning up. Add retry so that a failure during cleanup will not stop future cleanups

Author: jamesgorman2

rxnetty-common/src/main/java/io/reactivex/netty/client/pool/PoolConfig.java

@@ -37,7 +37,7 @@ public class PoolConfig<W, R> {
 
     public PoolConfig() {
         maxIdleTimeMillis = DEFAULT_MAX_IDLE_TIME_MILLIS;
-        idleConnCleanupTicker = Observable.timer(maxIdleTimeMillis, TimeUnit.MILLISECONDS);
+        idleConnCleanupTicker = Observable.interval(maxIdleTimeMillis, maxIdleTimeMillis, TimeUnit.MILLISECONDS);
         idleConnectionsHolder = new FIFOIdleConnectionsHolder<>();
         limitDeterminationStrategy = UnboundedPoolLimitDeterminationStrategy.INSTANCE;
     }

rxnetty-common/src/main/java/io/reactivex/netty/client/pool/PooledConnectionProviderImpl.java

@@ -70,18 +70,19 @@ public PooledConnectionProviderImpl(PoolConfig<W, R> poolConfig, HostConnector<W
         limitDeterminationStrategy = poolConfig.getPoolLimitDeterminationStrategy();
         maxIdleTimeMillis = poolConfig.getMaxIdleTimeMillis();
         // In case, there is no cleanup required, this observable should never give a tick.
-        idleConnCleanupSubscription = poolConfig.getIdleConnectionsCleanupTimer()
-                                                .doOnError(LogErrorAction.INSTANCE)
-                .retry() // Retry when there is an error in timer.
-                .concatMap(new IdleConnectionCleanupTask())
-                .onErrorResumeNext(new Func1<Throwable, Observable<Void>>() {
-                    @Override
-                    public Observable<Void> call(Throwable throwable) {
-                        logger.error("Ignoring error cleaning up idle connections.",
-                                     throwable);
-                        return Observable.empty();
-                    }
-                }).subscribe(Actions.empty()); // Errors are logged and ignored.
+        idleConnCleanupSubscription = poolConfig.getIdleConnCleanupTicker()
+            .doOnError(LogErrorAction.INSTANCE)
+            .retry() // Retry when there is an error in timer.
+            .concatMap(new IdleConnectionCleanupTask())
+            .doOnError(new Action1<Throwable>() {
+                @Override
+                public void call(Throwable throwable) {
+                    logger.error("Ignoring error cleaning up idle connections.",
+                        throwable);
+                }
+            })
+            .retry()
+            .subscribe();
 
         hostConnector.getHost()
                      .getCloseNotifier()