This is an automated email from the ASF dual-hosted git repository. btellier pushed a commit to branch dtm-workqueue in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 7082160d7a6f95fc143e32efed21539eac3f4a80 Author: Benoit TELLIER <btell...@linagora.com> AuthorDate: Mon Mar 18 22:07:05 2024 +0100 [ENHANCEMENT] Workqueue for the deleted message vault Deleted message vault copy can be long, doing it asynchronously would make the feature more reliable. --- .../mailbox/cassandra/DeleteMessageListener.java | 67 +++++- .../DeletedMessageVaultDeletionCallback.java | 24 +- .../james/CassandraRabbitMQJamesServerMain.java | 4 +- .../james/DistributedPOP3JamesServerMain.java | 4 +- ...ributedDeletedMessageVaultDeletionCallback.java | 262 +++++++++++++++++++++ .../DistributedDeletedMessageVaultModule.java | 69 ++++++ 6 files changed, 412 insertions(+), 18 deletions(-) diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java index 504b07c2f5..0b091ba330 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java @@ -23,6 +23,7 @@ import static org.apache.james.backends.cassandra.init.configuration.JamesExecut import static org.apache.james.backends.cassandra.init.configuration.JamesExecutionProfiles.ConsistencyChoice.WEAK; import static org.apache.james.util.FunctionalUtils.negate; +import java.util.Date; import java.util.Optional; import java.util.Set; @@ -30,6 +31,7 @@ import javax.inject.Inject; import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration; import org.apache.james.backends.cassandra.init.configuration.JamesExecutionProfiles; +import org.apache.james.blob.api.BlobId; import org.apache.james.blob.api.BlobStore; import org.apache.james.core.Username; import org.apache.james.events.Event; @@ -59,6 +61,7 @@ import org.apache.james.mailbox.model.ComposedMessageIdWithMetaData; import org.apache.james.mailbox.model.MailboxACL; import org.apache.james.mailbox.model.MailboxId; import org.apache.james.mailbox.model.MailboxPath; +import org.apache.james.mailbox.model.MessageId; import org.apache.james.mailbox.model.MessageMetaData; import org.apache.james.mailbox.model.MessageRange; import org.apache.james.mailbox.store.mail.MessageMapper; @@ -86,9 +89,71 @@ public class DeleteMessageListener implements EventListener.ReactiveGroupEventLi } + public static class DelectedMessageCopyCommand { + public static DelectedMessageCopyCommand of(MessageRepresentation message, MailboxId mailboxId, Username owner) { + return new DelectedMessageCopyCommand(message.getMessageId(), mailboxId, owner, message.getInternalDate(), + message.getSize(), !message.getAttachments().isEmpty(), message.getHeaderId(), message.getBodyId()); + } + + private final MessageId messageId; + private final MailboxId mailboxId; + private final Username owner; + private final Date internalDate; + private final long size; + private final boolean hasAttachments; + private final BlobId headerId; + private final BlobId bodyId; + + public DelectedMessageCopyCommand(MessageId messageId, MailboxId mailboxId, Username owner, Date internalDate, long size, boolean hasAttachments, BlobId headerId, BlobId bodyId) { + this.messageId = messageId; + this.mailboxId = mailboxId; + this.owner = owner; + this.internalDate = internalDate; + this.size = size; + this.hasAttachments = hasAttachments; + this.headerId = headerId; + this.bodyId = bodyId; + } + + public Username getOwner() { + return owner; + } + + public MessageId getMessageId() { + return messageId; + } + + public MailboxId getMailboxId() { + return mailboxId; + } + + public Date getInternalDate() { + return internalDate; + } + + public long getSize() { + return size; + } + + public boolean hasAttachments() { + return hasAttachments; + } + + public BlobId getHeaderId() { + return headerId; + } + + public BlobId getBodyId() { + return bodyId; + } + } + @FunctionalInterface public interface DeletionCallback { - Mono<Void> forMessage(MessageRepresentation message, MailboxId mailboxId, Username owner); + default Mono<Void> forMessage(MessageRepresentation message, MailboxId mailboxId, Username owner) { + return forMessage(DelectedMessageCopyCommand.of(message, mailboxId, owner)); + } + Mono<Void> forMessage(DelectedMessageCopyCommand copyCommand); } private final CassandraThreadDAO threadDAO; diff --git a/mailbox/plugin/deleted-messages-vault-cassandra/src/main/java/org/apache/james/vault/metadata/DeletedMessageVaultDeletionCallback.java b/mailbox/plugin/deleted-messages-vault-cassandra/src/main/java/org/apache/james/vault/metadata/DeletedMessageVaultDeletionCallback.java index b23a895f20..611fa69779 100644 --- a/mailbox/plugin/deleted-messages-vault-cassandra/src/main/java/org/apache/james/vault/metadata/DeletedMessageVaultDeletionCallback.java +++ b/mailbox/plugin/deleted-messages-vault-cassandra/src/main/java/org/apache/james/vault/metadata/DeletedMessageVaultDeletionCallback.java @@ -34,10 +34,7 @@ import javax.inject.Inject; import org.apache.james.blob.api.BlobStore; import org.apache.james.core.MailAddress; import org.apache.james.core.MaybeSender; -import org.apache.james.core.Username; import org.apache.james.mailbox.cassandra.DeleteMessageListener; -import org.apache.james.mailbox.cassandra.mail.MessageRepresentation; -import org.apache.james.mailbox.model.MailboxId; import org.apache.james.mailbox.model.MessageId; import org.apache.james.mime4j.MimeIOException; import org.apache.james.mime4j.codec.DecodeMonitor; @@ -71,29 +68,30 @@ public class DeletedMessageVaultDeletionCallback implements DeleteMessageListene } @Override - public Mono<Void> forMessage(MessageRepresentation message, MailboxId mailboxId, Username owner) { - return Mono.from(blobStore.readBytes(blobStore.getDefaultBucketName(), message.getHeaderId(), BlobStore.StoragePolicy.LOW_COST)) + public Mono<Void> forMessage(DeleteMessageListener.DelectedMessageCopyCommand copyCommand) { + return Mono.from(blobStore.readBytes(blobStore.getDefaultBucketName(), copyCommand.getHeaderId(), BlobStore.StoragePolicy.LOW_COST)) .flatMap(bytes -> { - Optional<Message> mimeMessage = parseMessage(new ByteArrayInputStream(bytes), message.getMessageId()); + Optional<Message> mimeMessage = parseMessage(new ByteArrayInputStream(bytes), copyCommand.getMessageId()); DeletedMessage deletedMessage = DeletedMessage.builder() - .messageId(message.getMessageId()) - .originMailboxes(mailboxId) - .user(owner) - .deliveryDate(ZonedDateTime.ofInstant(message.getInternalDate().toInstant(), ZoneOffset.UTC)) + .messageId(copyCommand.getMessageId()) + .originMailboxes(copyCommand.getMailboxId()) + .user(copyCommand.getOwner()) + .deliveryDate(ZonedDateTime.ofInstant(copyCommand.getInternalDate().toInstant(), ZoneOffset.UTC)) .deletionDate(ZonedDateTime.ofInstant(clock.instant(), ZoneOffset.UTC)) .sender(retrieveSender(mimeMessage)) .recipients(retrieveRecipients(mimeMessage)) - .hasAttachment(!message.getAttachments().isEmpty()) - .size(message.getSize()) + .hasAttachment(copyCommand.hasAttachments()) + .size(copyCommand.getSize()) .subject(mimeMessage.map(Message::getSubject)) .build(); - return Mono.from(blobStore.readReactive(blobStore.getDefaultBucketName(), message.getBodyId(), BlobStore.StoragePolicy.LOW_COST)) + return Mono.from(blobStore.readReactive(blobStore.getDefaultBucketName(), copyCommand.getBodyId(), BlobStore.StoragePolicy.LOW_COST)) .map(bodyStream -> new SequenceInputStream(new ByteArrayInputStream(bytes), bodyStream)) .flatMap(bodyStream -> Mono.from(deletedMessageVault.append(deletedMessage, bodyStream))); }); } + private Optional<Message> parseMessage(InputStream inputStream, MessageId messageId) { DefaultMessageBuilder messageBuilder = new DefaultMessageBuilder(); messageBuilder.setMimeEntityConfig(MimeConfig.PERMISSIVE); diff --git a/server/apps/distributed-app/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java b/server/apps/distributed-app/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java index d5b27726c9..794d210337 100644 --- a/server/apps/distributed-app/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java +++ b/server/apps/distributed-app/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java @@ -51,12 +51,12 @@ import org.apache.james.modules.data.CassandraVacationModule; import org.apache.james.modules.event.JMAPEventBusModule; import org.apache.james.modules.event.RabbitMQEventBusModule; import org.apache.james.modules.eventstore.CassandraEventStoreModule; -import org.apache.james.modules.mailbox.CassandraDeletedMessageVaultModule; import org.apache.james.modules.mailbox.CassandraMailboxModule; import org.apache.james.modules.mailbox.CassandraMailboxQuotaLegacyModule; import org.apache.james.modules.mailbox.CassandraMailboxQuotaModule; import org.apache.james.modules.mailbox.CassandraQuotaMailingModule; import org.apache.james.modules.mailbox.CassandraSessionModule; +import org.apache.james.modules.mailbox.DistributedDeletedMessageVaultModule; import org.apache.james.modules.mailbox.TikaMailboxModule; import org.apache.james.modules.mailrepository.CassandraMailRepositoryModule; import org.apache.james.modules.metrics.CassandraMetricsModule; @@ -228,7 +228,7 @@ public class CassandraRabbitMQJamesServerMain implements JamesServerMain { private static Module chooseDeletedMessageVault(VaultConfiguration vaultConfiguration) { if (vaultConfiguration.isEnabled()) { return Modules.combine( - new CassandraDeletedMessageVaultModule(), + new DistributedDeletedMessageVaultModule(), new DeletedMessageVaultRoutesModule()); } return binder -> { diff --git a/server/apps/distributed-pop3-app/src/main/java/org/apache/james/DistributedPOP3JamesServerMain.java b/server/apps/distributed-pop3-app/src/main/java/org/apache/james/DistributedPOP3JamesServerMain.java index d094aaab94..d42cc35d11 100644 --- a/server/apps/distributed-pop3-app/src/main/java/org/apache/james/DistributedPOP3JamesServerMain.java +++ b/server/apps/distributed-pop3-app/src/main/java/org/apache/james/DistributedPOP3JamesServerMain.java @@ -59,11 +59,11 @@ import org.apache.james.modules.event.JMAPEventBusModule; import org.apache.james.modules.event.RabbitMQEventBusModule; import org.apache.james.modules.eventstore.CassandraEventStoreModule; import org.apache.james.modules.mailbox.CassandraBlobStoreDependenciesModule; -import org.apache.james.modules.mailbox.CassandraDeletedMessageVaultModule; import org.apache.james.modules.mailbox.CassandraMailboxModule; import org.apache.james.modules.mailbox.CassandraMailboxQuotaLegacyModule; import org.apache.james.modules.mailbox.CassandraMailboxQuotaModule; import org.apache.james.modules.mailbox.CassandraSessionModule; +import org.apache.james.modules.mailbox.DistributedDeletedMessageVaultModule; import org.apache.james.modules.mailbox.TikaMailboxModule; import org.apache.james.modules.mailrepository.CassandraMailRepositoryModule; import org.apache.james.modules.metrics.CassandraMetricsModule; @@ -218,7 +218,7 @@ public class DistributedPOP3JamesServerMain implements JamesServerMain { private static Module chooseDeletedMessageVault(VaultConfiguration vaultConfiguration) { if (vaultConfiguration.isEnabled()) { return Modules.combine( - new CassandraDeletedMessageVaultModule(), + new DistributedDeletedMessageVaultModule(), new DeletedMessageVaultRoutesModule()); } return binder -> { diff --git a/server/container/guice/distributed/src/main/java/org/apache/james/modules/mailbox/DistributedDeletedMessageVaultDeletionCallback.java b/server/container/guice/distributed/src/main/java/org/apache/james/modules/mailbox/DistributedDeletedMessageVaultDeletionCallback.java new file mode 100644 index 0000000000..282f9007b4 --- /dev/null +++ b/server/container/guice/distributed/src/main/java/org/apache/james/modules/mailbox/DistributedDeletedMessageVaultDeletionCallback.java @@ -0,0 +1,262 @@ +/**************************************************************** + * 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.modules.mailbox; + +import static com.rabbitmq.client.MessageProperties.PERSISTENT_TEXT_PLAIN; +import static org.apache.james.backends.rabbitmq.Constants.ALLOW_QUORUM; +import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE; +import static org.apache.james.backends.rabbitmq.Constants.DIRECT_EXCHANGE; +import static org.apache.james.backends.rabbitmq.Constants.DURABLE; +import static org.apache.james.backends.rabbitmq.Constants.EMPTY_ROUTING_KEY; +import static org.apache.james.backends.rabbitmq.Constants.EXCLUSIVE; + +import java.util.Date; +import java.util.Optional; + +import javax.inject.Inject; + +import org.apache.james.backends.rabbitmq.RabbitMQConfiguration; +import org.apache.james.backends.rabbitmq.ReactorRabbitMQChannelPool; +import org.apache.james.blob.api.BlobId; +import org.apache.james.core.Username; +import org.apache.james.mailbox.cassandra.DeleteMessageListener; +import org.apache.james.mailbox.model.MailboxId; +import org.apache.james.mailbox.model.MessageId; +import org.apache.james.vault.metadata.DeletedMessageVaultDeletionCallback; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.rabbitmq.client.AMQP; + +import reactor.core.Disposable; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; +import reactor.rabbitmq.AcknowledgableDelivery; +import reactor.rabbitmq.BindingSpecification; +import reactor.rabbitmq.ConsumeOptions; +import reactor.rabbitmq.ExchangeSpecification; +import reactor.rabbitmq.OutboundMessage; +import reactor.rabbitmq.QueueSpecification; +import reactor.rabbitmq.Receiver; +import reactor.rabbitmq.Sender; + +public class DistributedDeletedMessageVaultDeletionCallback implements DeleteMessageListener.DeletionCallback { + public static final Logger LOGGER = LoggerFactory.getLogger(DistributedDeletedMessageVaultDeletionCallback.class); + + private static class CopyCommandDTO { + public static CopyCommandDTO of(DeleteMessageListener.DelectedMessageCopyCommand command) { + return new CopyCommandDTO( + command.getMessageId().serialize(), + command.getMailboxId().serialize(), + command.getOwner().asString(), + command.getInternalDate(), + command.getSize(), + command.hasAttachments(), + command.getHeaderId().asString(), + command.getBodyId().asString()); + } + + private final String messageId; + private final String mailboxId; + private final String owner; + private final Date internalDate; + private final long size; + private final boolean hasAttachments; + private final String headerId; + private final String bodyId; + + @JsonCreator + public CopyCommandDTO(@JsonProperty("messageId") String messageId, + @JsonProperty("mailboxId") String mailboxId, + @JsonProperty("owner") String owner, + @JsonProperty("internalDate") Date internalDate, + @JsonProperty("size") long size, + @JsonProperty("hasAttachments") boolean hasAttachments, + @JsonProperty("headerId") String headerId, + @JsonProperty("bodyId") String bodyId) { + this.messageId = messageId; + this.mailboxId = mailboxId; + this.owner = owner; + this.internalDate = internalDate; + this.size = size; + this.hasAttachments = hasAttachments; + this.headerId = headerId; + this.bodyId = bodyId; + } + + public String getMessageId() { + return messageId; + } + + public String getMailboxId() { + return mailboxId; + } + + public String getOwner() { + return owner; + } + + public Date getInternalDate() { + return internalDate; + } + + public long getSize() { + return size; + } + + public boolean isHasAttachments() { + return hasAttachments; + } + + public String getHeaderId() { + return headerId; + } + + public String getBodyId() { + return bodyId; + } + + @JsonIgnore + DeleteMessageListener.DelectedMessageCopyCommand asPojo(MailboxId.Factory mailboxIdFactory, MessageId.Factory messageIdFactory, BlobId.Factory blobIdFactory) { + return new DeleteMessageListener.DelectedMessageCopyCommand(messageIdFactory.fromString(messageId), + mailboxIdFactory.fromString(messageId), + Username.of(owner), + internalDate, + size, + hasAttachments, + blobIdFactory.from(headerId), + blobIdFactory.from(bodyId)); + } + } + + private static final String EXCHANGE = "deleted-message-vault"; + private static final String QUEUE = "deleted-message-vault-work-queue"; + private static final String DEAD_LETTER = QUEUE + "-dead-letter"; + private static final boolean REQUEUE = true; + private static final int QOS = 5; + + private final ReactorRabbitMQChannelPool channelPool; + private final RabbitMQConfiguration rabbitMQConfiguration; + private final DeletedMessageVaultDeletionCallback callback; + private final Sender sender; + private final ObjectMapper objectMapper; + private final MailboxId.Factory mailboxIdFactory; + private final MessageId.Factory messageIdFactory; + private final BlobId.Factory blobIdFactory; + private Receiver receiver; + private Disposable disposable; + + @Inject + public DistributedDeletedMessageVaultDeletionCallback(Sender sender, + ReactorRabbitMQChannelPool channelPool, + RabbitMQConfiguration rabbitMQConfiguration, + DeletedMessageVaultDeletionCallback callback, + MailboxId.Factory mailboxIdFactory, + MessageId.Factory messageIdFactory, + BlobId.Factory blobIdFactory) { + this.sender = sender; + this.rabbitMQConfiguration = rabbitMQConfiguration; + this.callback = callback; + this.mailboxIdFactory = mailboxIdFactory; + this.messageIdFactory = messageIdFactory; + this.blobIdFactory = blobIdFactory; + this.objectMapper = new ObjectMapper(); + this.channelPool = channelPool; + } + + public void init() { + Flux.concat( + sender.declareExchange(ExchangeSpecification.exchange(EXCHANGE) + .durable(DURABLE) + .type(DIRECT_EXCHANGE)), + sender.declareQueue(QueueSpecification.queue(DEAD_LETTER) + .durable(DURABLE) + .exclusive(!EXCLUSIVE) + .autoDelete(!AUTO_DELETE) + .arguments(rabbitMQConfiguration.workQueueArgumentsBuilder(!ALLOW_QUORUM) + .deadLetter(DEAD_LETTER) + .build())), + sender.declareQueue(QueueSpecification.queue(QUEUE) + .durable(DURABLE) + .exclusive(!EXCLUSIVE) + .autoDelete(!AUTO_DELETE) + .arguments(rabbitMQConfiguration.workQueueArgumentsBuilder(!ALLOW_QUORUM).build())), + sender.bind(BindingSpecification.binding() + .exchange(EXCHANGE) + .queue(QUEUE) + .routingKey(EMPTY_ROUTING_KEY))) + .then() + .block(); + + receiver = channelPool.createReceiver(); + disposable = receiver.consumeManualAck(QUEUE, new ConsumeOptions().qos(QOS)) + .map(delivery -> Mono.fromCallable(delivery::getBody) + .flatMap(bytes -> handleMessage(delivery, bytes))) + .subscribeOn(Schedulers.boundedElastic()) + .subscribe(); + } + + public void stop() { + Optional.ofNullable(disposable).ifPresent(Disposable::dispose); + Optional.ofNullable(receiver).ifPresent(Receiver::close); + } + + private Mono<Void> handleMessage(AcknowledgableDelivery delivery, byte[] bytes) { + try { + CopyCommandDTO copyCommandDTO = objectMapper.readValue(bytes, CopyCommandDTO.class); + + return callback.forMessage(copyCommandDTO.asPojo(mailboxIdFactory, messageIdFactory, blobIdFactory)) + .doOnError(e -> { + LOGGER.error("Failed executing deletion callback for {}", copyCommandDTO.messageId, e); + delivery.nack(REQUEUE); + }) + .doOnSuccess(any -> delivery.ack()) + .doOnCancel(() -> delivery.nack(REQUEUE)); + } catch (Exception e) { + LOGGER.error(" Deserialization error: reject poisonous message for distributed Deleted message vault callback", e); + // Deserialization error: reject poisonous messages + delivery.nack(!REQUEUE); + return Mono.empty(); + } + } + + @Override + public Mono<Void> forMessage(DeleteMessageListener.DelectedMessageCopyCommand command) { + CopyCommandDTO dto = CopyCommandDTO.of(command); + try { + byte[] bytes = objectMapper.writeValueAsBytes(dto); + return sender.send(Mono.just(new OutboundMessage(EXCHANGE, EMPTY_ROUTING_KEY, new AMQP.BasicProperties.Builder() + .deliveryMode(PERSISTENT_TEXT_PLAIN.getDeliveryMode()) + .priority(PERSISTENT_TEXT_PLAIN.getPriority()) + .contentType(PERSISTENT_TEXT_PLAIN.getContentType()) + .build(), bytes))); + } catch (JsonProcessingException e) { + return Mono.error(e); + } + } + + +} diff --git a/server/container/guice/distributed/src/main/java/org/apache/james/modules/mailbox/DistributedDeletedMessageVaultModule.java b/server/container/guice/distributed/src/main/java/org/apache/james/modules/mailbox/DistributedDeletedMessageVaultModule.java new file mode 100644 index 0000000000..25bd0fd7cd --- /dev/null +++ b/server/container/guice/distributed/src/main/java/org/apache/james/modules/mailbox/DistributedDeletedMessageVaultModule.java @@ -0,0 +1,69 @@ +/**************************************************************** + * 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.modules.mailbox; + +import org.apache.james.backends.cassandra.components.CassandraModule; +import org.apache.james.mailbox.cassandra.DeleteMessageListener; +import org.apache.james.modules.vault.DeletedMessageVaultModule; +import org.apache.james.vault.DeletedMessageVault; +import org.apache.james.vault.blob.BlobStoreDeletedMessageVault; +import org.apache.james.vault.blob.BucketNameGenerator; +import org.apache.james.vault.dto.DeletedMessageWithStorageInformationConverter; +import org.apache.james.vault.metadata.CassandraDeletedMessageMetadataVault; +import org.apache.james.vault.metadata.DeletedMessageMetadataModule; +import org.apache.james.vault.metadata.DeletedMessageMetadataVault; +import org.apache.james.vault.metadata.DeletedMessageVaultDeletionCallback; +import org.apache.james.vault.metadata.MetadataDAO; +import org.apache.james.vault.metadata.StorageInformationDAO; +import org.apache.james.vault.metadata.UserPerBucketDAO; + +import com.google.inject.AbstractModule; +import com.google.inject.Scopes; +import com.google.inject.multibindings.Multibinder; + +public class DistributedDeletedMessageVaultModule extends AbstractModule { + @Override + protected void configure() { + install(new DeletedMessageVaultModule()); + + Multibinder<CassandraModule> cassandraDataDefinitions = Multibinder.newSetBinder(binder(), CassandraModule.class); + cassandraDataDefinitions + .addBinding() + .toInstance(DeletedMessageMetadataModule.MODULE); + + bind(MetadataDAO.class).in(Scopes.SINGLETON); + bind(StorageInformationDAO.class).in(Scopes.SINGLETON); + bind(UserPerBucketDAO.class).in(Scopes.SINGLETON); + bind(DeletedMessageWithStorageInformationConverter.class).in(Scopes.SINGLETON); + + bind(CassandraDeletedMessageMetadataVault.class).in(Scopes.SINGLETON); + bind(DeletedMessageMetadataVault.class) + .to(CassandraDeletedMessageMetadataVault.class); + + bind(BucketNameGenerator.class).in(Scopes.SINGLETON); + bind(BlobStoreDeletedMessageVault.class).in(Scopes.SINGLETON); + bind(DeletedMessageVault.class) + .to(BlobStoreDeletedMessageVault.class); + + Multibinder.newSetBinder(binder(), DeleteMessageListener.DeletionCallback.class) + .addBinding() + .to(DistributedDeletedMessageVaultDeletionCallback.class); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: notifications-unsubscr...@james.apache.org For additional commands, e-mail: notifications-h...@james.apache.org