@MaxConnections(1) @Test(timeout = 60000) public void testPoolExhaustion() throws Exception { thrown.expectCause(isA(PoolExhaustedException.class)); pooledFactoryRule.getAConnection(); pooledFactoryRule.getProvider().newConnectionRequest().toBlocking().single(); }
@Override public Observable<Connection<R, W>> newConnectionRequest() { return Observable.create(new OnSubscribe<Connection<R, W>>() { @Override public void call(Subscriber<? super Connection<R, W>> subscriber) { if (isShutdown) { subscriber.onError(new IllegalStateException("Connection provider is shutdown.")); } idleConnectionsHolder.pollThisEventLoopConnections() .concatWith(connectIfAllowed()) .filter(new Func1<PooledConnection<R, W>, Boolean>() { @Override public Boolean call(PooledConnection<R, W> c) { boolean isUsable = c.isUsable(); if (!isUsable) { discardNow(c); } return isUsable; } }) .take(1) .lift(new ReuseSubscriberLinker()) .lift(new ConnectMetricsOperator()) .unsafeSubscribe(subscriber); } }); }
PooledConnectionProviderImpl extends PooledConnectionProvider<W, R> { @Override public Observable<Connection<R, W>> newConnectionRequest() { return Observable.create(new OnSubscribe<Connection<R, W>>() { @Override public void call(Subscriber<? super Connection<R, W>> subscriber) { if (isShutdown) { subscriber.onError(new IllegalStateException("Connection provider is shutdown.")); } idleConnectionsHolder.pollThisEventLoopConnections() .concatWith(connectIfAllowed()) .filter(new Func1<PooledConnection<R, W>, Boolean>() { @Override public Boolean call(PooledConnection<R, W> c) { boolean isUsable = c.isUsable(); if (!isUsable) { discardNow(c); } return isUsable; } }) .take(1) .lift(new ReuseSubscriberLinker()) .lift(new ConnectMetricsOperator()) .unsafeSubscribe(subscriber); } }); } }
PooledConnectionProviderImpl extends PooledConnectionProvider<W, R> { @Override public Observable<Connection<R, W>> newConnectionRequest() { return Observable.create(new OnSubscribe<Connection<R, W>>() { @Override public void call(Subscriber<? super Connection<R, W>> subscriber) { if (isShutdown) { subscriber.onError(new IllegalStateException("Connection provider is shutdown.")); } idleConnectionsHolder.pollThisEventLoopConnections() .concatWith(connectIfAllowed()) .filter(new Func1<PooledConnection<R, W>, Boolean>() { @Override public Boolean call(PooledConnection<R, W> c) { boolean isUsable = c.isUsable(); if (!isUsable) { discardNow(c); } return isUsable; } }) .take(1) .lift(new ReuseSubscriberLinker()) .lift(new ConnectMetricsOperator()) .unsafeSubscribe(subscriber); } }); } PooledConnectionProviderImpl(PoolConfig<W, R> poolConfig, HostConnector<W, R> hostConnector); }
PooledConnectionProviderImpl extends PooledConnectionProvider<W, R> { @Override public Observable<Connection<R, W>> newConnectionRequest() { return Observable.create(new OnSubscribe<Connection<R, W>>() { @Override public void call(Subscriber<? super Connection<R, W>> subscriber) { if (isShutdown) { subscriber.onError(new IllegalStateException("Connection provider is shutdown.")); } idleConnectionsHolder.pollThisEventLoopConnections() .concatWith(connectIfAllowed()) .filter(new Func1<PooledConnection<R, W>, Boolean>() { @Override public Boolean call(PooledConnection<R, W> c) { boolean isUsable = c.isUsable(); if (!isUsable) { discardNow(c); } return isUsable; } }) .take(1) .lift(new ReuseSubscriberLinker()) .lift(new ConnectMetricsOperator()) .unsafeSubscribe(subscriber); } }); } PooledConnectionProviderImpl(PoolConfig<W, R> poolConfig, HostConnector<W, R> hostConnector); @Override Observable<Connection<R, W>> newConnectionRequest(); @Override Observable<Void> release(final PooledConnection<?, ?> connection); @Override Observable<Void> discard(final PooledConnection<?, ?> connection); }
PooledConnectionProviderImpl extends PooledConnectionProvider<W, R> { @Override public Observable<Connection<R, W>> newConnectionRequest() { return Observable.create(new OnSubscribe<Connection<R, W>>() { @Override public void call(Subscriber<? super Connection<R, W>> subscriber) { if (isShutdown) { subscriber.onError(new IllegalStateException("Connection provider is shutdown.")); } idleConnectionsHolder.pollThisEventLoopConnections() .concatWith(connectIfAllowed()) .filter(new Func1<PooledConnection<R, W>, Boolean>() { @Override public Boolean call(PooledConnection<R, W> c) { boolean isUsable = c.isUsable(); if (!isUsable) { discardNow(c); } return isUsable; } }) .take(1) .lift(new ReuseSubscriberLinker()) .lift(new ConnectMetricsOperator()) .unsafeSubscribe(subscriber); } }); } PooledConnectionProviderImpl(PoolConfig<W, R> poolConfig, HostConnector<W, R> hostConnector); @Override Observable<Connection<R, W>> newConnectionRequest(); @Override Observable<Void> release(final PooledConnection<?, ?> connection); @Override Observable<Void> discard(final PooledConnection<?, ?> connection);  }