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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
