You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by hq...@apache.org on 2026/09/10 02:18:32 UTC
(james-project) 01/07: JAMES-4209 Extract Recovery info writing in a dedicated class
This is an automated email from the ASF dual-hosted git repository.
quantranhong1999 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 54a0dd66694e6f6d21e8cc1028d1a718cc6d2020
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Fri Sep 4 15:03:19 2026 +0200
JAMES-4209 Extract Recovery info writing in a dedicated class
---
.../cassandra/mail/CassandraMessageDAOV3.java | 40 +++-------
.../mail/ContentRecoveryMessageContentSaver.java | 87 ++++++++++++++++++++++
.../cassandra/mail/DefaultMessageContentSaver.java | 53 +++++++++++++
.../cassandra/mail/MessageContentSaver.java | 39 ++++++++++
4 files changed, 191 insertions(+), 28 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java
index 0b2eea3657..e5032a139f 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java
@@ -41,7 +41,6 @@ import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.T
import java.io.IOException;
import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.Optional;
@@ -68,8 +67,6 @@ import org.apache.james.mailbox.model.MessageAttachmentMetadata;
import org.apache.james.mailbox.model.StringBackedAttachmentId;
import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.CqlSession;
@@ -92,13 +89,12 @@ import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;
public class CassandraMessageDAOV3 {
- private static final Logger LOGGER = LoggerFactory.getLogger(CassandraMessageDAOV3.class);
private static final byte[] EMPTY_BYTE_ARRAY = {};
private final CassandraAsyncExecutor cassandraAsyncExecutor;
private final BlobStore blobStore;
- private final BlobStoreDAO blobStoreDAO;
private final BlobId.Factory blobIdFactory;
+ private final MessageContentSaver messageContentSaver;
private final PreparedStatement insert;
private final PreparedStatement delete;
private final PreparedStatement select;
@@ -117,8 +113,8 @@ public class CassandraMessageDAOV3 {
CassandraConfiguration cassandraConfiguration) {
this.cassandraAsyncExecutor = new CassandraAsyncExecutor(session);
this.blobStore = blobStore;
- this.blobStoreDAO = blobStoreDAO;
this.blobIdFactory = blobIdFactory;
+ this.messageContentSaver = messageContentSaver(blobStore, blobStoreDAO, blobIdFactory, cassandraConfiguration);
this.insert = prepareInsert(session);
this.delete = prepareDelete(session);
@@ -134,6 +130,15 @@ public class CassandraMessageDAOV3 {
this.optimisticConsistencyLevelProfile = JamesExecutionProfiles.getOptimisticConsistencyLevelProfile(session);
}
+ private static MessageContentSaver messageContentSaver(BlobStore blobStore, BlobStoreDAO blobStoreDAO,
+ BlobId.Factory blobIdFactory, CassandraConfiguration configuration) {
+ return switch (configuration.getBlobRecoveryMode()) {
+ case NONE -> new DefaultMessageContentSaver(blobStore);
+ case SYNCHRONOUS, ASYNCHRONOUS -> new ContentRecoveryMessageContentSaver(blobStore, blobStoreDAO,
+ blobIdFactory, configuration.getBlobRecoveryMode());
+ };
+ }
+
private PreparedStatement prepareSelect(CqlSession session) {
return session.prepare(selectFrom(TABLE_NAME)
.all()
@@ -210,31 +215,10 @@ public class CassandraMessageDAOV3 {
}
};
- Mono<BlobId> headerFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), headerContent, SIZE_BASED));
- Mono<BlobId> bodyFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), bodyByteSource, LOW_COST));
-
- return headerFuture.zipWith(bodyFuture)
- .flatMap(pair -> saveRecovery(pair.getT1(), pair.getT2()).thenReturn(pair));
+ return messageContentSaver.saveContent(headerContent, bodyByteSource);
});
}
- private Mono<Void> saveRecovery(BlobId headerId, BlobId bodyId) {
- return switch (configuration.getBlobRecoveryMode()) {
- case NONE -> Mono.empty();
- case SYNCHRONOUS -> writeRecoveryBlob(headerId, bodyId);
- case ASYNCHRONOUS -> Mono.fromRunnable(() ->
- writeRecoveryBlob(headerId, bodyId)
- .subscribeOn(Schedulers.parallel())
- .subscribe(ignored -> { }, e -> LOGGER.error("Failed to save recovery blob for header={} body={}", headerId.asString(), bodyId.asString(), e)));
- };
- }
-
- private Mono<Void> writeRecoveryBlob(BlobId headerId, BlobId bodyId) {
- BlobId recoveryBlobId = blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + headerId.asString());
- BlobStoreDAO.BytesBlob content = BlobStoreDAO.BytesBlob.of(bodyId.asString().getBytes(StandardCharsets.UTF_8));
- return Mono.from(blobStoreDAO.save(blobStore.getDefaultBucketName(), recoveryBlobId, content));
- }
-
private BoundStatement boundWriteStatement(MailboxMessage message, Tuple2<BlobId, BlobId> pair) {
CassandraMessageId messageId = (CassandraMessageId) message.getMessageId();
BoundStatement boundStatement = insert.bind()
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaver.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaver.java
new file mode 100644
index 0000000000..74c9175790
--- /dev/null
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaver.java
@@ -0,0 +1,87 @@
+/****************************************************************
+ * 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.mailbox.cassandra.mail;
+
+import java.nio.charset.StandardCharsets;
+
+import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration.BlobRecoveryMode;
+import org.apache.james.blob.api.BlobId;
+import org.apache.james.blob.api.BlobStore;
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+import com.google.common.io.ByteSource;
+
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+import reactor.util.function.Tuple2;
+
+/**
+ * Delegates the content write, then materializes the recovery information as a sidecar blob:
+ * the body blob id, stored under the header blob id prefixed by {@link BlobStoreDAO#RECOVERY_BLOB_PREFIX}.
+ *
+ * The sidecar write is either awaited ({@link BlobRecoveryMode#SYNCHRONOUS}) or performed on the side
+ * ({@link BlobRecoveryMode#ASYNCHRONOUS}).
+ */
+public class ContentRecoveryMessageContentSaver implements MessageContentSaver {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ContentRecoveryMessageContentSaver.class);
+
+ private final MessageContentSaver delegate;
+ private final BlobStore blobStore;
+ private final BlobStoreDAO blobStoreDAO;
+ private final BlobId.Factory blobIdFactory;
+ private final BlobRecoveryMode recoveryMode;
+
+ public ContentRecoveryMessageContentSaver(BlobStore blobStore, BlobStoreDAO blobStoreDAO,
+ BlobId.Factory blobIdFactory, BlobRecoveryMode recoveryMode) {
+ Preconditions.checkArgument(recoveryMode != BlobRecoveryMode.NONE,
+ "%s does not handle %s: rely on the delegate alone instead", ContentRecoveryMessageContentSaver.class.getSimpleName(), BlobRecoveryMode.NONE);
+ this.delegate = new DefaultMessageContentSaver(blobStore);
+ this.blobStore = blobStore;
+ this.blobStoreDAO = blobStoreDAO;
+ this.blobIdFactory = blobIdFactory;
+ this.recoveryMode = recoveryMode;
+ }
+
+ @Override
+ public Mono<Tuple2<BlobId, BlobId>> saveContent(byte[] headerBytes, ByteSource bodyByteSource) {
+ return delegate.saveContent(headerBytes, bodyByteSource)
+ .flatMap(pair -> saveRecovery(pair.getT1(), pair.getT2()).thenReturn(pair));
+ }
+
+ private Mono<Void> saveRecovery(BlobId headerId, BlobId bodyId) {
+ return switch (recoveryMode) {
+ case NONE -> Mono.empty();
+ case SYNCHRONOUS -> writeRecoveryBlob(headerId, bodyId);
+ case ASYNCHRONOUS -> Mono.fromRunnable(() ->
+ writeRecoveryBlob(headerId, bodyId)
+ .subscribeOn(Schedulers.parallel())
+ .subscribe(ignored -> { }, e -> LOGGER.error("Failed to save recovery blob for header={} body={}", headerId.asString(), bodyId.asString(), e)));
+ };
+ }
+
+ private Mono<Void> writeRecoveryBlob(BlobId headerId, BlobId bodyId) {
+ BlobId recoveryBlobId = blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + headerId.asString());
+ BlobStoreDAO.BytesBlob content = BlobStoreDAO.BytesBlob.of(bodyId.asString().getBytes(StandardCharsets.UTF_8));
+ return Mono.from(blobStoreDAO.save(blobStore.getDefaultBucketName(), recoveryBlobId, content));
+ }
+}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/DefaultMessageContentSaver.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/DefaultMessageContentSaver.java
new file mode 100644
index 0000000000..b575547df1
--- /dev/null
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/DefaultMessageContentSaver.java
@@ -0,0 +1,53 @@
+/****************************************************************
+ * 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.mailbox.cassandra.mail;
+
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED;
+
+import jakarta.inject.Inject;
+
+import org.apache.james.blob.api.BlobId;
+import org.apache.james.blob.api.BlobStore;
+
+import com.google.common.io.ByteSource;
+
+import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+
+/**
+ * Writes the headers and the body as two distinct blobs, without any recovery information.
+ */
+public class DefaultMessageContentSaver implements MessageContentSaver {
+ private final BlobStore blobStore;
+
+ @Inject
+ public DefaultMessageContentSaver(BlobStore blobStore) {
+ this.blobStore = blobStore;
+ }
+
+ @Override
+ public Mono<Tuple2<BlobId, BlobId>> saveContent(byte[] headerBytes, ByteSource bodyByteSource) {
+ Mono<BlobId> headerFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), headerBytes, SIZE_BASED));
+ Mono<BlobId> bodyFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), bodyByteSource, LOW_COST));
+
+ return headerFuture.zipWith(bodyFuture);
+ }
+}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/MessageContentSaver.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/MessageContentSaver.java
new file mode 100644
index 0000000000..682734a15e
--- /dev/null
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/MessageContentSaver.java
@@ -0,0 +1,39 @@
+/****************************************************************
+ * 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.mailbox.cassandra.mail;
+
+import org.apache.james.blob.api.BlobId;
+
+import com.google.common.io.ByteSource;
+
+import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+
+/**
+ * Saves the content of a message: its headers and its body.
+ *
+ * Implementations decide which recovery policy, if any, is applied alongside the content write.
+ */
+public interface MessageContentSaver {
+ /**
+ * @return the blob id of the headers (T1) and the blob id of the body (T2).
+ */
+ Mono<Tuple2<BlobId, BlobId>> saveContent(byte[] headerBytes, ByteSource bodyByteSource);
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org