james-server-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From btell...@apache.org
Subject [2/2] james-project git commit: JAMES-2544 Temporary fix RabbitMQ concurrency issues using a global lock
Date Fri, 05 Oct 2018 02:42:49 GMT
JAMES-2544 Temporary fix RabbitMQ concurrency issues using a global lock

We lower the amount of mail sent to avoid overwhelming Cassandra


Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/7ed58e9e
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/7ed58e9e
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/7ed58e9e

Branch: refs/heads/master
Commit: 7ed58e9e737e290e3033e3706156ba5fc5d18bfa
Parents: 6bfbc8e
Author: Benoit Tellier <btellier@linagora.com>
Authored: Thu Oct 4 15:49:01 2018 +0700
Committer: Benoit Tellier <btellier@linagora.com>
Committed: Fri Oct 5 09:42:10 2018 +0700

----------------------------------------------------------------------
 .../backend/rabbitmq/RabbitChannelPool.java     | 130 -------------------
 .../backend/rabbitmq/RabbitChannelPoolImpl.java | 115 ++++++++++++++++
 .../backend/rabbitmq/RabbitMQChannelPool.java   |  47 +++++++
 .../backend/rabbitmq/RabbitMQHealthCheck.java   |   8 +-
 .../backend/rabbitmq/SimpleChannelPool.java     |  42 ++++++
 .../backend/rabbitmq/RabbitMQExtension.java     |  14 +-
 .../rabbitmq/RabbitMQHealthCheckTest.java       |   2 +-
 .../james/queue/api/MailQueueContract.java      |   8 +-
 .../james/queue/rabbitmq/RabbitClient.java      |  12 +-
 .../queue/rabbitmq/RabbitMQMailQueueTest.java   |  18 +--
 10 files changed, 228 insertions(+), 168 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitChannelPool.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitChannelPool.java
b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitChannelPool.java
deleted file mode 100644
index 66ea5b9..0000000
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitChannelPool.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/****************************************************************
- * Licensed to the Apache Software Foundation (ASF) under one   *
- * or more contributor license agreements.  See the NOTICE file *
- * distributed with this work for additional information        *
- * regarding copyright ownership.  The ASF licenses this file   *
- * to you under the Apache License, Version 2.0 (the            *
- * "License"); you may not use this file except in compliance   *
- * with the License.  You may obtain a copy of the License at   *
- *                                                              *
- *   http://www.apache.org/licenses/LICENSE-2.0                 *
- *                                                              *
- * Unless required by applicable law or agreed to in writing,   *
- * software distributed under the License is distributed on an  *
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
- * KIND, either express or implied.  See the License for the    *
- * specific language governing permissions and limitations      *
- * under the License.                                           *
- ****************************************************************/
-
-package org.apache.james.backend.rabbitmq;
-
-import java.util.function.Supplier;
-
-import javax.annotation.PreDestroy;
-import javax.inject.Inject;
-
-import org.apache.commons.pool2.BasePooledObjectFactory;
-import org.apache.commons.pool2.ObjectPool;
-import org.apache.commons.pool2.PooledObject;
-import org.apache.commons.pool2.impl.DefaultPooledObject;
-import org.apache.commons.pool2.impl.GenericObjectPool;
-import org.apache.james.util.MemoizedSupplier;
-
-import com.github.fge.lambdas.Throwing;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-
-public class RabbitChannelPool {
-
-    public static class ConnectionFailedException extends RuntimeException {
-        public ConnectionFailedException(Throwable cause) {
-            super(cause);
-        }
-    }
-
-    private static class ChannelBasePooledObjectFactory extends BasePooledObjectFactory<Channel>
{
-        private final Supplier<Connection> connection;
-
-        public ChannelBasePooledObjectFactory(RabbitMQConnectionFactory factory) {
-            this.connection = MemoizedSupplier.of(
-                    Throwing.supplier(() -> factory.create()).sneakyThrow());
-        }
-
-        @Override
-        public Channel create() throws Exception {
-            return connection.get()
-                    .createChannel();
-        }
-
-        @Override
-        public PooledObject<Channel> wrap(Channel obj) {
-            return new DefaultPooledObject<>(obj);
-        }
-
-        @Override
-        public void destroyObject(PooledObject<Channel> pooledObject) throws Exception
{
-            Channel channel = pooledObject.getObject();
-            channel.close();
-        }
-    }
-
-    @FunctionalInterface
-    public interface RabbitFunction<T, E extends Throwable> {
-        T execute(Channel channel) throws E;
-    }
-
-    @FunctionalInterface
-    public interface RabbitConsumer<E extends Throwable> {
-        void execute(Channel channel) throws E;
-    }
-
-    private final ObjectPool<Channel> pool;
-
-    @Inject
-    public RabbitChannelPool(RabbitMQConnectionFactory factory) {
-        pool = new GenericObjectPool<>(
-            new ChannelBasePooledObjectFactory(factory));
-    }
-
-    public <T, E extends Throwable> T execute(RabbitFunction<T, E> f) throws
E, ConnectionFailedException {
-        Channel channel = borrowChannel();
-        try {
-            return f.execute(channel);
-        } finally {
-            returnChannel(channel);
-        }
-    }
-
-
-    public <E extends Throwable> void execute(RabbitConsumer<E> f) throws E,
ConnectionFailedException {
-        Channel channel = borrowChannel();
-        try {
-            f.execute(channel);
-        } finally {
-            returnChannel(channel);
-        }
-    }
-
-    @PreDestroy
-    public void close() {
-        pool.close();
-    }
-
-    private Channel borrowChannel() {
-        try {
-            return pool.borrowObject();
-        } catch (Exception e) {
-            throw new ConnectionFailedException(e);
-        }
-    }
-
-    private void returnChannel(Channel channel) {
-        try {
-            pool.returnObject(channel);
-        } catch (Exception ignore) {
-            //ignore when return is failing
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitChannelPoolImpl.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitChannelPoolImpl.java
b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitChannelPoolImpl.java
new file mode 100644
index 0000000..fcfdcba
--- /dev/null
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitChannelPoolImpl.java
@@ -0,0 +1,115 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one   *
+ * or more contributor license agreements.  See the NOTICE file *
+ * distributed with this work for additional information        *
+ * regarding copyright ownership.  The ASF licenses this file   *
+ * to you under the Apache License, Version 2.0 (the            *
+ * "License"); you may not use this file except in compliance   *
+ * with the License.  You may obtain a copy of the License at   *
+ *                                                              *
+ *   http://www.apache.org/licenses/LICENSE-2.0                 *
+ *                                                              *
+ * Unless required by applicable law or agreed to in writing,   *
+ * software distributed under the License is distributed on an  *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
+ * KIND, either express or implied.  See the License for the    *
+ * specific language governing permissions and limitations      *
+ * under the License.                                           *
+ ****************************************************************/
+
+package org.apache.james.backend.rabbitmq;
+
+import java.util.function.Supplier;
+
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+
+import org.apache.commons.pool2.BasePooledObjectFactory;
+import org.apache.commons.pool2.ObjectPool;
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.james.util.MemoizedSupplier;
+
+import com.github.fge.lambdas.Throwing;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+
+public class RabbitChannelPoolImpl implements RabbitMQChannelPool {
+
+    private static class ChannelBasePooledObjectFactory extends BasePooledObjectFactory<Channel>
{
+        private final Supplier<Connection> connection;
+
+        public ChannelBasePooledObjectFactory(RabbitMQConnectionFactory factory) {
+            this.connection = MemoizedSupplier.of(
+                    Throwing.supplier(() -> factory.create()).sneakyThrow());
+        }
+
+        @Override
+        public Channel create() throws Exception {
+            return connection.get()
+                    .createChannel();
+        }
+
+        @Override
+        public PooledObject<Channel> wrap(Channel obj) {
+            return new DefaultPooledObject<>(obj);
+        }
+
+        @Override
+        public void destroyObject(PooledObject<Channel> pooledObject) throws Exception
{
+            Channel channel = pooledObject.getObject();
+            channel.close();
+        }
+    }
+
+    private final ObjectPool<Channel> pool;
+
+    @Inject
+    public RabbitChannelPoolImpl(RabbitMQConnectionFactory factory) {
+        pool = new GenericObjectPool<>(
+            new ChannelBasePooledObjectFactory(factory));
+    }
+
+    @Override
+    public <T, E extends Throwable> T execute(RabbitFunction<T, E> f) throws
E, ConnectionFailedException {
+        Channel channel = borrowChannel();
+        try {
+            return f.execute(channel);
+        } finally {
+            returnChannel(channel);
+        }
+    }
+
+    @Override
+    public <E extends Throwable> void execute(RabbitConsumer<E> f) throws E,
ConnectionFailedException {
+        Channel channel = borrowChannel();
+        try {
+            f.execute(channel);
+        } finally {
+            returnChannel(channel);
+        }
+    }
+
+    @PreDestroy
+    public void close() {
+        pool.close();
+    }
+
+    private Channel borrowChannel() {
+        try {
+            return pool.borrowObject();
+        } catch (Exception e) {
+            throw new ConnectionFailedException(e);
+        }
+    }
+
+    private void returnChannel(Channel channel) {
+        try {
+            pool.returnObject(channel);
+        } catch (Exception ignore) {
+            //ignore when return is failing
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQChannelPool.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQChannelPool.java
b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQChannelPool.java
new file mode 100644
index 0000000..44b666d
--- /dev/null
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQChannelPool.java
@@ -0,0 +1,47 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one   *
+ * or more contributor license agreements.  See the NOTICE file *
+ * distributed with this work for additional information        *
+ * regarding copyright ownership.  The ASF licenses this file   *
+ * to you under the Apache License, Version 2.0 (the            *
+ * "License"); you may not use this file except in compliance   *
+ * with the License.  You may obtain a copy of the License at   *
+ *                                                              *
+ *   http://www.apache.org/licenses/LICENSE-2.0                 *
+ *                                                              *
+ * Unless required by applicable law or agreed to in writing,   *
+ * software distributed under the License is distributed on an  *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
+ * KIND, either express or implied.  See the License for the    *
+ * specific language governing permissions and limitations      *
+ * under the License.                                           *
+ ****************************************************************/
+
+package org.apache.james.backend.rabbitmq;
+
+import com.rabbitmq.client.Channel;
+
+public interface RabbitMQChannelPool {
+    class ConnectionFailedException extends RuntimeException {
+        public ConnectionFailedException(Throwable cause) {
+            super(cause);
+        }
+    }
+
+    @FunctionalInterface
+    interface RabbitFunction<T, E extends Throwable> {
+        T execute(Channel channel) throws E;
+    }
+
+    @FunctionalInterface
+    interface RabbitConsumer<E extends Throwable> {
+        void execute(Channel channel) throws E;
+    }
+
+    <T, E extends Throwable> T execute(RabbitFunction<T, E> f)
+        throws E, ConnectionFailedException;
+
+
+    <E extends Throwable> void execute(RabbitConsumer<E> f)
+        throws E, ConnectionFailedException;
+}

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheck.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheck.java
b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheck.java
index fd9f757..17a7ead 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheck.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheck.java
@@ -35,11 +35,11 @@ public class RabbitMQHealthCheck implements HealthCheck {
     private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQHealthCheck.class);
     private static final ComponentName COMPONENT_NAME = new ComponentName("RabbitMQ backend");
 
-    private final RabbitChannelPool rabbitChannelPool;
+    private final RabbitMQChannelPool rabbitChannelPoolImpl;
 
     @Inject
-    public RabbitMQHealthCheck(RabbitChannelPool rabbitChannelPool) throws NoSuchAlgorithmException,
KeyManagementException, URISyntaxException {
-        this.rabbitChannelPool = rabbitChannelPool;
+    public RabbitMQHealthCheck(RabbitMQChannelPool rabbitChannelPoolImpl) throws NoSuchAlgorithmException,
KeyManagementException, URISyntaxException {
+        this.rabbitChannelPoolImpl = rabbitChannelPoolImpl;
     }
 
     @Override
@@ -50,7 +50,7 @@ public class RabbitMQHealthCheck implements HealthCheck {
     @Override
     public Result check() {
         try {
-            return rabbitChannelPool.execute(channel -> {
+            return rabbitChannelPoolImpl.execute(channel -> {
                     if (channel.isOpen()) {
                         return Result.healthy(COMPONENT_NAME);
                     }

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/SimpleChannelPool.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/SimpleChannelPool.java
b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/SimpleChannelPool.java
new file mode 100644
index 0000000..8381aac
--- /dev/null
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/SimpleChannelPool.java
@@ -0,0 +1,42 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one   *
+ * or more contributor license agreements.  See the NOTICE file *
+ * distributed with this work for additional information        *
+ * regarding copyright ownership.  The ASF licenses this file   *
+ * to you under the Apache License, Version 2.0 (the            *
+ * "License"); you may not use this file except in compliance   *
+ * with the License.  You may obtain a copy of the License at   *
+ *                                                              *
+ *   http://www.apache.org/licenses/LICENSE-2.0                 *
+ *                                                              *
+ * Unless required by applicable law or agreed to in writing,   *
+ * software distributed under the License is distributed on an  *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
+ * KIND, either express or implied.  See the License for the    *
+ * specific language governing permissions and limitations      *
+ * under the License.                                           *
+ ****************************************************************/
+
+package org.apache.james.backend.rabbitmq;
+
+import java.io.IOException;
+
+import com.rabbitmq.client.Channel;
+
+public class SimpleChannelPool implements RabbitMQChannelPool {
+    private final Channel channel;
+
+    public SimpleChannelPool(RabbitMQConnectionFactory factory) throws IOException {
+        this.channel = factory.create().createChannel();
+    }
+
+    @Override
+    public synchronized  <T, E extends Throwable> T execute(RabbitFunction<T, E>
f) throws E, ConnectionFailedException {
+        return f.execute(channel);
+    }
+
+    @Override
+    public synchronized  <E extends Throwable> void execute(RabbitConsumer<E>
f) throws E, ConnectionFailedException {
+        f.execute(channel);
+    }
+}

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQExtension.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQExtension.java
b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQExtension.java
index 5326ffb..c8762e7 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQExtension.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQExtension.java
@@ -37,7 +37,7 @@ import com.nurkiewicz.asyncretry.AsyncRetryExecutor;
 public class RabbitMQExtension implements BeforeAllCallback, BeforeEachCallback, AfterAllCallback,
AfterEachCallback, ParameterResolver {
 
     private DockerRabbitMQ rabbitMQ;
-    private RabbitChannelPool rabbitChannelPool;
+    private RabbitChannelPoolImpl rabbitChannelPoolImpl;
 
     @Override
     public void beforeAll(ExtensionContext context) {
@@ -47,12 +47,12 @@ public class RabbitMQExtension implements BeforeAllCallback, BeforeEachCallback,
 
     @Override
     public void beforeEach(ExtensionContext extensionContext) throws Exception {
-        rabbitChannelPool = createRabbitChannelPool();
+        rabbitChannelPoolImpl = createRabbitChannelPool();
     }
 
     @Override
     public void afterEach(ExtensionContext context) throws Exception {
-        rabbitChannelPool.close();
+        rabbitChannelPoolImpl.close();
         rabbitMQ.reset();
     }
 
@@ -71,15 +71,15 @@ public class RabbitMQExtension implements BeforeAllCallback, BeforeEachCallback,
         return rabbitMQ;
     }
 
-    public RabbitChannelPool getRabbitChannelPool() {
-        return rabbitChannelPool;
+    public RabbitChannelPoolImpl getRabbitChannelPool() {
+        return rabbitChannelPoolImpl;
     }
 
     public DockerRabbitMQ getRabbitMQ() {
         return rabbitMQ;
     }
 
-    private RabbitChannelPool createRabbitChannelPool() throws URISyntaxException {
+    private RabbitChannelPoolImpl createRabbitChannelPool() throws URISyntaxException {
         RabbitMQConfiguration rabbitMQConfiguration = RabbitMQConfiguration.builder()
             .amqpUri(rabbitMQ.amqpUri())
             .managementUri(rabbitMQ.managementUri())
@@ -89,6 +89,6 @@ public class RabbitMQExtension implements BeforeAllCallback, BeforeEachCallback,
         RabbitMQConnectionFactory rabbitMQConnectionFactory = new RabbitMQConnectionFactory(
             rabbitMQConfiguration,
             new AsyncRetryExecutor(Executors.newSingleThreadScheduledExecutor()));
-        return new RabbitChannelPool(rabbitMQConnectionFactory);
+        return new RabbitChannelPoolImpl(rabbitMQConnectionFactory);
     }
 }

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheckTest.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheckTest.java
b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheckTest.java
index bc25ebc..b88ef8f 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheckTest.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheckTest.java
@@ -48,7 +48,7 @@ class RabbitMQHealthCheckTest {
                 new AsyncRetryExecutor(Executors.newSingleThreadScheduledExecutor()));
 
         healthCheck = new RabbitMQHealthCheck(
-            new RabbitChannelPool(rabbitMQConnectionFactory));
+            new RabbitChannelPoolImpl(rabbitMQConnectionFactory));
     }
 
     @Test

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java
b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java
index 8c6418b..b54983a 100644
--- a/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java
+++ b/server/queue/queue-api/src/test/java/org/apache/james/queue/api/MailQueueContract.java
@@ -353,8 +353,8 @@ public interface MailQueueContract {
         ConcurrentLinkedDeque<Mail> dequeuedMails = new ConcurrentLinkedDeque<>();
 
         int threadCount = 10;
-        int operationCount = 100;
-        int totalDequeuedMessages = 500;
+        int operationCount = 10;
+        int totalDequeuedMessages = 50;
         ConcurrentTestRunner.builder()
             .operation((threadNumber, step) -> {
                 if (step % 2 == 0) {
@@ -385,8 +385,8 @@ public interface MailQueueContract {
         ConcurrentLinkedDeque<Mail> dequeuedMails = new ConcurrentLinkedDeque<>();
 
         int threadCount = 10;
-        int operationCount = 150;
-        int totalDequeuedMessages = 500;
+        int operationCount = 15;
+        int totalDequeuedMessages = 50;
         ConcurrentTestRunner.builder()
             .operation((threadNumber, step) -> {
                 if (step % 3 == 0) {

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitClient.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitClient.java
b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitClient.java
index 8757e33..9263b1f 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitClient.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitClient.java
@@ -22,7 +22,7 @@ package org.apache.james.queue.rabbitmq;
 import java.io.IOException;
 import java.util.Optional;
 
-import org.apache.james.backend.rabbitmq.RabbitChannelPool;
+import org.apache.james.backend.rabbitmq.RabbitMQChannelPool;
 import org.apache.james.queue.api.MailQueue;
 
 import com.google.common.collect.ImmutableMap;
@@ -40,9 +40,9 @@ class RabbitClient {
     private static final String ROUTING_KEY = "";
     public static final boolean REQUEUE = true;
 
-    private final RabbitChannelPool channelPool;
+    private final RabbitMQChannelPool channelPool;
 
-    RabbitClient(RabbitChannelPool channelPool) {
+    RabbitClient(RabbitMQChannelPool channelPool) {
         this.channelPool = channelPool;
     }
 
@@ -69,17 +69,17 @@ class RabbitClient {
     }
 
     void ack(long deliveryTag) throws IOException {
-        RabbitChannelPool.RabbitConsumer<IOException> consumer = channel -> channel.basicAck(deliveryTag,
!MULTIPLE);
+        RabbitMQChannelPool.RabbitConsumer<IOException> consumer = channel -> channel.basicAck(deliveryTag,
!MULTIPLE);
         channelPool.execute(consumer);
     }
 
     void nack(long deliveryTag) throws IOException {
-        RabbitChannelPool.RabbitConsumer<IOException> consumer = channel -> channel.basicNack(deliveryTag,
!MULTIPLE, REQUEUE);
+        RabbitMQChannelPool.RabbitConsumer<IOException> consumer = channel -> channel.basicNack(deliveryTag,
!MULTIPLE, REQUEUE);
         channelPool.execute(consumer);
     }
 
     Optional<GetResponse> poll(MailQueueName name) throws IOException {
-        RabbitChannelPool.RabbitFunction<Optional<GetResponse>, IOException>
f = channel ->
+        RabbitMQChannelPool.RabbitFunction<Optional<GetResponse>, IOException>
f = channel ->
             Optional.ofNullable(channel.basicGet(name.toWorkQueueName().asString(), !AUTO_ACK));
         return channelPool.execute(f);
     }

http://git-wip-us.apache.org/repos/asf/james-project/blob/7ed58e9e/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
----------------------------------------------------------------------
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
index ee008a9..37507f5 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
@@ -35,10 +35,10 @@ import java.util.stream.Stream;
 import javax.mail.internet.MimeMessage;
 
 import org.apache.james.backend.rabbitmq.DockerRabbitMQ;
-import org.apache.james.backend.rabbitmq.RabbitChannelPool;
 import org.apache.james.backend.rabbitmq.RabbitMQConfiguration;
 import org.apache.james.backend.rabbitmq.RabbitMQConnectionFactory;
 import org.apache.james.backend.rabbitmq.RabbitMQExtension;
+import org.apache.james.backend.rabbitmq.SimpleChannelPool;
 import org.apache.james.backends.cassandra.CassandraCluster;
 import org.apache.james.backends.cassandra.CassandraClusterExtension;
 import org.apache.james.backends.cassandra.components.CassandraModule;
@@ -128,7 +128,7 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract,
MailQ
         RabbitMQConnectionFactory rabbitMQConnectionFactory = new RabbitMQConnectionFactory(rabbitMQConfiguration,
                 new AsyncRetryExecutor(Executors.newSingleThreadScheduledExecutor()));
 
-        RabbitClient rabbitClient = new RabbitClient(new RabbitChannelPool(rabbitMQConnectionFactory));
+        RabbitClient rabbitClient = new RabbitClient(new SimpleChannelPool(rabbitMQConnectionFactory));
         RabbitMQMailQueueFactory.PrivateFactory factory = new RabbitMQMailQueueFactory.PrivateFactory(
             metricTestSystem.getSpyMetricFactory(),
             metricTestSystem.getSpyGaugeRegistry(),
@@ -229,20 +229,6 @@ public class RabbitMQMailQueueTest implements ManageableMailQueueContract,
MailQ
     public void constructorShouldRegisterGetQueueSizeGauge(MailQueueMetricExtension.MailQueueMetricTestSystem
testSystem) {
     }
 
-    @Test
-    @Override
-    @Disabled("JAMES-2544 acknowledgement need to be done on the original channel, which
is not permitted by the channelPool")
-    public void concurrentEnqueueDequeueShouldNotFail() {
-
-    }
-
-    @Test
-    @Override
-    @Disabled("JAMES-2544 acknowledgement need to be done on the original channel, which
is not permitted by the channelPool")
-    public void concurrentEnqueueDequeueWithAckNackShouldNotFail() {
-    }
-
-
     private void enqueueSomeMails(Function<Integer, String> namePattern, int emailCount)
{
         IntStream.rangeClosed(1, emailCount)
             .forEach(Throwing.intConsumer(i -> enQueue(defaultMail()


---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org


Mime
View raw message