You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2022/06/30 10:56:58 UTC
[james-project] 01/01: Add a vhost configuration option to RabbitMQ
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch rabbit-vhost-list-queues
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit af63d96cc9547a9b83430f97e77c41a311c737e7
Author: Rene Cordier <rc...@linagora.com>
AuthorDate: Thu Jun 30 17:56:37 2022 +0700
Add a vhost configuration option to RabbitMQ
---
.../backends/rabbitmq/RabbitMQConfiguration.java | 28 +++++++++++++++++---
.../backends/rabbitmq/RabbitMQManagementAPI.java | 6 +++++
.../rabbitmq/RabbitMQConfigurationTest.java | 30 ++++++++++++++++++++++
.../modules/ROOT/pages/configure/rabbitmq.adoc | 5 ++++
.../sample-configuration/rabbitmq.properties | 4 +++
.../sample-configuration/rabbitmq.properties | 4 +++
.../rabbitmq/RabbitMQMailQueueManagement.java | 9 +++++--
.../rabbitmq/RabbitMqMailQueueFactoryTest.java | 13 ++++++++++
src/site/xdoc/server/config-rabbitmq.xml | 5 ++++
upgrade-instructions.md | 20 +++++++++++++++
10 files changed, 118 insertions(+), 6 deletions(-)
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java
index 75f096fdec..f9f8e42bb6 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java
@@ -306,6 +306,7 @@ public class RabbitMQConfiguration {
private static final String QUEUE_TTL = "notification.queue.ttl";
private static final String EVENT_BUS_NOTIFICATION_DURABILITY_ENABLED = "event.bus.notification.durability.enabled";
private static final String EVENT_BUS_PUBLISH_CONFIRM_ENABLED = "event.bus.publish.confirm.enabled";
+ private static final String VHOST = "vhost";
public static class ManagementCredentials {
@@ -402,6 +403,7 @@ public class RabbitMQConfiguration {
private Optional<Long> queueTTL;
private Optional<Boolean> eventBusPublishConfirmEnabled;
private Optional<Boolean> eventBusNotificationDurabilityEnabled;
+ private Optional<String> vhost;
private Builder(URI amqpUri, URI managementUri, ManagementCredentials managementCredentials) {
this.amqpUri = amqpUri;
@@ -423,6 +425,7 @@ public class RabbitMQConfiguration {
this.queueTTL = Optional.empty();
this.eventBusPublishConfirmEnabled = Optional.empty();
this.eventBusNotificationDurabilityEnabled = Optional.empty();
+ this.vhost = Optional.empty();
}
public Builder maxRetries(int maxRetries) {
@@ -513,6 +516,11 @@ public class RabbitMQConfiguration {
return this;
}
+ public Builder vhost(Optional<String> vhost) {
+ this.vhost = vhost;
+ return this;
+ }
+
public RabbitMQConfiguration build() {
Preconditions.checkNotNull(amqpUri, "'amqpUri' should not be null");
Preconditions.checkNotNull(managementUri, "'managementUri' should not be null");
@@ -535,7 +543,8 @@ public class RabbitMQConfiguration {
hostsDefaultingToUri(),
queueTTL,
eventBusPublishConfirmEnabled.orElse(true),
- eventBusNotificationDurabilityEnabled.orElse(true));
+ eventBusNotificationDurabilityEnabled.orElse(true),
+ vhost);
}
private List<Host> hostsDefaultingToUri() {
@@ -584,6 +593,8 @@ public class RabbitMQConfiguration {
Optional<Long> queueTTL = Optional.ofNullable(configuration.getLong(QUEUE_TTL, null));
+ Optional<String> vhost = Optional.ofNullable(configuration.getString(VHOST, null));
+
ManagementCredentials managementCredentials = ManagementCredentials.from(configuration);
return builder()
.amqpUri(amqpUri)
@@ -598,6 +609,7 @@ public class RabbitMQConfiguration {
.queueTTL(queueTTL)
.eventBusNotificationDurabilityEnabled(configuration.getBoolean(EVENT_BUS_NOTIFICATION_DURABILITY_ENABLED, null))
.eventBusPublishConfirmEnabled(configuration.getBoolean(EVENT_BUS_PUBLISH_CONFIRM_ENABLED, null))
+ .vhost(vhost)
.build();
}
@@ -668,12 +680,14 @@ public class RabbitMQConfiguration {
private final Optional<Long> queueTTL;
private final boolean eventBusPublishConfirmEnabled;
private final boolean eventBusNotificationDurabilityEnabled;
+ private final Optional<String> vhost;
private RabbitMQConfiguration(URI uri, URI managementUri, ManagementCredentials managementCredentials, int maxRetries, int minDelayInMs,
int connectionTimeoutInMs, int channelRpcTimeoutInMs, int handshakeTimeoutInMs, int shutdownTimeoutInMs,
int networkRecoveryIntervalInMs, Boolean useSsl, Boolean useSslManagement, SSLConfiguration sslConfiguration,
boolean useQuorumQueues, int quorumQueueReplicationFactor, List<Host> hosts, Optional<Long> queueTTL,
- boolean eventBusPublishConfirmEnabled, boolean eventBusNotificationDurabilityEnabled) {
+ boolean eventBusPublishConfirmEnabled, boolean eventBusNotificationDurabilityEnabled,
+ Optional<String> vhost) {
this.uri = uri;
this.managementUri = managementUri;
this.managementCredentials = managementCredentials;
@@ -693,6 +707,7 @@ public class RabbitMQConfiguration {
this.queueTTL = queueTTL;
this.eventBusPublishConfirmEnabled = eventBusPublishConfirmEnabled;
this.eventBusNotificationDurabilityEnabled = eventBusNotificationDurabilityEnabled;
+ this.vhost = vhost;
}
public URI getUri() {
@@ -772,6 +787,10 @@ public class RabbitMQConfiguration {
return eventBusNotificationDurabilityEnabled;
}
+ public Optional<String> getVhost() {
+ return vhost;
+ }
+
@Override
public final boolean equals(Object o) {
if (o instanceof RabbitMQConfiguration) {
@@ -795,7 +814,8 @@ public class RabbitMQConfiguration {
&& Objects.equals(this.hosts, that.hosts)
&& Objects.equals(this.queueTTL, that.queueTTL)
&& Objects.equals(this.eventBusPublishConfirmEnabled, that.eventBusPublishConfirmEnabled)
- && Objects.equals(this.eventBusNotificationDurabilityEnabled, that.eventBusNotificationDurabilityEnabled);
+ && Objects.equals(this.eventBusNotificationDurabilityEnabled, that.eventBusNotificationDurabilityEnabled)
+ && Objects.equals(this.vhost, that.vhost);
}
return false;
}
@@ -804,6 +824,6 @@ public class RabbitMQConfiguration {
public final int hashCode() {
return Objects.hash(uri, managementUri, maxRetries, minDelayInMs, connectionTimeoutInMs, quorumQueueReplicationFactor, useQuorumQueues, hosts,
channelRpcTimeoutInMs, handshakeTimeoutInMs, shutdownTimeoutInMs, networkRecoveryIntervalInMs, managementCredentials, useSsl, useSslManagement,
- sslConfiguration, queueTTL, eventBusPublishConfirmEnabled, eventBusNotificationDurabilityEnabled);
+ sslConfiguration, queueTTL, eventBusPublishConfirmEnabled, eventBusNotificationDurabilityEnabled, vhost);
}
}
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java
index 5661a4d314..59488fbc18 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java
@@ -435,6 +435,12 @@ public interface RabbitMQManagementAPI {
@RequestLine("GET /api/queues")
List<MessageQueue> listQueues();
+ @RequestLine("GET /api/queues/{vhost}")
+ List<MessageQueue> listVhostQueues(@Param("vhost") String vhost);
+
+ @RequestLine("PUT /api/vhosts/{vhost}")
+ List<MessageQueue> addVhost(@Param("vhost") String vhost);
+
@RequestLine(value = "GET /api/queues/{vhost}/{name}", decodeSlash = false)
MessageQueueDetails queueDetails(@Param("vhost") String vhost, @Param("name") String name);
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java
index 49fd796719..32f9a2c2cd 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java
@@ -455,6 +455,36 @@ class RabbitMQConfigurationTest {
.isInstanceOf(ConversionException.class);
}
+ @Test
+ void fromShouldReturnEmptyVhostValueByDefault() {
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ String amqpUri = "amqp://james:james@rabbitmqhost:5672";
+ configuration.addProperty("uri", amqpUri);
+ String managementUri = "http://james:james@rabbitmqhost:15672/api/";
+ configuration.addProperty("management.uri", managementUri);
+ configuration.addProperty("management.user", DEFAULT_USER);
+ configuration.addProperty("management.password", DEFAULT_PASSWORD_STRING);
+
+ assertThat(RabbitMQConfiguration.from(configuration).getVhost())
+ .isEqualTo(Optional.empty());
+ }
+
+ @Test
+ void fromShouldReturnVhostValueWhenGiven() {
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ String amqpUri = "amqp://james:james@rabbitmqhost:5672";
+ configuration.addProperty("uri", amqpUri);
+ String managementUri = "http://james:james@rabbitmqhost:15672/api/";
+ configuration.addProperty("management.uri", managementUri);
+ configuration.addProperty("management.user", DEFAULT_USER);
+ configuration.addProperty("management.password", DEFAULT_PASSWORD_STRING);
+
+ configuration.addProperty("vhost", "test");
+
+ assertThat(RabbitMQConfiguration.from(configuration).getVhost())
+ .isEqualTo(Optional.of("test"));
+ }
+
@Nested
class ManagementCredentialsTest {
@Test
diff --git a/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc b/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
index d07fb14a52..b3c2665ea7 100644
--- a/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
+++ b/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
@@ -99,6 +99,11 @@ A coma separated list of hosts, example: hosts=ip1:5672,ip2:5672
| event.bus.notification.durability.enabled
| Whether or not the queue backing notifications should be durable. Optional boolean, defaults to true.
+
+| vhost
+| Optional string. The virtual host used by James to create queues and exchanges on Rabbitmq.
+If omitted, it will use the default Rabbitmq one "/".
+
|===
== RabbitMQ MailQueue Configuration
diff --git a/server/apps/distributed-app/sample-configuration/rabbitmq.properties b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
index 3c196413bb..5694d43cbb 100644
--- a/server/apps/distributed-app/sample-configuration/rabbitmq.properties
+++ b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
@@ -5,6 +5,10 @@
# Mandatory
uri=amqp://rabbitmq:5672
+# Vhost to use for creating queues and exchanges
+# Optional, will default to default rabbitmq vhost "/" if not declared
+# vhost=vhost1
+
# Optional, default to the host specified as part of the URI.
# Allow creating cluster aware connections.
# hosts=ip1:5672,ip2:5672
diff --git a/server/apps/distributed-pop3-app/sample-configuration/rabbitmq.properties b/server/apps/distributed-pop3-app/sample-configuration/rabbitmq.properties
index c65652e64b..d891444948 100644
--- a/server/apps/distributed-pop3-app/sample-configuration/rabbitmq.properties
+++ b/server/apps/distributed-pop3-app/sample-configuration/rabbitmq.properties
@@ -5,6 +5,10 @@
# Mandatory
uri=amqp://rabbitmq:5672
+# Vhost to use for creating queues and exchanges
+# Optional, will default to default rabbitmq vhost "/" if not declared
+# vhost=vhost1
+
# Optional, default to the host specified as part of the URI.
# Allow creating cluster aware connections.
# hosts=ip1:5672,ip2:5672
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueManagement.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueManagement.java
index 75ea266c4a..a653d16547 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueManagement.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueManagement.java
@@ -32,19 +32,23 @@ import com.google.common.annotations.VisibleForTesting;
public class RabbitMQMailQueueManagement {
private final RabbitMQManagementAPI api;
+ private final Optional<String> vhost;
@Inject
RabbitMQMailQueueManagement(RabbitMQConfiguration configuration) {
- api = RabbitMQManagementAPI.from(configuration);
+ this.api = RabbitMQManagementAPI.from(configuration);
+ this.vhost = configuration.getVhost();
}
@VisibleForTesting
RabbitMQMailQueueManagement(RabbitMQManagementAPI api) {
this.api = api;
+ this.vhost = Optional.empty();
}
Stream<MailQueueName> listCreatedMailQueueNames() {
- return api.listQueues()
+ return vhost.map(api::listVhostQueues)
+ .orElse(api.listQueues())
.stream()
.map(RabbitMQManagementAPI.MessageQueue::getName)
.map(MailQueueName::fromRabbitWorkQueueName)
@@ -52,6 +56,7 @@ public class RabbitMQMailQueueManagement {
.distinct();
}
+ @VisibleForTesting
public void deleteAllQueues() {
api.listQueues()
.forEach(queue -> api.deleteQueue("/", queue.getName()));
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
index 39983d7d55..cea76fdd8a 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMqMailQueueFactoryTest.java
@@ -22,6 +22,7 @@ package org.apache.james.queue.rabbitmq;
import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.backends.rabbitmq.Constants.EXCLUSIVE;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -30,6 +31,7 @@ import static org.mockito.Mockito.when;
import java.time.Clock;
import org.apache.james.backends.rabbitmq.RabbitMQExtension;
+import org.apache.james.backends.rabbitmq.RabbitMQManagementAPI;
import org.apache.james.blob.api.HashBlobId;
import org.apache.james.blob.mail.MimeMessageStore;
import org.apache.james.metrics.api.NoopGaugeRegistry;
@@ -48,6 +50,7 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import reactor.rabbitmq.QueueSpecification;
class RabbitMqMailQueueFactoryTest implements MailQueueFactoryContract<RabbitMQMailQueue> {
+ private static final String VHOST = "vhost1";
private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
@RegisterExtension
@@ -109,4 +112,14 @@ class RabbitMqMailQueueFactoryTest implements MailQueueFactoryContract<RabbitMQM
.doesNotThrowAnyException();
}
+ @Test
+ void getQueuesShouldBeEmptyWhenListingOtherVhost() throws Exception {
+ RabbitMQManagementAPI api = rabbitMQExtension.managementAPI();
+ api.addVhost(VHOST);
+
+ mailQueueFactory.createQueue(NAME_1);
+ mailQueueFactory.createQueue(NAME_2);
+
+ assertThat(api.listVhostQueues(VHOST)).isEmpty();
+ }
}
diff --git a/src/site/xdoc/server/config-rabbitmq.xml b/src/site/xdoc/server/config-rabbitmq.xml
index 9acbc64e22..69ef2e6ea7 100644
--- a/src/site/xdoc/server/config-rabbitmq.xml
+++ b/src/site/xdoc/server/config-rabbitmq.xml
@@ -138,6 +138,11 @@
<dt><strong>event.bus.notification.durability.enabled</strong></dt>
<dd>Whether or not the queue backing notifications should be durable. Optional boolean, defaults to true.</dd>
+ <dt><strong>vhost</strong></dt>
+ <dd>Optional string.
+ The virtual host used by James to create queues and exchanges on Rabbitmq.
+ If omitted, it will use the default Rabbitmq one "/".
+ </dd>
</dl>
</section>
diff --git a/upgrade-instructions.md b/upgrade-instructions.md
index 1e02f2f8b3..c905f5cb5e 100644
--- a/upgrade-instructions.md
+++ b/upgrade-instructions.md
@@ -18,6 +18,26 @@ Change list:
- [Adding authorized_users column to user table](#adding-authorized_users-column-to-user-table)
- [Migration to Cassandra driver 4](#migration-to-cassandra-driver-4)
+- [Adding vhost option to Rabbitmq configuration](#adding-vhost-option-to-rabbitmq-configuration)
+
+### Adding vhost option to Rabbitmq configuration
+
+Date: 30/06/2022
+
+Concerned product: Distributed James
+
+It is possible now to specify a virtual host to be able to create exchanges and queues on a different one
+than the default "/". You just need to add in `rabbitmq.properties` file:
+
+```
+vhost=[VHOST]
+```
+
+Also don't forget to add it to the uri, like that:
+
+```
+amqp://[USERNAME]:[PASSWORD]@[HOST]:[PORT]/[VHOST]
+```
### Migration to Cassandra driver 4
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org