You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by ma...@apache.org on 2019/07/01 12:58:51 UTC
[james-project] 02/02: JAMES-2774 Close borrowed Channels in
ReactorRabbitMQChannelPool
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 54240a360fcdefa36e555ba1f91e00f03b4a2944
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Thu Jun 27 11:34:58 2019 +0200
JAMES-2774 Close borrowed Channels in ReactorRabbitMQChannelPool
---
.../mailbox/events/ReactorRabbitMQChannelPool.java | 16 ++++++++---
.../events/ReactorRabbitMQChannelPoolTest.java | 31 +++++++++++++++++++---
2 files changed, 40 insertions(+), 7 deletions(-)
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/ReactorRabbitMQChannelPool.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/ReactorRabbitMQChannelPool.java
index 121ccc6..19ddf7e 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/ReactorRabbitMQChannelPool.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/ReactorRabbitMQChannelPool.java
@@ -20,6 +20,8 @@
package org.apache.james.mailbox.events;
import java.time.Duration;
+import java.util.Comparator;
+import java.util.concurrent.ConcurrentSkipListSet;
import java.util.function.BiConsumer;
import org.apache.commons.pool2.BasePooledObjectFactory;
@@ -86,24 +88,30 @@ class ReactorRabbitMQChannelPool implements ChannelPool {
private static final long MAXIMUM_BORROW_TIMEOUT_IN_MS = Duration.ofSeconds(5).toMillis();
private final GenericObjectPool<Channel> pool;
- private final ChannelFactory channelFactory;
+ private final ConcurrentSkipListSet<Channel> borrowedChannels;
ReactorRabbitMQChannelPool(Mono<Connection> connectionMono, int poolSize) {
- this.channelFactory = new ChannelFactory(connectionMono);
+ ChannelFactory channelFactory = new ChannelFactory(connectionMono);
GenericObjectPoolConfig<Channel> config = new GenericObjectPoolConfig<>();
config.setMaxTotal(poolSize);
this.pool = new GenericObjectPool<>(channelFactory, config);
+ this.borrowedChannels = new ConcurrentSkipListSet<>(Comparator.comparingInt(System::identityHashCode));
}
@Override
public Mono<? extends Channel> getChannelMono() {
- return Mono.fromCallable(() -> pool.borrowObject(MAXIMUM_BORROW_TIMEOUT_IN_MS));
+ return Mono.fromCallable(() -> {
+ Channel channel = pool.borrowObject(MAXIMUM_BORROW_TIMEOUT_IN_MS);
+ borrowedChannels.add(channel);
+ return channel;
+ });
}
@Override
public BiConsumer<SignalType, Channel> getChannelCloseHandler() {
return (signalType, channel) -> {
+ borrowedChannels.remove(channel);
if (!channel.isOpen() || signalType != SignalType.ON_COMPLETE) {
invalidateObject(channel);
return;
@@ -125,6 +133,8 @@ class ReactorRabbitMQChannelPool implements ChannelPool {
@Override
public void close() {
+ borrowedChannels.forEach(channel -> getChannelCloseHandler().accept(SignalType.ON_NEXT, channel));
+ borrowedChannels.clear();
pool.close();
}
}
diff --git a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/ReactorRabbitMQChannelPoolTest.java b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/ReactorRabbitMQChannelPoolTest.java
index 9adccb6..0295f5f 100644
--- a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/ReactorRabbitMQChannelPoolTest.java
+++ b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/ReactorRabbitMQChannelPoolTest.java
@@ -19,6 +19,7 @@
package org.apache.james.mailbox.events;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.time.Duration;
@@ -38,6 +39,7 @@ import com.github.fge.lambdas.Throwing;
import com.rabbitmq.client.Channel;
import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
import reactor.rabbitmq.ChannelPool;
class ReactorRabbitMQChannelPoolTest implements ChannelPoolContract {
@@ -59,14 +61,18 @@ class ReactorRabbitMQChannelPoolTest implements ChannelPoolContract {
@Override
public ChannelPool getChannelPool(int poolSize) {
- ReactorRabbitMQChannelPool channelPool = new ReactorRabbitMQChannelPool(
- rabbitMQExtension.getRabbitConnectionPool().getResilientConnection(),
- poolSize);
+ ReactorRabbitMQChannelPool channelPool = generateChannelPool(poolSize);
channelPools.add(channelPool);
return channelPool;
}
+ private ReactorRabbitMQChannelPool generateChannelPool(int poolSize) {
+ return new ReactorRabbitMQChannelPool(
+ rabbitMQExtension.getRabbitConnectionPool().getResilientConnection(),
+ poolSize);
+ }
+
// Pool wait timeout is an expected exception
@Test
void concurrentRequestOnChannelMonoLeadToPoolWaitTimeoutException() {
@@ -81,7 +87,24 @@ class ReactorRabbitMQChannelPoolTest implements ChannelPoolContract {
.runSuccessfullyWithin(Duration.ofSeconds(30)))
.isInstanceOf(ExecutionException.class)
.hasMessageContaining("java.util.NoSuchElementException: Timeout waiting for idle object");
+ }
- listChannels.forEach(Throwing.consumer(Channel::close));
+ @Test
+ void usedChannelShouldBeClosedWhenPoolIsClosed() {
+ ChannelPool channelPool = generateChannelPool(2);
+ Channel channel = channelPool.getChannelMono().block();
+ assertThat(channel.isOpen()).isTrue();
+ channelPool.close();
+ assertThat(channel.isOpen()).isFalse();
+ }
+
+ @Test
+ void notUsedChannelShouldBeClosedWhenPoolIsClosed() {
+ ChannelPool channelPool = generateChannelPool(2);
+ Channel channel = channelPool.getChannelMono().block();
+ assertThat(channel.isOpen()).isTrue();
+ channelPool.getChannelCloseHandler().accept(SignalType.ON_NEXT, channel);
+ channelPool.close();
+ assertThat(channel.isOpen()).isFalse();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org