@Test(timeout = 60000) public void testMetricEventCallback() throws Throwable { TrackableMetricEventsListener eventsListener = new TrackableMetricEventsListener(); pooledFactoryRule.init(DEFAULT_MAX_CONNECTIONS, MockEventPublisher.<ClientEventListener>enabled(), MockEventPublisher.enabled(), eventsListener); final PooledConnection<String, String> connection = pooledFactoryRule.getAConnection(); assertThat("Unexpected acquire attempted count.", eventsListener.getAcquireAttemptedCount(), is(1L)); assertThat("Unexpected acquire succedded count.", eventsListener.getAcquireSucceededCount(), is(1L)); assertThat("Unexpected acquire failed count.", eventsListener.getAcquireFailedCount(), is(0L)); pooledFactoryRule.returnToIdle(connection); assertThat("Unexpected release attempted count.", eventsListener.getReleaseAttemptedCount(), is(1L)); assertThat("Unexpected release succeeded count.", eventsListener.getReleaseSucceededCount(), is(1L)); assertThat("Unexpected release failed count.", eventsListener.getReleaseFailedCount(), is(0L)); final PooledConnection<String, String> reusedConn = pooledFactoryRule.getAConnection(); Assert.assertEquals("Reused connection not same as original.", connection, reusedConn); assertThat("Unexpected acquire attempted count.", eventsListener.getAcquireAttemptedCount(), is(2L)); assertThat("Unexpected acquire succedded count.", eventsListener.getAcquireSucceededCount(), is(2L)); assertThat("Unexpected acquire failed count.", eventsListener.getAcquireFailedCount(), is(0L)); assertThat("Unexpected reuse count.", eventsListener.getReuseCount(), is(1L)); pooledFactoryRule.closeAndAwait(reusedConn); assertThat("Unexpected release attempted count.", eventsListener.getReleaseAttemptedCount(), is(2L)); assertThat("Unexpected release succeeded count.", eventsListener.getReleaseSucceededCount(), is(2L)); assertThat("Unexpected release failed count.", eventsListener.getReleaseFailedCount(), is(0L)); pooledFactoryRule.provider.discard(reusedConn).toBlocking().lastOrDefault(null); assertThat("Unexpected release attempted count.", eventsListener.getReleaseAttemptedCount(), is(2L)); assertThat("Unexpected release succeeded count.", eventsListener.getReleaseSucceededCount(), is(2L)); assertThat("Unexpected release failed count.", eventsListener.getReleaseFailedCount(), is(0L)); assertThat("Unexpected connection eviction count.", eventsListener.getEvictionCount(), is(2L)); }
@Override public Observable<Void> discard(final PooledConnection<?, ?> connection) { return connection.discard().doOnSubscribe(new Action0() { @Override public void call() { EventPublisher eventPublisher = connection.unsafeNettyChannel().attr(EVENT_PUBLISHER).get(); if (eventPublisher.publishingEnabled()) { ClientEventListener eventListener = connection.unsafeNettyChannel() .attr(CLIENT_EVENT_LISTENER).get(); eventListener.onPooledConnectionEviction(); } limitDeterminationStrategy.releasePermit(); } }); }
PooledConnectionProviderImpl extends PooledConnectionProvider<W, R> { @Override public Observable<Void> discard(final PooledConnection<?, ?> connection) { return connection.discard().doOnSubscribe(new Action0() { @Override public void call() { EventPublisher eventPublisher = connection.unsafeNettyChannel().attr(EVENT_PUBLISHER).get(); if (eventPublisher.publishingEnabled()) { ClientEventListener eventListener = connection.unsafeNettyChannel() .attr(CLIENT_EVENT_LISTENER).get(); eventListener.onPooledConnectionEviction(); } limitDeterminationStrategy.releasePermit(); } }); } }
PooledConnectionProviderImpl extends PooledConnectionProvider<W, R> { @Override public Observable<Void> discard(final PooledConnection<?, ?> connection) { return connection.discard().doOnSubscribe(new Action0() { @Override public void call() { EventPublisher eventPublisher = connection.unsafeNettyChannel().attr(EVENT_PUBLISHER).get(); if (eventPublisher.publishingEnabled()) { ClientEventListener eventListener = connection.unsafeNettyChannel() .attr(CLIENT_EVENT_LISTENER).get(); eventListener.onPooledConnectionEviction(); } limitDeterminationStrategy.releasePermit(); } }); } PooledConnectionProviderImpl(PoolConfig<W, R> poolConfig, HostConnector<W, R> hostConnector); }
PooledConnectionProviderImpl extends PooledConnectionProvider<W, R> { @Override public Observable<Void> discard(final PooledConnection<?, ?> connection) { return connection.discard().doOnSubscribe(new Action0() { @Override public void call() { EventPublisher eventPublisher = connection.unsafeNettyChannel().attr(EVENT_PUBLISHER).get(); if (eventPublisher.publishingEnabled()) { ClientEventListener eventListener = connection.unsafeNettyChannel() .attr(CLIENT_EVENT_LISTENER).get(); eventListener.onPooledConnectionEviction(); } limitDeterminationStrategy.releasePermit(); } }); } 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<Void> discard(final PooledConnection<?, ?> connection) { return connection.discard().doOnSubscribe(new Action0() { @Override public void call() { EventPublisher eventPublisher = connection.unsafeNettyChannel().attr(EVENT_PUBLISHER).get(); if (eventPublisher.publishingEnabled()) { ClientEventListener eventListener = connection.unsafeNettyChannel() .attr(CLIENT_EVENT_LISTENER).get(); eventListener.onPooledConnectionEviction(); } limitDeterminationStrategy.releasePermit(); } }); } 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);  }