@Test(timeout = 60000) public void testConnectFailed() throws Exception { PooledConnectionProvider<String, String> factory; PoolConfig<String, String> config = new PoolConfig<>(); config.idleConnectionsHolder(pooledFactoryRule.holder); MockEventPublisher<ClientEventListener> publisher = MockEventPublisher.disabled(); ClientEventListener listener = new ClientEventListener(); EmbeddedConnectionProvider connectionProvider = new EmbeddedConnectionProvider(publisher, true, listener); Host host = new Host(new InetSocketAddress("127.0.0.1", 0)); HostConnector<String, String> connector = new HostConnector<>(host, connectionProvider, publisher, publisher, listener); factory = new PooledConnectionProviderImpl<>(config, connector); TestSubscriber<Object> subscriber = new TestSubscriber<>(); factory.newConnectionRequest().subscribe(subscriber); subscriber.assertTerminalEvent(); assertThat("Error not returned to connect.", subscriber.getOnCompletedEvents(), is(empty())); assertThat("Error not returned to connect.", subscriber.getOnNextEvents(), is(empty())); }
@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);  }