You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2022/04/20 15:13:41 UTC
[james-project] 04/06: JAMES-3749 Allow disable durability, publish confirms
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit f9069aff92c97e69f2fda48dd74560a72bbeabab
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Apr 15 12:39:21 2022 +0700
JAMES-3749 Allow disable durability, publish confirms
Doing so enable massive gains in throughtput thus some
operators might be interested in it...
---
.../backends/rabbitmq/RabbitMQConfiguration.java | 44 ++++++++++++++++++++--
.../org/apache/james/events/EventDispatcher.java | 19 +++++++---
.../james/events/KeyRegistrationHandler.java | 3 +-
.../org/apache/james/events/RabbitMQEventBus.java | 4 +-
.../modules/ROOT/pages/configure/rabbitmq.adoc | 13 ++++++-
.../org/apache/james/queue/rabbitmq/Enqueuer.java | 38 ++++++++++++-------
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 2 +-
.../view/RabbitMQMailQueueConfiguration.java | 26 ++++++++++---
src/site/xdoc/server/config-rabbitmq.xml | 10 +++++
9 files changed, 123 insertions(+), 36 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 48588d478b..75f096fdec 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
@@ -304,6 +304,8 @@ public class RabbitMQConfiguration {
private static final String SSL_KEY_STORE_PATH = "ssl.keystore";
private static final String SSL_KEY_STORE_PASSWORD = "ssl.keystore.password";
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";
public static class ManagementCredentials {
@@ -398,6 +400,8 @@ public class RabbitMQConfiguration {
private Optional<Integer> quorumQueueReplicationFactor;
private Optional<SSLConfiguration> sslConfiguration;
private Optional<Long> queueTTL;
+ private Optional<Boolean> eventBusPublishConfirmEnabled;
+ private Optional<Boolean> eventBusNotificationDurabilityEnabled;
private Builder(URI amqpUri, URI managementUri, ManagementCredentials managementCredentials) {
this.amqpUri = amqpUri;
@@ -417,6 +421,8 @@ public class RabbitMQConfiguration {
this.quorumQueueReplicationFactor = Optional.empty();
this.hosts = ImmutableList.builder();
this.queueTTL = Optional.empty();
+ this.eventBusPublishConfirmEnabled = Optional.empty();
+ this.eventBusNotificationDurabilityEnabled = Optional.empty();
}
public Builder maxRetries(int maxRetries) {
@@ -471,6 +477,16 @@ public class RabbitMQConfiguration {
return this;
}
+ public Builder eventBusPublishConfirmEnabled(Boolean eventBusPublishConfirmEnabled) {
+ this.eventBusPublishConfirmEnabled = Optional.ofNullable(eventBusPublishConfirmEnabled);
+ return this;
+ }
+
+ public Builder eventBusNotificationDurabilityEnabled(Boolean eventBusNotificationDurabilityEnabled) {
+ this.eventBusNotificationDurabilityEnabled = Optional.ofNullable(eventBusNotificationDurabilityEnabled);
+ return this;
+ }
+
public Builder useQuorumQueues(Boolean useQuorumQueues) {
this.useQuorumQueues = Optional.ofNullable(useQuorumQueues);
return this;
@@ -517,7 +533,9 @@ public class RabbitMQConfiguration {
useQuorumQueues.orElse(false),
quorumQueueReplicationFactor.orElse(DEFAULT_QUORUM_QUEUE_REPLICATION_FACTOR),
hostsDefaultingToUri(),
- queueTTL);
+ queueTTL,
+ eventBusPublishConfirmEnabled.orElse(true),
+ eventBusNotificationDurabilityEnabled.orElse(true));
}
private List<Host> hostsDefaultingToUri() {
@@ -578,6 +596,8 @@ public class RabbitMQConfiguration {
.quorumQueueReplicationFactor(quorumQueueReplicationFactor)
.hosts(hosts)
.queueTTL(queueTTL)
+ .eventBusNotificationDurabilityEnabled(configuration.getBoolean(EVENT_BUS_NOTIFICATION_DURABILITY_ENABLED, null))
+ .eventBusPublishConfirmEnabled(configuration.getBoolean(EVENT_BUS_PUBLISH_CONFIRM_ENABLED, null))
.build();
}
@@ -646,11 +666,14 @@ public class RabbitMQConfiguration {
private final List<Host> hosts;
private final ManagementCredentials managementCredentials;
private final Optional<Long> queueTTL;
+ private final boolean eventBusPublishConfirmEnabled;
+ private final boolean eventBusNotificationDurabilityEnabled;
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 useQuorumQueues, int quorumQueueReplicationFactor, List<Host> hosts, Optional<Long> queueTTL,
+ boolean eventBusPublishConfirmEnabled, boolean eventBusNotificationDurabilityEnabled) {
this.uri = uri;
this.managementUri = managementUri;
this.managementCredentials = managementCredentials;
@@ -668,6 +691,8 @@ public class RabbitMQConfiguration {
this.quorumQueueReplicationFactor = quorumQueueReplicationFactor;
this.hosts = hosts;
this.queueTTL = queueTTL;
+ this.eventBusPublishConfirmEnabled = eventBusPublishConfirmEnabled;
+ this.eventBusNotificationDurabilityEnabled = eventBusNotificationDurabilityEnabled;
}
public URI getUri() {
@@ -739,6 +764,14 @@ public class RabbitMQConfiguration {
return queueTTL;
}
+ public boolean isEventBusPublishConfirmEnabled() {
+ return eventBusPublishConfirmEnabled;
+ }
+
+ public boolean isEventBusNotificationDurabilityEnabled() {
+ return eventBusNotificationDurabilityEnabled;
+ }
+
@Override
public final boolean equals(Object o) {
if (o instanceof RabbitMQConfiguration) {
@@ -760,7 +793,9 @@ public class RabbitMQConfiguration {
&& Objects.equals(this.useSslManagement, that.useSslManagement)
&& Objects.equals(this.sslConfiguration, that.sslConfiguration)
&& Objects.equals(this.hosts, that.hosts)
- && Objects.equals(this.queueTTL, that.queueTTL);
+ && Objects.equals(this.queueTTL, that.queueTTL)
+ && Objects.equals(this.eventBusPublishConfirmEnabled, that.eventBusPublishConfirmEnabled)
+ && Objects.equals(this.eventBusNotificationDurabilityEnabled, that.eventBusNotificationDurabilityEnabled);
}
return false;
}
@@ -768,6 +803,7 @@ public class RabbitMQConfiguration {
@Override
public final int hashCode() {
return Objects.hash(uri, managementUri, maxRetries, minDelayInMs, connectionTimeoutInMs, quorumQueueReplicationFactor, useQuorumQueues, hosts,
- channelRpcTimeoutInMs, handshakeTimeoutInMs, shutdownTimeoutInMs, networkRecoveryIntervalInMs, managementCredentials, useSsl, useSslManagement, sslConfiguration, queueTTL);
+ channelRpcTimeoutInMs, handshakeTimeoutInMs, shutdownTimeoutInMs, networkRecoveryIntervalInMs, managementCredentials, useSsl, useSslManagement,
+ sslConfiguration, queueTTL, eventBusPublishConfirmEnabled, eventBusNotificationDurabilityEnabled);
}
}
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java b/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java
index 6270c645ac..f598bf1a59 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java
@@ -32,6 +32,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Set;
+import org.apache.james.backends.rabbitmq.RabbitMQConfiguration;
import org.apache.james.events.RoutingKeyConverter.RoutingKey;
import org.apache.james.util.MDCBuilder;
import org.apache.james.util.MDCStructuredLogger;
@@ -67,11 +68,12 @@ public class EventDispatcher {
private final AMQP.BasicProperties basicProperties;
private final ListenerExecutor listenerExecutor;
private final EventDeadLetters deadLetters;
+ private final RabbitMQConfiguration configuration;
EventDispatcher(NamingStrategy namingStrategy, EventBusId eventBusId, EventSerializer eventSerializer, Sender sender,
LocalListenerRegistry localListenerRegistry,
ListenerExecutor listenerExecutor,
- EventDeadLetters deadLetters) {
+ EventDeadLetters deadLetters, RabbitMQConfiguration configuration) {
this.namingStrategy = namingStrategy;
this.eventSerializer = eventSerializer;
this.sender = sender;
@@ -84,6 +86,7 @@ public class EventDispatcher {
.build();
this.listenerExecutor = listenerExecutor;
this.deadLetters = deadLetters;
+ this.configuration = configuration;
}
void start() {
@@ -183,11 +186,15 @@ public class EventDispatcher {
if (routingKeys.isEmpty()) {
return Mono.empty();
}
- return sender.sendWithPublishConfirms(toMessages(serializedEvent, routingKeys))
- .subscribeOn(Schedulers.elastic()) // channel.confirmSelect is synchronous
- .filter(outboundMessageResult -> !outboundMessageResult.isAck())
- .next()
- .handle((result, sink) -> sink.error(new Exception("Publish was not acked")));
+ if (configuration.isEventBusPublishConfirmEnabled()) {
+ return sender.sendWithPublishConfirms(toMessages(serializedEvent, routingKeys))
+ .subscribeOn(Schedulers.elastic()) // channel.confirmSelect is synchronous
+ .filter(outboundMessageResult -> !outboundMessageResult.isAck())
+ .next()
+ .handle((result, sink) -> sink.error(new Exception("Publish was not acked")));
+ } else {
+ return sender.send(toMessages(serializedEvent, routingKeys));
+ }
}
private Flux<OutboundMessage> toMessages(byte[] serializedEvent, Collection<RoutingKey> routingKeys) {
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java b/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java
index 593cb9d9c1..0ab240adb8 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java
@@ -20,7 +20,6 @@
package org.apache.james.events;
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.apache.james.events.RabbitMQEventBus.EVENT_BUS_ID;
@@ -121,7 +120,7 @@ class KeyRegistrationHandler {
private void declareQueue(Sender sender) {
sender.declareQueue(
QueueSpecification.queue(registrationQueue.asString())
- .durable(DURABLE)
+ .durable(configuration.isEventBusNotificationDurabilityEnabled())
.exclusive(!EXCLUSIVE)
.autoDelete(AUTO_DELETE)
.arguments(configuration.workQueueArgumentsBuilder().build()))
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java b/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java
index b8c85c788f..8755320d83 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java
@@ -90,7 +90,7 @@ public class RabbitMQEventBus implements EventBus, Startable {
LocalListenerRegistry localListenerRegistry = new LocalListenerRegistry();
keyRegistrationHandler = new KeyRegistrationHandler(namingStrategy, eventBusId, eventSerializer, sender, receiverProvider, routingKeyConverter, localListenerRegistry, listenerExecutor, retryBackoff, configuration, metricFactory);
groupRegistrationHandler = new GroupRegistrationHandler(namingStrategy, eventSerializer, channelPool, sender, receiverProvider, retryBackoff, eventDeadLetters, listenerExecutor, eventBusId, configuration);
- eventDispatcher = new EventDispatcher(namingStrategy, eventBusId, eventSerializer, sender, localListenerRegistry, listenerExecutor, eventDeadLetters);
+ eventDispatcher = new EventDispatcher(namingStrategy, eventBusId, eventSerializer, sender, localListenerRegistry, listenerExecutor, eventDeadLetters, configuration);
eventDispatcher.start();
keyRegistrationHandler.start();
@@ -110,7 +110,7 @@ public class RabbitMQEventBus implements EventBus, Startable {
LocalListenerRegistry localListenerRegistry = new LocalListenerRegistry();
keyRegistrationHandler = new KeyRegistrationHandler(namingStrategy, eventBusId, eventSerializer, sender, receiverProvider, routingKeyConverter, localListenerRegistry, listenerExecutor, retryBackoff, configuration, metricFactory);
groupRegistrationHandler = new GroupRegistrationHandler(namingStrategy, eventSerializer, channelPool, sender, receiverProvider, retryBackoff, eventDeadLetters, listenerExecutor, eventBusId, configuration);
- eventDispatcher = new EventDispatcher(namingStrategy, eventBusId, eventSerializer, sender, localListenerRegistry, listenerExecutor, eventDeadLetters);
+ eventDispatcher = new EventDispatcher(namingStrategy, eventBusId, eventSerializer, sender, localListenerRegistry, listenerExecutor, eventDeadLetters, configuration);
keyRegistrationHandler.declareQueue();
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 540d4b2173..6b97764135 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
@@ -88,8 +88,17 @@ False (default value) results in the usage of classic queues.
| hosts
| Optional, default to the host specified as part of the URI.
-| Allow creating cluster aware connections.
-| A coma separated list of hosts, example: hosts=ip1:5672,ip2:5672
+Allow creating cluster aware connections.
+A coma separated list of hosts, example: hosts=ip1:5672,ip2:5672
+
+| mailqueue.publish.confirm.enabled
+| Whether or not to enable publish confirms for the mail queue. Optional boolean, defaults to true.
+
+| event.bus.publish.confirm.enabled
+| Whether or not to enable publish confirms for the event bus. Optional boolean, defaults to true.
+
+| event.bus.notification.durability.enabled
+| Whether or not the queue backing notifications should be durable. Optional boolean, defaults to true.
|===
== RabbitMQ MailQueue Configuration
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
index b72139c942..e7ad613971 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Enqueuer.java
@@ -34,6 +34,7 @@ import org.apache.james.blob.mail.MimeMessagePartsId;
import org.apache.james.metrics.api.Metric;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.rabbitmq.view.RabbitMQMailQueueConfiguration;
import org.apache.james.queue.rabbitmq.view.api.MailQueueView;
import org.apache.james.queue.rabbitmq.view.cassandra.CassandraMailQueueBrowser;
import org.apache.mailet.Mail;
@@ -56,16 +57,18 @@ class Enqueuer {
private final MailReferenceSerializer mailReferenceSerializer;
private final Metric enqueueMetric;
private final MailQueueView mailQueueView;
+ private final RabbitMQMailQueueConfiguration configuration;
private final Clock clock;
Enqueuer(MailQueueName name, Sender sender, Store<MimeMessage, MimeMessagePartsId> mimeMessageStore,
MailReferenceSerializer serializer, MetricFactory metricFactory,
- MailQueueView mailQueueView, Clock clock) {
+ MailQueueView mailQueueView, RabbitMQMailQueueConfiguration configuration, Clock clock) {
this.name = name;
this.sender = sender;
this.mimeMessageStore = mimeMessageStore;
this.mailReferenceSerializer = serializer;
this.mailQueueView = mailQueueView;
+ this.configuration = configuration;
this.clock = clock;
this.enqueueMetric = metricFactory.generate(ENQUEUED_METRIC_NAME_PREFIX + name.asString());
}
@@ -112,19 +115,26 @@ class Enqueuer {
EMPTY_ROUTING_KEY,
basicProperties,
getMailReferenceBytes(mailReference));
- return sender.sendWithPublishConfirms(Mono.just(data))
- .subscribeOn(Schedulers.elastic()) // channel.confirmSelect is synchronous
- .next()
- .handle((result, sink) -> {
- if (!result.isAck()) {
- sink.error(new MailQueue.MailQueueException("Publish was not acked"));
- } else {
- sink.complete();
- }
- })
- // AutoRecoveringConnection blocks this forever
- .timeout(Duration.ofSeconds(10), Mono.error(() -> new MailQueue.MailQueueException("Timeout enqueueing " + mailReference.getMail().getName())))
- .then();
+
+ if (configuration.isMailQueuePublishConfirmEnabled()) {
+ return sender.sendWithPublishConfirms(Mono.just(data))
+ .subscribeOn(Schedulers.elastic()) // channel.confirmSelect is synchronous
+ .next()
+ .handle((result, sink) -> {
+ if (!result.isAck()) {
+ sink.error(new MailQueue.MailQueueException("Publish was not acked"));
+ } else {
+ sink.complete();
+ }
+ })
+ // AutoRecoveringConnection blocks this forever
+ .timeout(Duration.ofSeconds(10), Mono.error(() -> new MailQueue.MailQueueException("Timeout enqueueing " + mailReference.getMail().getName())))
+ .then();
+ } else {
+ return sender.send(Mono.just(data))
+ // AutoRecoveringConnection blocks this forever
+ .timeout(Duration.ofSeconds(10), Mono.error(() -> new MailQueue.MailQueueException("Timeout enqueueing " + mailReference.getMail().getName())));
+ }
}
private EnqueuedItem toEnqueuedItems(MailReference mailReference) {
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
index ac8ae2490c..14e355c8c4 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
@@ -100,7 +100,7 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
metricFactory,
mailQueueName,
new Enqueuer(mailQueueName, sender, mimeMessageStore, mailReferenceSerializer,
- metricFactory, mailQueueView, clock),
+ metricFactory, mailQueueView, configuration, clock),
new Dequeuer(mailQueueName, receiverProvider, mailLoader, mailReferenceSerializer,
metricFactory, mailQueueView, prefetchCount),
mailQueueView,
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/RabbitMQMailQueueConfiguration.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/RabbitMQMailQueueConfiguration.java
index 01455170b0..d0c1edb649 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/RabbitMQMailQueueConfiguration.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/RabbitMQMailQueueConfiguration.java
@@ -28,7 +28,8 @@ public class RabbitMQMailQueueConfiguration {
private static final boolean DEFAULT_SIZE_METRICS_DISABLED = false;
public static class Builder {
- private Optional<Boolean> sizeMetricsEnabled;
+ private Optional<Boolean> sizeMetricsEnabled = Optional.empty();
+ private Optional<Boolean> mailQueuePublishConfirmEnabled = Optional.empty();
public Builder sizeMetricsEnabled(boolean sizeMetricsEnabled) {
this.sizeMetricsEnabled = Optional.of(sizeMetricsEnabled);
@@ -40,12 +41,19 @@ public class RabbitMQMailQueueConfiguration {
return this;
}
+ public Builder mailQueuePublishConfirmEnabled(Boolean mailQueuePublishConfirmEnabled) {
+ this.mailQueuePublishConfirmEnabled = Optional.ofNullable(mailQueuePublishConfirmEnabled);
+ return this;
+ }
+
public RabbitMQMailQueueConfiguration build() {
- return new RabbitMQMailQueueConfiguration(sizeMetricsEnabled.orElse(DEFAULT_SIZE_METRICS_DISABLED));
+ return new RabbitMQMailQueueConfiguration(sizeMetricsEnabled.orElse(DEFAULT_SIZE_METRICS_DISABLED),
+ mailQueuePublishConfirmEnabled.orElse(true));
}
}
public static final String SIZE_METRICS_ENABLED_PROPERTY = "mailqueue.size.metricsEnabled";
+ private static final String MAIL_QUEUE_PUBLISH_CONFIRM_ENABLED = "mailqueue.publish.confirm.enabled";
public static Builder builder() {
return new Builder();
@@ -54,6 +62,7 @@ public class RabbitMQMailQueueConfiguration {
public static RabbitMQMailQueueConfiguration from(Configuration configuration) {
return builder()
.sizeMetricsEnabled(Optional.ofNullable(configuration.getBoolean(SIZE_METRICS_ENABLED_PROPERTY, null)))
+ .mailQueuePublishConfirmEnabled(configuration.getBoolean(MAIL_QUEUE_PUBLISH_CONFIRM_ENABLED, null))
.build();
}
@@ -70,9 +79,15 @@ public class RabbitMQMailQueueConfiguration {
}
private final boolean sizeMetricsEnabled;
+ private final boolean mailQueuePublishConfirmEnabled;
- private RabbitMQMailQueueConfiguration(boolean sizeMetricsEnabled) {
+ private RabbitMQMailQueueConfiguration(boolean sizeMetricsEnabled, boolean mailQueuePublishConfirmEnabled) {
this.sizeMetricsEnabled = sizeMetricsEnabled;
+ this.mailQueuePublishConfirmEnabled = mailQueuePublishConfirmEnabled;
+ }
+
+ public boolean isMailQueuePublishConfirmEnabled() {
+ return mailQueuePublishConfirmEnabled;
}
public boolean isSizeMetricsEnabled() {
@@ -84,13 +99,14 @@ public class RabbitMQMailQueueConfiguration {
if (o instanceof RabbitMQMailQueueConfiguration) {
RabbitMQMailQueueConfiguration that = (RabbitMQMailQueueConfiguration) o;
- return Objects.equals(this.sizeMetricsEnabled, that.sizeMetricsEnabled);
+ return Objects.equals(this.sizeMetricsEnabled, that.sizeMetricsEnabled)
+ && Objects.equals(this.mailQueuePublishConfirmEnabled, that.mailQueuePublishConfirmEnabled);
}
return false;
}
@Override
public final int hashCode() {
- return Objects.hash(sizeMetricsEnabled);
+ return Objects.hash(sizeMetricsEnabled, mailQueuePublishConfirmEnabled);
}
}
diff --git a/src/site/xdoc/server/config-rabbitmq.xml b/src/site/xdoc/server/config-rabbitmq.xml
index 17d5781a88..bbb47fbd9a 100644
--- a/src/site/xdoc/server/config-rabbitmq.xml
+++ b/src/site/xdoc/server/config-rabbitmq.xml
@@ -128,6 +128,16 @@
<dd>Optional integer, defaults is 3600000.
This is used only on queues used to share notification patterns, are exclusive to a node. If omitted, it will not add the TTL configure when declaring queues.
Configure queue ttl (in ms). References: https://www.rabbitmq.com/ttl.html#queue-ttl.</dd>
+
+ <dt><strong>mailqueue.publish.confirm.enabled</strong></dt>
+ <dd>Whether or not to enable publish confirms for the mail queue. Optional boolean, defaults to true.</dd>
+
+ <dt><strong>event.bus.publish.confirm.enabled</strong></dt>
+ <dd>Whether or not to enable publish confirms for the event bus. Optional boolean, defaults to true.</dd>
+
+ <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>
+
</dl>
</section>
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org