james-server-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From adup...@apache.org
Subject [7/7] james-project git commit: JAMES-2545 Rename RabbitMQ backend package
Date Wed, 12 Sep 2018 08:03:29 GMT
JAMES-2545 Rename RabbitMQ backend package


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

Branch: refs/heads/master
Commit: 74114e9365a6b6593a714150f710a086b6647b46
Parents: 0b7748a
Author: Benoit Tellier <btellier@linagora.com>
Authored: Tue Sep 11 15:54:09 2018 +0700
Committer: Antoine Duprat <aduprat@linagora.com>
Committed: Wed Sep 12 10:01:13 2018 +0200

----------------------------------------------------------------------
 .../mailqueue/RabbitMQConfiguration.java        | 117 -------
 .../backend/mailqueue/RabbitMQHealthCheck.java  |  67 ----
 .../backend/rabbitmq/RabbitMQConfiguration.java | 117 +++++++
 .../backend/rabbitmq/RabbitMQHealthCheck.java   |  67 ++++
 .../DockerClusterRabbitMQExtension.java         | 130 --------
 .../james/backend/mailqueue/DockerRabbitMQ.java | 167 ----------
 .../mailqueue/DockerRabbitMQExtension.java      |  53 ---
 .../mailqueue/DockerRabbitMQExtensionTest.java  |  50 ---
 .../backend/mailqueue/InMemoryConsumer.java     |  61 ----
 .../backend/mailqueue/RabbitMQClusterTest.java  | 294 -----------------
 .../mailqueue/RabbitMQConfigurationTest.java    | 134 --------
 .../backend/mailqueue/RabbitMQFixture.java      |  50 ---
 .../mailqueue/RabbitMQHealthCheckTest.java      |  83 -----
 .../james/backend/mailqueue/RabbitMQTest.java   | 330 -------------------
 .../backend/mailqueue/RabbitMQWaitStrategy.java |  67 ----
 .../ReusableDockerRabbitMQExtension.java        |  59 ----
 .../DockerClusterRabbitMQExtension.java         | 130 ++++++++
 .../james/backend/rabbitmq/DockerRabbitMQ.java  | 167 ++++++++++
 .../rabbitmq/DockerRabbitMQExtension.java       |  53 +++
 .../rabbitmq/DockerRabbitMQExtensionTest.java   |  50 +++
 .../backend/rabbitmq/InMemoryConsumer.java      |  61 ++++
 .../backend/rabbitmq/RabbitMQClusterTest.java   | 294 +++++++++++++++++
 .../rabbitmq/RabbitMQConfigurationTest.java     | 134 ++++++++
 .../james/backend/rabbitmq/RabbitMQFixture.java |  50 +++
 .../rabbitmq/RabbitMQHealthCheckTest.java       |  83 +++++
 .../james/backend/rabbitmq/RabbitMQTest.java    | 330 +++++++++++++++++++
 .../backend/rabbitmq/RabbitMQWaitStrategy.java  |  67 ++++
 .../ReusableDockerRabbitMQExtension.java        |  59 ++++
 .../james/modules/rabbitmq/RabbitMQModule.java  |   2 +-
 .../queue/rabbitmq/RabbitMQMailQueueTest.java   |   4 +-
 .../rabbitmq/RabbitMqMailQueueFactoryTest.java  |   4 +-
 .../src/test/resources/logback-test.xml         |   2 +-
 32 files changed, 1668 insertions(+), 1668 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/main/java/org/apache/james/backend/mailqueue/RabbitMQConfiguration.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/mailqueue/RabbitMQConfiguration.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/mailqueue/RabbitMQConfiguration.java
deleted file mode 100644
index 362db74..0000000
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/mailqueue/RabbitMQConfiguration.java
+++ /dev/null
@@ -1,117 +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.mailqueue;
-
-import java.net.URI;
-import java.util.Objects;
-
-import org.apache.commons.configuration.PropertiesConfiguration;
-
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
-
-public class RabbitMQConfiguration {
-    @FunctionalInterface
-    public interface RequireAmqpUri {
-        RequireManagementUri amqpUri(URI amqpUri);
-    }
-
-    @FunctionalInterface
-    public interface RequireManagementUri {
-        Builder managementUri(URI managementUri);
-    }
-
-    public static class Builder {
-        private final URI amqpUri;
-        private final URI managementUri;
-
-        private Builder(URI amqpUri, URI managementUri) {
-            this.amqpUri = amqpUri;
-            this.managementUri = managementUri;
-        }
-
-        public RabbitMQConfiguration build() {
-            Preconditions.checkNotNull(amqpUri, "'amqpUri' should not be null");
-            Preconditions.checkNotNull(managementUri, "'managementUri' should not be null");
-            return new RabbitMQConfiguration(amqpUri, managementUri);
-        }
-    }
-
-    private static final String URI_PROPERTY_NAME = "uri";
-    private static final String MANAGEMENT_URI_PROPERTY_NAME = "management.uri";
-
-    public static RequireAmqpUri builder() {
-        return amqpUri -> managementUri -> new Builder(amqpUri, managementUri);
-    }
-
-    public static RabbitMQConfiguration from(PropertiesConfiguration configuration) {
-        String uriAsString = configuration.getString(URI_PROPERTY_NAME);
-        Preconditions.checkState(!Strings.isNullOrEmpty(uriAsString), "You need to specify the URI of RabbitMQ");
-        URI amqpUri = checkURI(uriAsString);
-
-        String managementUriAsString = configuration.getString(MANAGEMENT_URI_PROPERTY_NAME);
-        Preconditions.checkState(!Strings.isNullOrEmpty(managementUriAsString), "You need to specify the management URI of RabbitMQ");
-        URI managementUri = checkURI(managementUriAsString);
-
-        return builder()
-            .amqpUri(amqpUri)
-            .managementUri(managementUri)
-            .build();
-    }
-
-    private static URI checkURI(String uri) {
-        try {
-            return URI.create(uri);
-        } catch (Exception e) {
-            throw new IllegalStateException("You need to specify a valid URI", e);
-        }
-    }
-
-    private final URI uri;
-    private final URI managementUri;
-
-    private RabbitMQConfiguration(URI uri, URI managementUri) {
-        this.uri = uri;
-        this.managementUri = managementUri;
-    }
-
-    public URI getUri() {
-        return uri;
-    }
-
-    public URI getManagementUri() {
-        return managementUri;
-    }
-
-    @Override
-    public final boolean equals(Object o) {
-        if (o instanceof RabbitMQConfiguration) {
-            RabbitMQConfiguration that = (RabbitMQConfiguration) o;
-
-            return Objects.equals(this.uri, that.uri)
-                && Objects.equals(this.managementUri, that.managementUri);
-        }
-        return false;
-    }
-
-    @Override
-    public final int hashCode() {
-        return Objects.hash(uri, managementUri);
-    }
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/main/java/org/apache/james/backend/mailqueue/RabbitMQHealthCheck.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/mailqueue/RabbitMQHealthCheck.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/mailqueue/RabbitMQHealthCheck.java
deleted file mode 100644
index 2287cb6..0000000
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/mailqueue/RabbitMQHealthCheck.java
+++ /dev/null
@@ -1,67 +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.mailqueue;
-
-import java.net.URISyntaxException;
-import java.security.KeyManagementException;
-import java.security.NoSuchAlgorithmException;
-
-import javax.inject.Inject;
-
-import org.apache.james.core.healthcheck.ComponentName;
-import org.apache.james.core.healthcheck.HealthCheck;
-import org.apache.james.core.healthcheck.Result;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-
-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 ConnectionFactory connectionFactory;
-
-    @Inject
-    public RabbitMQHealthCheck(RabbitMQConfiguration configuration) throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException {
-        this.connectionFactory = new ConnectionFactory();
-        this.connectionFactory.setUri(configuration.getUri());
-    }
-
-    @Override
-    public ComponentName componentName() {
-        return COMPONENT_NAME;
-    }
-
-    @Override
-    public Result check() {
-        try (Connection connection = connectionFactory.newConnection()) {
-            if (connection.isOpen()) {
-                return Result.healthy(COMPONENT_NAME);
-            }
-            LOGGER.error("The created connection was not opened");
-            return Result.unhealthy(COMPONENT_NAME);
-        } catch (Exception e) {
-            LOGGER.error("Unhealthy RabbitMQ instances: could not establish a connection", e);
-            return Result.unhealthy(COMPONENT_NAME);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQConfiguration.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQConfiguration.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQConfiguration.java
new file mode 100644
index 0000000..4093431
--- /dev/null
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQConfiguration.java
@@ -0,0 +1,117 @@
+/****************************************************************
+ * 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.net.URI;
+import java.util.Objects;
+
+import org.apache.commons.configuration.PropertiesConfiguration;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+public class RabbitMQConfiguration {
+    @FunctionalInterface
+    public interface RequireAmqpUri {
+        RequireManagementUri amqpUri(URI amqpUri);
+    }
+
+    @FunctionalInterface
+    public interface RequireManagementUri {
+        Builder managementUri(URI managementUri);
+    }
+
+    public static class Builder {
+        private final URI amqpUri;
+        private final URI managementUri;
+
+        private Builder(URI amqpUri, URI managementUri) {
+            this.amqpUri = amqpUri;
+            this.managementUri = managementUri;
+        }
+
+        public RabbitMQConfiguration build() {
+            Preconditions.checkNotNull(amqpUri, "'amqpUri' should not be null");
+            Preconditions.checkNotNull(managementUri, "'managementUri' should not be null");
+            return new RabbitMQConfiguration(amqpUri, managementUri);
+        }
+    }
+
+    private static final String URI_PROPERTY_NAME = "uri";
+    private static final String MANAGEMENT_URI_PROPERTY_NAME = "management.uri";
+
+    public static RequireAmqpUri builder() {
+        return amqpUri -> managementUri -> new Builder(amqpUri, managementUri);
+    }
+
+    public static RabbitMQConfiguration from(PropertiesConfiguration configuration) {
+        String uriAsString = configuration.getString(URI_PROPERTY_NAME);
+        Preconditions.checkState(!Strings.isNullOrEmpty(uriAsString), "You need to specify the URI of RabbitMQ");
+        URI amqpUri = checkURI(uriAsString);
+
+        String managementUriAsString = configuration.getString(MANAGEMENT_URI_PROPERTY_NAME);
+        Preconditions.checkState(!Strings.isNullOrEmpty(managementUriAsString), "You need to specify the management URI of RabbitMQ");
+        URI managementUri = checkURI(managementUriAsString);
+
+        return builder()
+            .amqpUri(amqpUri)
+            .managementUri(managementUri)
+            .build();
+    }
+
+    private static URI checkURI(String uri) {
+        try {
+            return URI.create(uri);
+        } catch (Exception e) {
+            throw new IllegalStateException("You need to specify a valid URI", e);
+        }
+    }
+
+    private final URI uri;
+    private final URI managementUri;
+
+    private RabbitMQConfiguration(URI uri, URI managementUri) {
+        this.uri = uri;
+        this.managementUri = managementUri;
+    }
+
+    public URI getUri() {
+        return uri;
+    }
+
+    public URI getManagementUri() {
+        return managementUri;
+    }
+
+    @Override
+    public final boolean equals(Object o) {
+        if (o instanceof RabbitMQConfiguration) {
+            RabbitMQConfiguration that = (RabbitMQConfiguration) o;
+
+            return Objects.equals(this.uri, that.uri)
+                && Objects.equals(this.managementUri, that.managementUri);
+        }
+        return false;
+    }
+
+    @Override
+    public final int hashCode() {
+        return Objects.hash(uri, managementUri);
+    }
+}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/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
new file mode 100644
index 0000000..b850e26
--- /dev/null
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/RabbitMQHealthCheck.java
@@ -0,0 +1,67 @@
+/****************************************************************
+ * 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.net.URISyntaxException;
+import java.security.KeyManagementException;
+import java.security.NoSuchAlgorithmException;
+
+import javax.inject.Inject;
+
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+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 ConnectionFactory connectionFactory;
+
+    @Inject
+    public RabbitMQHealthCheck(RabbitMQConfiguration configuration) throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException {
+        this.connectionFactory = new ConnectionFactory();
+        this.connectionFactory.setUri(configuration.getUri());
+    }
+
+    @Override
+    public ComponentName componentName() {
+        return COMPONENT_NAME;
+    }
+
+    @Override
+    public Result check() {
+        try (Connection connection = connectionFactory.newConnection()) {
+            if (connection.isOpen()) {
+                return Result.healthy(COMPONENT_NAME);
+            }
+            LOGGER.error("The created connection was not opened");
+            return Result.unhealthy(COMPONENT_NAME);
+        } catch (Exception e) {
+            LOGGER.error("Unhealthy RabbitMQ instances: could not establish a connection", e);
+            return Result.unhealthy(COMPONENT_NAME);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerClusterRabbitMQExtension.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerClusterRabbitMQExtension.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerClusterRabbitMQExtension.java
deleted file mode 100644
index dd382d0..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerClusterRabbitMQExtension.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.mailqueue;
-
-import java.nio.charset.StandardCharsets;
-
-import org.apache.james.util.Runnables;
-import org.junit.jupiter.api.extension.AfterEachCallback;
-import org.junit.jupiter.api.extension.BeforeEachCallback;
-import org.junit.jupiter.api.extension.ExtensionContext;
-import org.junit.jupiter.api.extension.ParameterContext;
-import org.junit.jupiter.api.extension.ParameterResolutionException;
-import org.junit.jupiter.api.extension.ParameterResolver;
-import org.testcontainers.containers.Network;
-
-import com.github.fge.lambdas.Throwing;
-import com.google.common.collect.ImmutableList;
-import com.google.common.hash.Hashing;
-import com.rabbitmq.client.Address;
-
-public class DockerClusterRabbitMQExtension implements BeforeEachCallback, AfterEachCallback, ParameterResolver {
-
-    public static final String RABBIT_1 = "rabbit1";
-    public static final String RABBIT_2 = "rabbit2";
-    public static final String RABBIT_3 = "rabbit3";
-    private DockerRabbitMQCluster cluster;
-    private Network network;
-
-    @Override
-    public void beforeEach(ExtensionContext context) {
-        String cookie = Hashing.sha256().hashString("secret cookie here", StandardCharsets.UTF_8).toString();
-
-        network = Network.NetworkImpl.builder()
-            .enableIpv6(false)
-            .createNetworkCmdModifiers(ImmutableList.of())
-            .build();
-
-        DockerRabbitMQ rabbitMQ1 = DockerRabbitMQ.withCookieAndNodeName(RABBIT_1, cookie, "rabbit@rabbit1", network);
-        DockerRabbitMQ rabbitMQ2 = DockerRabbitMQ.withCookieAndNodeName(RABBIT_2, cookie, "rabbit@rabbit2", network);
-        DockerRabbitMQ rabbitMQ3 = DockerRabbitMQ.withCookieAndNodeName(RABBIT_3, cookie, "rabbit@rabbit3", network);
-
-        Runnables.runParallel(
-            rabbitMQ1::start,
-            rabbitMQ2::start,
-            rabbitMQ3::start);
-
-        Runnables.runParallel(
-            Throwing.runnable(() -> rabbitMQ2.join(rabbitMQ1)),
-            Throwing.runnable(() -> rabbitMQ3.join(rabbitMQ1)));
-
-
-
-        Runnables.runParallel(
-            Throwing.runnable(rabbitMQ2::startApp),
-            Throwing.runnable(rabbitMQ3::startApp));
-
-        cluster = new DockerRabbitMQCluster(rabbitMQ1, rabbitMQ2, rabbitMQ3);
-    }
-
-    @Override
-    public void afterEach(ExtensionContext context) throws Exception {
-        cluster.stop();
-        network.close();
-    }
-
-    @Override
-    public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
-        return (parameterContext.getParameter().getType() == DockerRabbitMQCluster.class);
-    }
-
-    @Override
-    public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
-        return cluster;
-    }
-
-    public static class DockerRabbitMQCluster {
-
-        private final DockerRabbitMQ rabbitMQ1;
-        private final DockerRabbitMQ rabbitMQ2;
-        private final DockerRabbitMQ rabbitMQ3;
-
-        public DockerRabbitMQCluster(DockerRabbitMQ rabbitMQ1, DockerRabbitMQ rabbitMQ2, DockerRabbitMQ rabbitMQ3) {
-            this.rabbitMQ1 = rabbitMQ1;
-            this.rabbitMQ2 = rabbitMQ2;
-            this.rabbitMQ3 = rabbitMQ3;
-        }
-
-        public void stop() {
-            Runnables.runParallel(
-                Throwing.runnable(rabbitMQ1::stop).orDoNothing(),
-                Throwing.runnable(rabbitMQ2::stop).orDoNothing(),
-                Throwing.runnable(rabbitMQ3::stop).orDoNothing());
-        }
-
-        public DockerRabbitMQ getRabbitMQ1() {
-            return rabbitMQ1;
-        }
-
-        public DockerRabbitMQ getRabbitMQ2() {
-            return rabbitMQ2;
-        }
-
-        public DockerRabbitMQ getRabbitMQ3() {
-            return rabbitMQ3;
-        }
-
-        public ImmutableList<Address> getAddresses() {
-            return ImmutableList.of(
-                new Address(rabbitMQ1.getHostIp(), rabbitMQ1.getPort()),
-                new Address(rabbitMQ2.getHostIp(), rabbitMQ2.getPort()),
-                new Address(rabbitMQ3.getHostIp(), rabbitMQ3.getPort()));
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQ.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQ.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQ.java
deleted file mode 100644
index 29f6757..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQ.java
+++ /dev/null
@@ -1,167 +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.mailqueue;
-
-import java.util.Optional;
-import java.util.UUID;
-
-import org.apache.james.util.docker.Images;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.DockerClientFactory;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.wait.strategy.Wait;
-import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
-
-import com.google.common.collect.ImmutableMap;
-import com.rabbitmq.client.ConnectionFactory;
-
-public class DockerRabbitMQ {
-    private static final Logger LOGGER = LoggerFactory.getLogger(DockerRabbitMQ.class);
-
-    private static final String DEFAULT_RABBIT_NODE = "my-rabbit";
-    private static final int DEFAULT_RABBITMQ_PORT = 5672;
-    private static final int DEFAULT_RABBITMQ_ADMIN_PORT = 15672;
-    private static final String DEFAULT_RABBITMQ_USERNAME = "guest";
-    private static final String DEFAULT_RABBITMQ_PASSWORD = "guest";
-    private static final String RABBITMQ_ERLANG_COOKIE = "RABBITMQ_ERLANG_COOKIE";
-    private static final String RABBITMQ_NODENAME = "RABBITMQ_NODENAME";
-
-    private final GenericContainer<?> container;
-    private final Optional<String> nodeName;
-
-    public static DockerRabbitMQ withCookieAndNodeName(String hostName, String erlangCookie, String nodeName, Network network) {
-        return new DockerRabbitMQ(Optional.ofNullable(hostName), Optional.ofNullable(erlangCookie), Optional.ofNullable(nodeName),
-            Optional.of(network));
-    }
-
-    public static DockerRabbitMQ withoutCookie() {
-        return new DockerRabbitMQ(Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty());
-    }
-
-    @SuppressWarnings("resource")
-    private DockerRabbitMQ(Optional<String> hostName, Optional<String> erlangCookie, Optional<String> nodeName, Optional<Network> net) {
-        container = new GenericContainer<>(Images.RABBITMQ)
-                .withCreateContainerCmdModifier(cmd -> cmd.withName(hostName.orElse(randomName())))
-                .withCreateContainerCmdModifier(cmd -> cmd.withHostName(hostName.orElse(DEFAULT_RABBIT_NODE)))
-                .withExposedPorts(DEFAULT_RABBITMQ_PORT, DEFAULT_RABBITMQ_ADMIN_PORT)
-                .waitingFor(new WaitAllStrategy()
-                    .withStrategy(Wait.forHttp("").forPort(DEFAULT_RABBITMQ_ADMIN_PORT))
-                    .withStrategy(RabbitMQWaitStrategy.withDefaultTimeout(this)))
-                .withLogConsumer(frame -> LOGGER.debug(frame.getUtf8String()))
-                .withCreateContainerCmdModifier(cmd -> cmd.getHostConfig()
-                    .withTmpFs(ImmutableMap.of("/var/lib/rabbitmq/mnesia", "rw,noexec,nosuid,size=100m")));
-        net.ifPresent(container::withNetwork);
-        erlangCookie.ifPresent(cookie -> container.withEnv(RABBITMQ_ERLANG_COOKIE, cookie));
-        nodeName.ifPresent(name -> container.withEnv(RABBITMQ_NODENAME, name));
-        this.nodeName = nodeName;
-    }
-
-    private String randomName() {
-        return UUID.randomUUID().toString();
-    }
-
-    public String getHostIp() {
-        return container.getContainerIpAddress();
-    }
-
-    public Integer getPort() {
-        return container.getMappedPort(DEFAULT_RABBITMQ_PORT);
-    }
-
-    public Integer getAdminPort() {
-        return container.getMappedPort(DEFAULT_RABBITMQ_ADMIN_PORT);
-    }
-
-    public String getUsername() {
-        return DEFAULT_RABBITMQ_USERNAME;
-    }
-
-    public String getPassword() {
-        return DEFAULT_RABBITMQ_PASSWORD;
-    }
-
-    public ConnectionFactory connectionFactory() {
-        ConnectionFactory connectionFactory = new ConnectionFactory();
-        connectionFactory.setHost(getHostIp());
-        connectionFactory.setPort(getPort());
-        connectionFactory.setUsername(getUsername());
-        connectionFactory.setPassword(getPassword());
-        return connectionFactory;
-    }
-
-    public void start() {
-        container.start();
-    }
-
-    public void stop() {
-        container.stop();
-    }
-
-    public void restart() {
-        DockerClientFactory.instance().client()
-            .restartContainerCmd(container.getContainerId());
-    }
-
-    public GenericContainer<?> container() {
-        return container;
-    }
-
-    public String node() {
-        return nodeName.get();
-    }
-
-    public void join(DockerRabbitMQ rabbitMQ) throws Exception {
-        stopApp();
-        joinCluster(rabbitMQ);
-    }
-
-    public void stopApp() throws java.io.IOException, InterruptedException {
-        String stdout = container()
-            .execInContainer("rabbitmqctl", "stop_app")
-            .getStdout();
-        LOGGER.debug("stop_app: {}", stdout);
-    }
-
-    private void joinCluster(DockerRabbitMQ rabbitMQ) throws java.io.IOException, InterruptedException {
-        String stdout = container()
-            .execInContainer("rabbitmqctl", "join_cluster", rabbitMQ.node())
-            .getStdout();
-        LOGGER.debug("join_cluster: {}", stdout);
-    }
-
-    public void startApp() throws Exception {
-        String stdout = container()
-                .execInContainer("rabbitmqctl", "start_app")
-                .getStdout();
-        LOGGER.debug("start_app: {}", stdout);
-    }
-
-    public void reset() throws Exception {
-        stopApp();
-
-        String stdout = container()
-            .execInContainer("rabbitmqctl", "reset")
-            .getStdout();
-        LOGGER.debug("reset: {}", stdout);
-
-        startApp();
-    }
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQExtension.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQExtension.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQExtension.java
deleted file mode 100644
index ea11f81..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQExtension.java
+++ /dev/null
@@ -1,53 +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.mailqueue;
-
-import org.junit.jupiter.api.extension.AfterEachCallback;
-import org.junit.jupiter.api.extension.BeforeEachCallback;
-import org.junit.jupiter.api.extension.ExtensionContext;
-import org.junit.jupiter.api.extension.ParameterContext;
-import org.junit.jupiter.api.extension.ParameterResolutionException;
-import org.junit.jupiter.api.extension.ParameterResolver;
-
-public class DockerRabbitMQExtension implements BeforeEachCallback, AfterEachCallback, ParameterResolver {
-
-    private DockerRabbitMQ rabbitMQ;
-
-    @Override
-    public void beforeEach(ExtensionContext context) {
-        rabbitMQ = DockerRabbitMQ.withoutCookie();
-        rabbitMQ.start();
-    }
-
-    @Override
-    public void afterEach(ExtensionContext context) {
-        rabbitMQ.stop();
-    }
-
-    @Override
-    public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
-        return (parameterContext.getParameter().getType() == DockerRabbitMQ.class);
-    }
-
-    @Override
-    public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
-        return rabbitMQ;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQExtensionTest.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQExtensionTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQExtensionTest.java
deleted file mode 100644
index aaabc5b..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/DockerRabbitMQExtensionTest.java
+++ /dev/null
@@ -1,50 +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.mailqueue;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-
-@ExtendWith(DockerRabbitMQExtension.class)
-public class DockerRabbitMQExtensionTest {
-
-    private ConnectionFactory connectionFactory;
-
-    @BeforeEach
-    public void setup(DockerRabbitMQ rabbitMQ) {
-        connectionFactory = new ConnectionFactory();
-        connectionFactory.setHost(rabbitMQ.getHostIp());
-        connectionFactory.setPort(rabbitMQ.getPort());
-        connectionFactory.setUsername(rabbitMQ.getUsername());
-        connectionFactory.setPassword(rabbitMQ.getPassword());
-    }
-
-    @Test
-    public void containerShouldBeUp() throws Exception {
-        try (Connection connection = connectionFactory.newConnection()) {
-            assertThat(connection.isOpen()).isTrue();
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/InMemoryConsumer.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/InMemoryConsumer.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/InMemoryConsumer.java
deleted file mode 100644
index e6fe021..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/InMemoryConsumer.java
+++ /dev/null
@@ -1,61 +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.mailqueue;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.DefaultConsumer;
-import com.rabbitmq.client.Envelope;
-
-public class InMemoryConsumer extends DefaultConsumer {
-
-    @FunctionalInterface
-    interface Operation {
-        void perform();
-    }
-
-    private final ConcurrentLinkedQueue<Integer> messages;
-    private final Operation operation;
-
-    public InMemoryConsumer(Channel channel) {
-        this(channel, () -> { });
-    }
-
-    public InMemoryConsumer(Channel channel, Operation operation) {
-        super(channel);
-        this.operation = operation;
-        this.messages = new ConcurrentLinkedQueue<>();
-    }
-
-    @Override
-    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
-        operation.perform();
-        Integer payload = Integer.valueOf(new String(body, StandardCharsets.UTF_8));
-        messages.add(payload);
-    }
-
-    public Queue<Integer> getConsumedMessages() {
-        return messages;
-    }
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQClusterTest.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQClusterTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQClusterTest.java
deleted file mode 100644
index 5467840..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQClusterTest.java
+++ /dev/null
@@ -1,294 +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.mailqueue;
-
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.AUTO_DELETE;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.DIRECT;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.DURABLE;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.EXCHANGE_NAME;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.EXCLUSIVE;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.NO_PROPERTIES;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.ROUTING_KEY;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.awaitAtMostOneMinute;
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.IntStream;
-
-import org.apache.james.backend.mailqueue.DockerClusterRabbitMQExtension.DockerRabbitMQCluster;
-import org.awaitility.Awaitility;
-import org.awaitility.Duration;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.api.Nested;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.github.fge.lambdas.Throwing;
-import com.github.steveash.guavate.Guavate;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.rabbitmq.client.Address;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-
-@ExtendWith(DockerClusterRabbitMQExtension.class)
-class RabbitMQClusterTest {
-
-    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQClusterTest.class);
-
-    private static final String QUEUE = "queue";
-
-    @Nested
-    class ClusterSharing {
-
-        private ConnectionFactory node1ConnectionFactory;
-        private ConnectionFactory node2ConnectionFactory;
-        private Connection node1Connection;
-        private Connection node2Connection;
-        private Channel node1Channel;
-        private Channel node2Channel;
-
-        @BeforeEach
-        void setup(DockerRabbitMQCluster cluster) throws IOException, TimeoutException {
-            node1ConnectionFactory = cluster.getRabbitMQ1().connectionFactory();
-            node2ConnectionFactory = cluster.getRabbitMQ2().connectionFactory();
-            node1Connection = node1ConnectionFactory.newConnection();
-            node2Connection = node2ConnectionFactory.newConnection();
-            node1Channel = node1Connection.createChannel();
-            node2Channel = node2Connection.createChannel();
-        }
-
-        @AfterEach
-        void tearDown() {
-            closeQuietly(node1Channel, node2Channel, node1Connection, node2Connection);
-        }
-
-        @Test
-        void rabbitMQManagerShouldReturnThreeNodesWhenAskingForStatus(DockerRabbitMQCluster cluster) throws Exception {
-            String stdout = cluster.getRabbitMQ1().container()
-                .execInContainer("rabbitmqctl", "cluster_status")
-                .getStdout();
-
-            assertThat(stdout)
-                .contains(
-                    DockerClusterRabbitMQExtension.RABBIT_1,
-                    DockerClusterRabbitMQExtension.RABBIT_2,
-                    DockerClusterRabbitMQExtension.RABBIT_3);
-        }
-
-        @Test
-        void queuesShouldBeShared() throws Exception {
-            node1Channel.exchangeDeclare(EXCHANGE_NAME, DIRECT, DURABLE);
-            node1Channel.queueDeclare(QUEUE, DURABLE, !EXCLUSIVE, !AUTO_DELETE, ImmutableMap.of()).getQueue();
-            node1Channel.queueBind(QUEUE, EXCHANGE_NAME, ROUTING_KEY);
-
-            int nbMessages = 10;
-            IntStream.range(0, nbMessages)
-                .mapToObj(i -> asBytes(String.valueOf(i)))
-                .forEach(Throwing.consumer(
-                    bytes -> node1Channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
-
-
-            InMemoryConsumer consumer2 = new InMemoryConsumer(node2Channel);
-            node2Channel.basicConsume(QUEUE, consumer2);
-
-            awaitAtMostOneMinute.until(() -> consumer2.getConsumedMessages().size() == nbMessages);
-
-            List<Integer> expectedResult = IntStream.range(0, nbMessages).boxed().collect(Guavate.toImmutableList());
-            assertThat(consumer2.getConsumedMessages()).containsOnlyElementsOf(expectedResult);
-        }
-
-        @Test
-        void queuesShouldBeDeclarableOnAnotherNode() throws Exception {
-            node1Channel.exchangeDeclare(EXCHANGE_NAME, DIRECT, DURABLE);
-            node2Channel.queueDeclare(QUEUE, DURABLE, !EXCLUSIVE, !AUTO_DELETE, ImmutableMap.of()).getQueue();
-            node2Channel.queueBind(QUEUE, EXCHANGE_NAME, ROUTING_KEY);
-
-            int nbMessages = 10;
-            IntStream.range(0, nbMessages)
-                .mapToObj(i -> asBytes(String.valueOf(i)))
-                .forEach(Throwing.consumer(
-                    bytes -> node1Channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
-
-            InMemoryConsumer consumer2 = new InMemoryConsumer(node2Channel);
-            node2Channel.basicConsume(QUEUE, consumer2);
-
-            awaitAtMostOneMinute.until(() -> consumer2.getConsumedMessages().size() == nbMessages);
-
-            List<Integer> expectedResult = IntStream.range(0, nbMessages).boxed().collect(Guavate.toImmutableList());
-            assertThat(consumer2.getConsumedMessages()).containsOnlyElementsOf(expectedResult);
-        }
-
-    }
-
-    @Nested
-    class ClusterNodesFailure {
-
-        private ConnectionFactory node1ConnectionFactory;
-        private Connection resilientConnection;
-        private Channel resilientChannel;
-        private Connection node2Connection;
-        private Channel node2Channel;
-
-        @BeforeEach
-        void setup(DockerRabbitMQCluster cluster) throws IOException, TimeoutException {
-            node1ConnectionFactory = cluster.getRabbitMQ1().connectionFactory();
-            resilientConnection = node1ConnectionFactory.newConnection(cluster.getAddresses());
-            resilientChannel = resilientConnection.createChannel();
-            ConnectionFactory node2ConnectionFactory = cluster.getRabbitMQ2().connectionFactory();
-            node2Connection = node2ConnectionFactory.newConnection();
-            node2Channel = node2Connection.createChannel();
-        }
-
-        @AfterEach
-        void tearDown() {
-            closeQuietly(resilientConnection, resilientChannel);
-        }
-
-        @Disabled("JAMES-2334 For some reason, we are unable to recover topology when reconnecting" +
-            "See https://github.com/rabbitmq/rabbitmq-server/issues/959")
-        @Test
-        void nodeKillingWhenProducing(DockerRabbitMQCluster cluster) throws Exception {
-            resilientChannel.exchangeDeclare(EXCHANGE_NAME, DIRECT, DURABLE);
-            resilientChannel.queueDeclare(QUEUE, DURABLE, !EXCLUSIVE, !AUTO_DELETE, ImmutableMap.of()).getQueue();
-            resilientChannel.queueBind(QUEUE, EXCHANGE_NAME, ROUTING_KEY);
-
-            int nbMessages = 20;
-            int firstBatchSize = nbMessages / 2;
-            IntStream.range(0, firstBatchSize)
-                .mapToObj(i -> asBytes(String.valueOf(i)))
-                .forEach(Throwing.consumer(
-                    bytes -> resilientChannel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
-
-            InMemoryConsumer consumer = new InMemoryConsumer(node2Channel);
-            node2Channel.basicConsume(QUEUE, consumer);
-            awaitAtMostOneMinute.until(() -> consumer.getConsumedMessages().size() == firstBatchSize);
-
-            cluster.getRabbitMQ1().stop();
-
-            IntStream.range(firstBatchSize, nbMessages)
-                .mapToObj(i -> asBytes(String.valueOf(i)))
-                .forEach(this::tryPublishWithRetry);
-
-            awaitAtMostOneMinute.until(() -> consumer.getConsumedMessages().size() == nbMessages);
-
-            List<Integer> expectedResult = IntStream.range(0, nbMessages).boxed().collect(Guavate.toImmutableList());
-            assertThat(consumer.getConsumedMessages()).containsOnlyElementsOf(expectedResult);
-        }
-
-        private void tryPublishWithRetry(byte[] bytes) {
-            Awaitility.waitAtMost(Duration.ONE_MINUTE).pollInterval(Duration.ONE_SECOND).until(() -> tryPublish(bytes));
-        }
-
-        private boolean tryPublish(byte[] bytes) {
-            try {
-                resilientChannel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes);
-                return true;
-            } catch (Exception e) {
-                LOGGER.error("failed publish", e);
-                return false;
-            }
-        }
-
-        @Test
-        void connectingToAClusterWithAFailedRabbit(DockerRabbitMQCluster cluster) throws Exception {
-            ConnectionFactory node3ConnectionFactory = cluster.getRabbitMQ3().connectionFactory();
-            ImmutableList<Address> clusterAddresses = cluster.getAddresses();
-            cluster.getRabbitMQ3().stop();
-
-            try (Connection connection = node3ConnectionFactory.newConnection(clusterAddresses);
-                 Channel channel = connection.createChannel()) {
-
-                channel.exchangeDeclare(EXCHANGE_NAME, DIRECT, DURABLE);
-                channel.queueDeclare(QUEUE, DURABLE, !EXCLUSIVE, !AUTO_DELETE, ImmutableMap.of()).getQueue();
-                channel.queueBind(QUEUE, EXCHANGE_NAME, ROUTING_KEY);
-
-                int nbMessages = 10;
-                IntStream.range(0, nbMessages)
-                    .mapToObj(i -> asBytes(String.valueOf(i)))
-                    .forEach(Throwing.consumer(
-                        bytes -> channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
-
-                InMemoryConsumer consumer = new InMemoryConsumer(channel);
-                channel.basicConsume(QUEUE, consumer);
-
-                awaitAtMostOneMinute.until(() -> consumer.getConsumedMessages().size() == nbMessages);
-
-                List<Integer> expectedResult = IntStream.range(0, nbMessages).boxed().collect(Guavate.toImmutableList());
-                assertThat(consumer.getConsumedMessages()).containsOnlyElementsOf(expectedResult);
-            }
-        }
-
-        @Test
-        void nodeKillingWhenConsuming(DockerRabbitMQCluster cluster) throws Exception {
-            node2Channel.exchangeDeclare(EXCHANGE_NAME, DIRECT, DURABLE);
-            node2Channel.queueDeclare(QUEUE, DURABLE, !EXCLUSIVE, !AUTO_DELETE, ImmutableMap.of()).getQueue();
-            node2Channel.queueBind(QUEUE, EXCHANGE_NAME, ROUTING_KEY);
-
-            int nbMessages = 10;
-            IntStream.range(0, nbMessages)
-                .mapToObj(i -> asBytes(String.valueOf(i)))
-                .forEach(Throwing.consumer(
-                    bytes -> resilientChannel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
-
-            AtomicInteger counter = new AtomicInteger(0);
-            InMemoryConsumer consumer = new InMemoryConsumer(resilientChannel,
-                () -> stopWhenHalfProcessed(cluster, nbMessages, counter));
-            resilientChannel.basicConsume(QUEUE, consumer);
-
-            awaitAtMostOneMinute.until(() -> consumer.getConsumedMessages().size() == nbMessages);
-
-            List<Integer> expectedResult = IntStream.range(0, nbMessages).boxed().collect(Guavate.toImmutableList());
-            assertThat(consumer.getConsumedMessages()).containsOnlyElementsOf(expectedResult);
-        }
-
-        private void stopWhenHalfProcessed(DockerRabbitMQCluster cluster, int nbMessages, AtomicInteger counter) {
-            if (counter.incrementAndGet() == nbMessages / 2) {
-                cluster.getRabbitMQ1().stop();
-            }
-        }
-
-    }
-
-    private void closeQuietly(AutoCloseable... closeables) {
-        Arrays.stream(closeables).forEach(this::closeQuietly);
-    }
-
-    private void closeQuietly(AutoCloseable closeable) {
-        try {
-            closeable.close();
-        } catch (Exception e) {
-            //ignore error
-        }
-    }
-
-    private byte[] asBytes(String message) {
-        return message.getBytes(StandardCharsets.UTF_8);
-    }
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQConfigurationTest.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQConfigurationTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQConfigurationTest.java
deleted file mode 100644
index 4da2a91..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQConfigurationTest.java
+++ /dev/null
@@ -1,134 +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.mailqueue;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-import java.net.URI;
-
-import org.apache.commons.configuration.PropertiesConfiguration;
-import org.junit.jupiter.api.Test;
-
-import nl.jqno.equalsverifier.EqualsVerifier;
-
-class RabbitMQConfigurationTest {
-
-    @Test
-    void shouldRespectBeanContract() {
-        EqualsVerifier.forClass(RabbitMQConfiguration.class).verify();
-    }
-
-    @Test
-    void fromShouldThrowWhenURIIsNotInTheConfiguration() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-
-        assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessage("You need to specify the URI of RabbitMQ");
-    }
-
-    @Test
-    void fromShouldThrowWhenURIIsNull() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-        configuration.addProperty("uri", null);
-
-        assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessage("You need to specify the URI of RabbitMQ");
-    }
-
-    @Test
-    void fromShouldThrowWhenURIIsEmpty() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-        configuration.addProperty("uri", "");
-
-        assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessage("You need to specify the URI of RabbitMQ");
-    }
-
-    @Test
-    void fromShouldThrowWhenURIIsInvalid() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-        configuration.addProperty("uri", ":invalid");
-
-        assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessage("You need to specify a valid URI");
-    }
-
-    @Test
-    void fromShouldThrowWhenManagementURIIsNotInTheConfiguration() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-        configuration.addProperty("uri", "amqp://james:james@rabbitmq_host:5672");
-
-        assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessage("You need to specify the management URI of RabbitMQ");
-    }
-
-    @Test
-    void fromShouldThrowWhenManagementURIIsNull() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-        configuration.addProperty("uri", "amqp://james:james@rabbitmq_host:5672");
-        configuration.addProperty("management.uri", null);
-
-        assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessage("You need to specify the management URI of RabbitMQ");
-    }
-
-    @Test
-    void fromShouldThrowWhenManagementURIIsEmpty() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-        configuration.addProperty("uri", "amqp://james:james@rabbitmq_host:5672");
-        configuration.addProperty("management.uri", "");
-
-        assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessage("You need to specify the management URI of RabbitMQ");
-    }
-
-    @Test
-    void fromShouldThrowWhenManagementURIIsInvalid() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-        configuration.addProperty("uri", "amqp://james:james@rabbitmq_host:5672");
-        configuration.addProperty("management.uri", ":invalid");
-
-        assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
-            .isInstanceOf(IllegalStateException.class)
-            .hasMessage("You need to specify a valid URI");
-    }
-
-    @Test
-    void fromShouldReturnTheConfigurationWhenRequiredParametersAreGiven() {
-        PropertiesConfiguration configuration = new PropertiesConfiguration();
-        String amqpUri = "amqp://james:james@rabbitmq_host:5672";
-        configuration.addProperty("uri", amqpUri);
-        String managementUri = "http://james:james@rabbitmq_host:15672/api/";
-        configuration.addProperty("management.uri", managementUri);
-
-        assertThat(RabbitMQConfiguration.from(configuration))
-            .isEqualTo(RabbitMQConfiguration.builder()
-                .amqpUri(URI.create(amqpUri))
-                .managementUri(URI.create(managementUri))
-                .build());
-    }
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQFixture.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQFixture.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQFixture.java
deleted file mode 100644
index 8b83a96..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQFixture.java
+++ /dev/null
@@ -1,50 +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.mailqueue;
-
-import static org.awaitility.Duration.ONE_HUNDRED_MILLISECONDS;
-import static org.awaitility.Duration.ONE_MINUTE;
-
-import org.awaitility.Awaitility;
-import org.awaitility.Duration;
-import org.awaitility.core.ConditionFactory;
-
-import com.rabbitmq.client.AMQP;
-
-public class RabbitMQFixture {
-    public static final boolean DURABLE = true;
-    public static final boolean AUTO_ACK = true;
-    public static final AMQP.BasicProperties NO_PROPERTIES = null;
-    public static final String EXCHANGE_NAME = "exchangeName";
-    public static final String ROUTING_KEY = "routingKey";
-    public static final String DIRECT = "direct";
-    public static final boolean EXCLUSIVE = true;
-    public static final boolean AUTO_DELETE = true;
-    public static final String WORK_QUEUE = "workQueue";
-
-    public static Duration slowPacedPollInterval = ONE_HUNDRED_MILLISECONDS;
-    public static ConditionFactory calmlyAwait = Awaitility.with()
-        .pollInterval(slowPacedPollInterval)
-        .and()
-        .with()
-        .pollDelay(slowPacedPollInterval)
-        .await();
-    public static ConditionFactory awaitAtMostOneMinute = calmlyAwait.atMost(ONE_MINUTE);
-}

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQHealthCheckTest.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQHealthCheckTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQHealthCheckTest.java
deleted file mode 100644
index 1352590..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQHealthCheckTest.java
+++ /dev/null
@@ -1,83 +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.mailqueue;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.net.URI;
-
-import org.apache.james.core.healthcheck.Result;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-@ExtendWith(DockerRabbitMQExtension.class)
-class RabbitMQHealthCheckTest {
-    private RabbitMQHealthCheck healthCheck;
-
-    @BeforeEach
-    void setUp(DockerRabbitMQ rabbitMQ) throws Exception {
-        URI amqpUri = URI.create("amqp://" + rabbitMQ.getHostIp() + ":" + rabbitMQ.getPort());
-        URI managementUri = URI.create("http://" + rabbitMQ.getHostIp() + ":" + rabbitMQ.getAdminPort());
-
-        healthCheck = new RabbitMQHealthCheck(
-            RabbitMQConfiguration.builder()
-                .amqpUri(amqpUri)
-                .managementUri(managementUri)
-                .build());
-    }
-
-    @Test
-    void checkShouldReturnHealthyWhenRabbitMQIsRunning() {
-        Result check = healthCheck.check();
-
-        assertThat(check.isHealthy()).isTrue();
-    }
-
-    @Test
-    void checkShouldReturnUnhealthyWhenRabbitMQIsNotRunning(DockerRabbitMQ rabbitMQ) throws Exception {
-        rabbitMQ.stopApp();
-
-        Result check = healthCheck.check();
-
-        assertThat(check.isHealthy()).isFalse();
-    }
-
-    @Test
-    void checkShouldDetectWhenRabbitMQRecovered(DockerRabbitMQ rabbitMQ) throws Exception {
-        rabbitMQ.stopApp();
-        healthCheck.check();
-
-        rabbitMQ.startApp();
-
-        Result check = healthCheck.check();
-        assertThat(check.isHealthy()).isTrue();
-    }
-
-    @Test
-    void checkShouldDetectWhenRabbitMQFail(DockerRabbitMQ rabbitMQ) throws Exception {
-        healthCheck.check();
-
-        rabbitMQ.stopApp();
-
-        Result check = healthCheck.check();
-        assertThat(check.isHealthy()).isFalse();
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/james-project/blob/74114e93/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQTest.java
----------------------------------------------------------------------
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQTest.java
deleted file mode 100644
index f54ba25..0000000
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/mailqueue/RabbitMQTest.java
+++ /dev/null
@@ -1,330 +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.mailqueue;
-
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.AUTO_ACK;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.AUTO_DELETE;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.DIRECT;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.DURABLE;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.EXCHANGE_NAME;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.EXCLUSIVE;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.NO_PROPERTIES;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.ROUTING_KEY;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.WORK_QUEUE;
-import static org.apache.james.backend.mailqueue.RabbitMQFixture.awaitAtMostOneMinute;
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.Queue;
-import java.util.concurrent.TimeoutException;
-import java.util.stream.IntStream;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Nested;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-import com.github.fge.lambdas.Throwing;
-import com.github.steveash.guavate.Guavate;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Iterables;
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
-
-@ExtendWith(DockerRabbitMQExtension.class)
-class RabbitMQTest {
-
-    @Nested
-    class SingleConsumerTest {
-
-        private ConnectionFactory connectionFactory;
-        private Connection connection;
-        private Channel channel;
-
-        @BeforeEach
-        void setup(DockerRabbitMQ rabbitMQ) throws IOException, TimeoutException {
-            connectionFactory = rabbitMQ.connectionFactory();
-            connection = connectionFactory.newConnection();
-            channel = connection.createChannel();
-        }
-
-        @AfterEach
-        void tearDown() {
-            closeQuietly(connection, channel);
-        }
-
-        @Test
-        void publishedEventWithoutSubscriberShouldNotBeLost() throws Exception {
-            String queueName = createQueue(channel);
-            publishAMessage(channel);
-            awaitAtMostOneMinute.until(() -> messageReceived(channel, queueName));
-        }
-
-        @Test
-        void demonstrateDurability(DockerRabbitMQ rabbitMQ) throws Exception {
-            String queueName = createQueue(channel);
-            publishAMessage(channel);
-
-            rabbitMQ.restart();
-
-            awaitAtMostOneMinute.until(() -> containerIsRestarted(rabbitMQ));
-            assertThat(channel.basicGet(queueName, !AUTO_ACK)).isNotNull();
-        }
-
-        private Boolean containerIsRestarted(DockerRabbitMQ rabbitMQ) {
-            try {
-                rabbitMQ.connectionFactory().newConnection();
-                return true;
-            } catch (Exception e) {
-                return false;
-            }
-        }
-
-        private String createQueue(Channel channel) throws IOException {
-            channel.exchangeDeclare(EXCHANGE_NAME, DIRECT, DURABLE);
-            String queueName = channel.queueDeclare().getQueue();
-            channel.queueBind(queueName, EXCHANGE_NAME, ROUTING_KEY);
-            return queueName;
-        }
-
-        private void publishAMessage(Channel channel) throws IOException {
-            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, asBytes("Hello, world!"));
-        }
-
-        private Boolean messageReceived(Channel channel, String queueName) {
-            try {
-                return channel.basicGet(queueName, !AUTO_ACK) != null;
-            } catch (Exception e) {
-                return false;
-            }
-        }
-    }
-
-    @Nested
-    class FourConnections {
-
-        private ConnectionFactory connectionFactory1;
-        private ConnectionFactory connectionFactory2;
-        private ConnectionFactory connectionFactory3;
-        private ConnectionFactory connectionFactory4;
-        private Connection connection1;
-        private Connection connection2;
-        private Connection connection3;
-        private Connection connection4;
-        private Channel channel1;
-        private Channel channel2;
-        private Channel channel3;
-        private Channel channel4;
-
-        @BeforeEach
-        void setup(DockerRabbitMQ rabbitMQ) throws IOException, TimeoutException {
-            connectionFactory1 = rabbitMQ.connectionFactory();
-            connectionFactory2 = rabbitMQ.connectionFactory();
-            connectionFactory3 = rabbitMQ.connectionFactory();
-            connectionFactory4 = rabbitMQ.connectionFactory();
-            connection1 = connectionFactory1.newConnection();
-            connection2 = connectionFactory2.newConnection();
-            connection3 = connectionFactory3.newConnection();
-            connection4 = connectionFactory4.newConnection();
-            channel1 = connection1.createChannel();
-            channel2 = connection2.createChannel();
-            channel3 = connection3.createChannel();
-            channel4 = connection4.createChannel();
-        }
-
-        @AfterEach
-        void tearDown() {
-            closeQuietly(
-                channel1, channel2, channel3, channel4,
-                connection1, connection2, connection3, connection4);
-        }
-
-        @Nested
-        class BroadCast {
-
-            // In the following case, each consumer will receive the messages produced by the
-            // producer
-            // To do so, each consumer will bind it's queue to the producer exchange.
-            @Test
-            void rabbitMQShouldSupportTheBroadcastCase() throws Exception {
-                // Declare a single exchange and three queues attached to it.
-                channel1.exchangeDeclare(EXCHANGE_NAME, DIRECT, DURABLE);
-
-                String queue2 = channel2.queueDeclare().getQueue();
-                channel2.queueBind(queue2, EXCHANGE_NAME, ROUTING_KEY);
-                String queue3 = channel3.queueDeclare().getQueue();
-                channel3.queueBind(queue3, EXCHANGE_NAME, ROUTING_KEY);
-                String queue4 = channel4.queueDeclare().getQueue();
-                channel4.queueBind(queue4, EXCHANGE_NAME, ROUTING_KEY);
-
-                InMemoryConsumer consumer2 = new InMemoryConsumer(channel2);
-                InMemoryConsumer consumer3 = new InMemoryConsumer(channel3);
-                InMemoryConsumer consumer4 = new InMemoryConsumer(channel4);
-                channel2.basicConsume(queue2, consumer2);
-                channel3.basicConsume(queue3, consumer3);
-                channel4.basicConsume(queue4, consumer4);
-
-                // the publisher will produce 10 messages
-                IntStream.range(0, 10)
-                    .mapToObj(String::valueOf)
-                    .map(RabbitMQTest.this::asBytes)
-                    .forEach(Throwing.consumer(
-                        bytes -> channel1.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
-
-                awaitAtMostOneMinute.until(
-                    () -> countReceivedMessages(consumer2, consumer3, consumer4) == 30);
-
-                ImmutableList<Integer> expectedResult = IntStream.range(0, 10).boxed().collect(Guavate.toImmutableList());
-                // Check every subscriber have receive all the messages.
-                assertThat(consumer2.getConsumedMessages()).containsOnlyElementsOf(expectedResult);
-                assertThat(consumer3.getConsumedMessages()).containsOnlyElementsOf(expectedResult);
-                assertThat(consumer4.getConsumedMessages()).containsOnlyElementsOf(expectedResult);
-            }
-        }
-
-        @Nested
-        class WorkQueue {
-
-            // In the following case, consumers will receive the messages produced by the
-            // producer but will share them.
-            // To do so, we will bind a single queue to the producer exchange.
-            @Test
-            void rabbitMQShouldSupportTheWorkQueueCase() throws Exception {
-                int nbMessages = 100;
-
-                // Declare the exchange and a single queue attached to it.
-                channel1.exchangeDeclare(EXCHANGE_NAME, "direct", DURABLE);
-                channel1.queueDeclare(WORK_QUEUE, DURABLE, !EXCLUSIVE, AUTO_DELETE, ImmutableMap.of());
-                channel1.queueBind(WORK_QUEUE, EXCHANGE_NAME, ROUTING_KEY);
-
-                // Publisher will produce 100 messages
-                IntStream.range(0, nbMessages)
-                    .mapToObj(String::valueOf)
-                    .map(RabbitMQTest.this::asBytes)
-                    .forEach(Throwing.consumer(
-                        bytes -> channel1.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
-
-                InMemoryConsumer consumer2 = new InMemoryConsumer(channel2);
-                InMemoryConsumer consumer3 = new InMemoryConsumer(channel3);
-                InMemoryConsumer consumer4 = new InMemoryConsumer(channel4);
-                channel2.basicConsume(WORK_QUEUE, consumer2);
-                channel3.basicConsume(WORK_QUEUE, consumer3);
-                channel4.basicConsume(WORK_QUEUE, consumer4);
-
-                awaitAtMostOneMinute.until(
-                    () -> countReceivedMessages(consumer2, consumer3, consumer4) == nbMessages);
-
-                ImmutableList<Integer> expectedResult = IntStream.range(0, nbMessages).boxed().collect(Guavate.toImmutableList());
-
-                assertThat(
-                    Iterables.concat(
-                        consumer2.getConsumedMessages(),
-                        consumer3.getConsumedMessages(),
-                        consumer4.getConsumedMessages()))
-                    .containsOnlyElementsOf(expectedResult);
-            }
-
-        }
-
-        @Nested
-        class Routing {
-            @Test
-            void rabbitMQShouldSupportRouting() throws Exception {
-                String conversation1 = "c1";
-                String conversation2 = "c2";
-                String conversation3 = "c3";
-                String conversation4 = "c4";
-
-                // Declare the exchange and a single queue attached to it.
-                channel1.exchangeDeclare(EXCHANGE_NAME, DIRECT, DURABLE);
-
-                String queue1 = channel1.queueDeclare().getQueue();
-                // 1 will follow conversation 1 and 2
-                channel1.queueBind(queue1, EXCHANGE_NAME, conversation1);
-                channel1.queueBind(queue1, EXCHANGE_NAME, conversation2);
-
-                String queue2 = channel2.queueDeclare().getQueue();
-                // 2 will follow conversation 2 and 3
-                channel2.queueBind(queue2, EXCHANGE_NAME, conversation2);
-                channel2.queueBind(queue2, EXCHANGE_NAME, conversation3);
-
-                String queue3 = channel3.queueDeclare().getQueue();
-                // 3 will follow conversation 3 and 4
-                channel3.queueBind(queue3, EXCHANGE_NAME, conversation3);
-                channel3.queueBind(queue3, EXCHANGE_NAME, conversation4);
-
-                String queue4 = channel4.queueDeclare().getQueue();
-                // 4 will follow conversation 1 and 4
-                channel4.queueBind(queue4, EXCHANGE_NAME, conversation1);
-                channel4.queueBind(queue4, EXCHANGE_NAME, conversation4);
-
-                channel1.basicPublish(EXCHANGE_NAME, conversation1, NO_PROPERTIES, asBytes("1"));
-                channel2.basicPublish(EXCHANGE_NAME, conversation2, NO_PROPERTIES, asBytes("2"));
-                channel3.basicPublish(EXCHANGE_NAME, conversation3, NO_PROPERTIES, asBytes("3"));
-                channel4.basicPublish(EXCHANGE_NAME, conversation4, NO_PROPERTIES, asBytes("4"));
-
-                InMemoryConsumer consumer1 = new InMemoryConsumer(channel1);
-                InMemoryConsumer consumer2 = new InMemoryConsumer(channel2);
-                InMemoryConsumer consumer3 = new InMemoryConsumer(channel3);
-                InMemoryConsumer consumer4 = new InMemoryConsumer(channel4);
-                channel1.basicConsume(queue1, consumer1);
-                channel2.basicConsume(queue2, consumer2);
-                channel3.basicConsume(queue3, consumer3);
-                channel4.basicConsume(queue4, consumer4);
-
-                awaitAtMostOneMinute.until(() -> countReceivedMessages(consumer1, consumer2, consumer3, consumer4) == 8);
-
-                assertThat(consumer1.getConsumedMessages()).containsOnly(1, 2);
-                assertThat(consumer2.getConsumedMessages()).containsOnly(2, 3);
-                assertThat(consumer3.getConsumedMessages()).containsOnly(3, 4);
-                assertThat(consumer4.getConsumedMessages()).containsOnly(1, 4);
-            }
-        }
-
-        private long countReceivedMessages(InMemoryConsumer... consumers) {
-            return Arrays.stream(consumers)
-                .map(InMemoryConsumer::getConsumedMessages)
-                .mapToLong(Queue::size)
-                .sum();
-        }
-
-    }
-
-    private void closeQuietly(AutoCloseable... closeables) {
-        Arrays.stream(closeables).forEach(this::closeQuietly);
-    }
-
-    private void closeQuietly(AutoCloseable closeable) {
-        try {
-            closeable.close();
-        } catch (Exception e) {
-            //ignore error
-        }
-    }
-
-    private byte[] asBytes(String message) {
-        return message.getBytes(StandardCharsets.UTF_8);
-    }
-
-}


---------------------------------------------------------------------
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