You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by bt...@apache.org on 2020/01/13 09:55:46 UTC
[james-project] branch master updated (f2d7aba -> e3c0b3f)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git.
from f2d7aba [Refactoring] AWSS3BlobPutter reuse S3Client and TransferManager
new bea796d JAMES-2921 BlobStore API should include a StorageStrategy
new 7cfa6fe JAMES-2921 Propose an Hybrid blobStore
new bf918b6 JAMES-2921 HybridBlobStore upgrade instructions
new 23d311d JAMES-2921 configure hybrid blobStore threshold
new e0508a7 JAMES-2921 Documentation changes
new 700dfec JAMES-2921 BlobStoreContract should test all storage policies basic behaviour
new b44bc7d JAMES-2921 Propose an Hybrid blobStore
new f22e18f JAMES-2921 s/swiftBlobStoreProvider/objectStorageBlobStoreProvider
new 694d36a [Refactoring] throwing tests for BlobStore should actually read() to ensure lazy streams are evaluated
new e3c0b3f [Refactoring] fix Reactor Intellij inspections
The 10 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../destination/conf/blob.properties | 10 +-
.../destination/conf/blob.properties | 10 +-
.../cassandra/mail/CassandraAttachmentMapper.java | 6 +-
.../cassandra/mail/CassandraMessageDAO.java | 6 +-
.../cassandra/mail/CassandraMessageIdMapper.java | 24 +-
.../cassandra/mail/CassandraMessageMapper.java | 29 +-
.../mail/migration/AttachmentV2Migration.java | 4 +-
.../mail/CassandraAttachmentFallbackTest.java | 7 +-
.../mail/migration/AttachmentV2MigrationTest.java | 16 +-
.../vault/blob/BlobStoreDeletedMessageVault.java | 5 +-
server/blob/blob-api/pom.xml | 5 +
.../java/org/apache/james/blob/api/BlobStore.java | 14 +-
.../apache/james/blob/api/MetricableBlobStore.java | 8 +-
.../main/java/org/apache/james/blob/api/Store.java | 6 +-
.../apache/james/blob/api/BlobStoreContract.java | 121 +++--
.../james/blob/api/BucketBlobStoreContract.java | 31 +-
.../james/blob/api/DeleteBlobStoreContract.java | 25 +-
.../blob/api/MetricableBlobStoreContract.java | 25 +-
server/blob/blob-cassandra/pom.xml | 5 +
.../james/blob/cassandra/CassandraBlobStore.java | 9 +-
.../blob/cassandra/CassandraBlobStoreTest.java | 9 +-
.../file/LocalFileBlobExportMechanismTest.java | 11 +-
server/blob/blob-memory/pom.xml | 5 +
.../apache/james/blob/memory/MemoryBlobStore.java | 6 +-
.../blob/objectstorage/ObjectStorageBlobStore.java | 10 +-
.../ObjectStorageBlobStoreContract.java | 3 +-
.../objectstorage/ObjectStorageBlobStoreTest.java | 15 +-
server/blob/blob-union/pom.xml | 4 +
.../apache/james/blob/union/HybridBlobStore.java | 231 ++++++++
.../apache/james/blob/union/UnionBlobStore.java | 218 --------
.../james/blob/union/HybridBlobStoreTest.java | 536 ++++++++++++++++++
.../james/blob/union/UnionBlobStoreTest.java | 603 ---------------------
.../apache/james/blob/mail/MimeMessageStore.java | 6 +-
.../swift/ObjectStorageBlobStoreModuleTest.java | 3 +-
.../blobstore/BlobStoreChoosingConfiguration.java | 6 +-
.../modules/blobstore/BlobStoreChoosingModule.java | 27 +-
.../BlobStoreChoosingConfigurationTest.java | 14 +-
.../blobstore/BlobStoreChoosingModuleTest.java | 72 ++-
.../memory/vacation/MemoryVacationRepository.java | 6 +-
.../james/webadmin/vault/routes/ExportService.java | 4 +-
src/site/xdoc/server/config-blobstore.xml | 9 +-
.../linshare/LinshareBlobExportMechanismTest.java | 9 +-
upgrade-instructions.md | 22 +-
43 files changed, 1169 insertions(+), 1026 deletions(-)
create mode 100644 server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
delete mode 100644 server/blob/blob-union/src/main/java/org/apache/james/blob/union/UnionBlobStore.java
create mode 100644 server/blob/blob-union/src/test/java/org/apache/james/blob/union/HybridBlobStoreTest.java
delete mode 100644 server/blob/blob-union/src/test/java/org/apache/james/blob/union/UnionBlobStoreTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 07/10: JAMES-2921 Propose an Hybrid blobStore
Posted by bt...@apache.org.
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 b44bc7d93901773ad442a341cb1ba2d94848c449
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jan 7 16:48:56 2020 +0700
JAMES-2921 Propose an Hybrid blobStore
---
.../main/java/org/apache/james/blob/union/HybridBlobStore.java | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
index 03328a0..93e1532 100644
--- a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
+++ b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
@@ -40,11 +40,11 @@ import reactor.core.publisher.Mono;
public class HybridBlobStore implements BlobStore {
@FunctionalInterface
public interface RequireLowCost {
- RequirePerforming lowCost(BlobStore blobStore);
+ RequireHighPerformance lowCost(BlobStore blobStore);
}
@FunctionalInterface
- public interface RequirePerforming {
+ public interface RequireHighPerformance {
RequireConfiguration highPerformance(BlobStore blobStore);
}
@@ -73,13 +73,13 @@ public class HybridBlobStore implements BlobStore {
}
public static class Configuration {
- public static final int DEFAULT_SIZE_THREASHOLD = 32 * 1024;
- public static final Configuration DEFAULT = new Configuration(DEFAULT_SIZE_THREASHOLD);
+ public static final int DEFAULT_SIZE_THRESHOLD = 32 * 1024;
+ public static final Configuration DEFAULT = new Configuration(DEFAULT_SIZE_THRESHOLD);
private static final String PROPERTY_NAME = "hybrid.size.threshold";
public static Configuration from(org.apache.commons.configuration2.Configuration propertiesConfiguration) {
return new Configuration(Optional.ofNullable(propertiesConfiguration.getInteger(PROPERTY_NAME, null))
- .orElse(DEFAULT_SIZE_THREASHOLD));
+ .orElse(DEFAULT_SIZE_THRESHOLD));
}
private final int sizeThreshold;
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 10/10: [Refactoring] fix Reactor Intellij
inspections
Posted by bt...@apache.org.
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 e3c0b3fd96e127bfb74fbe00c6cbac25d1d36856
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Fri Dec 13 14:47:34 2019 +0100
[Refactoring] fix Reactor Intellij inspections
---
.../cassandra/mail/CassandraMessageIdMapper.java | 24 ++++++++++--------
.../cassandra/mail/CassandraMessageMapper.java | 29 +++++++++++++++-------
.../james/blob/cassandra/CassandraBlobStore.java | 5 ++--
.../memory/vacation/MemoryVacationRepository.java | 6 ++---
4 files changed, 38 insertions(+), 26 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
index 1b15350..3a206c3 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
@@ -258,22 +258,24 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
private Mono<Pair<Flags, ComposedMessageIdWithMetaData>> updateFlags(MailboxId mailboxId, MessageId messageId, Flags newState, MessageManager.FlagsUpdateMode updateMode) throws MailboxException {
CassandraId cassandraId = (CassandraId) mailboxId;
- ComposedMessageIdWithMetaData oldComposedId = imapUidDAO.retrieve((CassandraMessageId) messageId, Optional.of(cassandraId))
- .next()
- .blockOptional()
- .orElseThrow(MailboxDeleteDuringUpdateException::new);
+ return imapUidDAO.retrieve((CassandraMessageId) messageId, Optional.of(cassandraId))
+ .single()
+ .switchIfEmpty(Mono.error(MailboxDeleteDuringUpdateException::new))
+ .flatMap(oldComposedId -> updateFlags(newState, updateMode, cassandraId, oldComposedId));
+ }
+ private Mono<Pair<Flags, ComposedMessageIdWithMetaData>> updateFlags(Flags newState, MessageManager.FlagsUpdateMode updateMode, CassandraId cassandraId, ComposedMessageIdWithMetaData oldComposedId) {
Flags newFlags = new FlagsUpdateCalculator(newState, updateMode).buildNewFlags(oldComposedId.getFlags());
if (identicalFlags(oldComposedId, newFlags)) {
return Mono.just(Pair.of(oldComposedId.getFlags(), oldComposedId));
+ } else {
+ return Mono
+ .fromCallable(() -> new ComposedMessageIdWithMetaData(
+ oldComposedId.getComposedMessageId(),
+ newFlags,
+ modSeqProvider.nextModSeq(cassandraId)))
+ .flatMap(newComposedId -> updateFlags(oldComposedId, newComposedId));
}
-
- ComposedMessageIdWithMetaData newComposedId = new ComposedMessageIdWithMetaData(
- oldComposedId.getComposedMessageId(),
- newFlags,
- modSeqProvider.nextModSeq(cassandraId));
-
- return updateFlags(oldComposedId, newComposedId);
}
private boolean identicalFlags(ComposedMessageIdWithMetaData oldComposedId, Flags newFlags) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
index ebc29a7..ed93cc2 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
@@ -274,15 +274,26 @@ public class CassandraMessageMapper implements MessageMapper {
}
private MailboxMessage addUidAndModseq(MailboxMessage message, CassandraId mailboxId) throws MailboxException {
- final Mono<MessageUid> messageUidMono = uidProvider.nextUid(mailboxId).cache();
- final Mono<ModSeq> nextModSeqMono = modSeqProvider.nextModSeq(mailboxId).cache();
- Flux.merge(messageUidMono, nextModSeqMono).then();
-
- message.setUid(messageUidMono.blockOptional()
- .orElseThrow(() -> new MailboxException("Can not find a UID to save " + message.getMessageId() + " in " + mailboxId)));
- message.setModSeq(nextModSeqMono.blockOptional()
- .orElseThrow(() -> new MailboxException("Can not find a MODSEQ to save " + message.getMessageId() + " in " + mailboxId)));
-
+ Mono<MessageUid> messageUidMono = uidProvider
+ .nextUid(mailboxId)
+ .switchIfEmpty(Mono.error(() -> new MailboxException("Can not find a UID to save " + message.getMessageId() + " in " + mailboxId)));
+
+ Mono<ModSeq> nextModSeqMono = modSeqProvider.nextModSeq(mailboxId)
+ .switchIfEmpty(Mono.error(() -> new MailboxException("Can not find a MODSEQ to save " + message.getMessageId() + " in " + mailboxId)));
+
+ try {
+ Mono.zip(messageUidMono, nextModSeqMono)
+ .doOnNext(tuple -> {
+ message.setUid(tuple.getT1());
+ message.setModSeq(tuple.getT2());
+ })
+ .block();
+ } catch (RuntimeException e) {
+ if (e.getCause() instanceof MailboxException) {
+ throw (MailboxException)e.getCause();
+ }
+ throw e;
+ }
return message;
}
diff --git a/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/CassandraBlobStore.java b/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/CassandraBlobStore.java
index cd179f4..1993f84 100644
--- a/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/CassandraBlobStore.java
+++ b/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/CassandraBlobStore.java
@@ -128,13 +128,12 @@ public class CassandraBlobStore implements BlobStore {
}
private Flux<ByteBuffer> readBlobParts(BucketName bucketName, BlobId blobId) {
- Integer rowCount = selectRowCount(bucketName, blobId)
+ return selectRowCount(bucketName, blobId)
.publishOn(Schedulers.elastic())
.single()
.onErrorResume(NoSuchElementException.class, e -> Mono.error(
new ObjectNotFoundException(String.format("Could not retrieve blob metadata for %s", blobId))))
- .block();
- return Flux.range(0, rowCount)
+ .flatMapMany(rowCount -> Flux.range(0, rowCount))
.publishOn(Schedulers.elastic(), PREFETCH)
.flatMapSequential(partIndex -> readPart(bucketName, blobId, partIndex)
.single()
diff --git a/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/vacation/MemoryVacationRepository.java b/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/vacation/MemoryVacationRepository.java
index 5e3b085..5f06c24 100644
--- a/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/vacation/MemoryVacationRepository.java
+++ b/server/data/data-jmap/src/main/java/org/apache/james/jmap/memory/vacation/MemoryVacationRepository.java
@@ -48,9 +48,9 @@ public class MemoryVacationRepository implements VacationRepository {
public Mono<Void> modifyVacation(AccountId accountId, VacationPatch vacationPatch) {
Preconditions.checkNotNull(accountId);
Preconditions.checkNotNull(vacationPatch);
- Vacation oldVacation = retrieveVacation(accountId).block();
- vacationMap.put(accountId, vacationPatch.patch(oldVacation));
- return Mono.empty();
+ return retrieveVacation(accountId)
+ .doOnNext(oldVacation -> vacationMap.put(accountId, vacationPatch.patch(oldVacation)))
+ .then();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/10: JAMES-2921 BlobStore API should include a
StorageStrategy
Posted by bt...@apache.org.
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 bea796d50e3311030ef57ff655be8723b769db8f
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jan 3 17:52:26 2020 +0700
JAMES-2921 BlobStore API should include a StorageStrategy
---
.../cassandra/mail/CassandraAttachmentMapper.java | 6 +-
.../cassandra/mail/CassandraMessageDAO.java | 6 +-
.../mail/migration/AttachmentV2Migration.java | 4 +-
.../mail/CassandraAttachmentFallbackTest.java | 7 +-
.../mail/migration/AttachmentV2MigrationTest.java | 16 +++--
.../vault/blob/BlobStoreDeletedMessageVault.java | 5 +-
.../java/org/apache/james/blob/api/BlobStore.java | 14 ++--
.../apache/james/blob/api/MetricableBlobStore.java | 8 +--
.../main/java/org/apache/james/blob/api/Store.java | 6 +-
.../apache/james/blob/api/BlobStoreContract.java | 31 +++++----
.../james/blob/api/BucketBlobStoreContract.java | 27 ++++----
.../james/blob/api/DeleteBlobStoreContract.java | 23 ++++---
.../blob/api/MetricableBlobStoreContract.java | 25 +++----
.../james/blob/cassandra/CassandraBlobStore.java | 4 +-
.../blob/cassandra/CassandraBlobStoreTest.java | 9 +--
.../file/LocalFileBlobExportMechanismTest.java | 11 +--
.../apache/james/blob/memory/MemoryBlobStore.java | 6 +-
.../blob/objectstorage/ObjectStorageBlobStore.java | 10 +--
.../ObjectStorageBlobStoreContract.java | 3 +-
.../objectstorage/ObjectStorageBlobStoreTest.java | 15 ++--
.../apache/james/blob/union/UnionBlobStore.java | 33 +++++----
.../james/blob/union/UnionBlobStoreTest.java | 79 +++++++++++-----------
.../apache/james/blob/mail/MimeMessageStore.java | 6 +-
.../swift/ObjectStorageBlobStoreModuleTest.java | 3 +-
.../james/webadmin/vault/routes/ExportService.java | 4 +-
.../linshare/LinshareBlobExportMechanismTest.java | 9 +--
26 files changed, 205 insertions(+), 165 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
index fddf40f..867a002 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentMapper.java
@@ -19,6 +19,8 @@
package org.apache.james.mailbox.cassandra.mail;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+
import java.util.Collection;
import java.util.List;
@@ -110,7 +112,7 @@ public class CassandraAttachmentMapper implements AttachmentMapper {
@Override
public void storeAttachmentForOwner(Attachment attachment, Username owner) throws MailboxException {
ownerDAO.addOwner(attachment.getAttachmentId(), owner)
- .then(blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes()))
+ .then(blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST))
.map(blobId -> CassandraAttachmentDAOV2.from(attachment, blobId))
.flatMap(attachmentDAOV2::storeAttachment)
.block();
@@ -137,7 +139,7 @@ public class CassandraAttachmentMapper implements AttachmentMapper {
}
public Mono<Void> storeAttachmentAsync(Attachment attachment, MessageId ownerMessageId) {
- return blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes())
+ return blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST)
.map(blobId -> CassandraAttachmentDAOV2.from(attachment, blobId))
.flatMap(daoAttachment -> storeAttachmentWithIndex(daoAttachment, ownerMessageId));
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java
index ac232a4..16f81fd 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAO.java
@@ -23,6 +23,8 @@ import static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageIds.MESSAGE_ID;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageV2Table.ATTACHMENTS;
import static org.apache.james.mailbox.cassandra.table.CassandraMessageV2Table.BODY;
@@ -182,8 +184,8 @@ public class CassandraMessageDAO {
byte[] headerContent = IOUtils.toByteArray(message.getHeaderContent());
byte[] bodyContent = IOUtils.toByteArray(message.getBodyContent());
- Mono<BlobId> bodyFuture = blobStore.save(blobStore.getDefaultBucketName(), bodyContent);
- Mono<BlobId> headerFuture = blobStore.save(blobStore.getDefaultBucketName(), headerContent);
+ Mono<BlobId> bodyFuture = blobStore.save(blobStore.getDefaultBucketName(), bodyContent, LOW_COST);
+ Mono<BlobId> headerFuture = blobStore.save(blobStore.getDefaultBucketName(), headerContent, SIZE_BASED);
return headerFuture.zipWith(bodyFuture);
} catch (IOException e) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2Migration.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2Migration.java
index 333224a..25474f5 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2Migration.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2Migration.java
@@ -19,6 +19,8 @@
package org.apache.james.mailbox.cassandra.mail.migration;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+
import javax.inject.Inject;
import org.apache.james.backends.cassandra.migration.Migration;
@@ -55,7 +57,7 @@ public class AttachmentV2Migration implements Migration {
}
private Mono<Void> migrateAttachment(Attachment attachment) {
- return blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes())
+ return blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST)
.map(blobId -> CassandraAttachmentDAOV2.from(attachment, blobId))
.flatMap(attachmentDAOV2::storeAttachment)
.then(attachmentDAOV1.deleteAttachment(attachment.getAttachmentId()));
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentFallbackTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentFallbackTest.java
index 5c45b28..4cf6df1 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentFallbackTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentFallbackTest.java
@@ -19,6 +19,7 @@
package org.apache.james.mailbox.cassandra.mail;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -98,7 +99,7 @@ class CassandraAttachmentFallbackTest {
.bytes("{\"property\":`\"different\"}".getBytes(StandardCharsets.UTF_8))
.build();
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes()).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST).block();
attachmentDAOV2.storeAttachment(CassandraAttachmentDAOV2.from(attachment, blobId)).block();
attachmentDAO.storeAttachment(otherAttachment).block();
@@ -133,7 +134,7 @@ class CassandraAttachmentFallbackTest {
.bytes("{\"property\":`\"different\"}".getBytes(StandardCharsets.UTF_8))
.build();
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes()).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST).block();
attachmentDAOV2.storeAttachment(CassandraAttachmentDAOV2.from(attachment, blobId)).block();
attachmentDAO.storeAttachment(otherAttachment).block();
@@ -168,7 +169,7 @@ class CassandraAttachmentFallbackTest {
.bytes("{\"property\":`\"different\"}".getBytes(StandardCharsets.UTF_8))
.build();
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes()).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), attachment.getBytes(), LOW_COST).block();
attachmentDAOV2.storeAttachment(CassandraAttachmentDAOV2.from(attachment, blobId)).block();
attachmentDAO.storeAttachment(otherAttachment).block();
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2MigrationTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2MigrationTest.java
index 405ad9c..1ddaf1b 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2MigrationTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/migration/AttachmentV2MigrationTest.java
@@ -19,8 +19,10 @@
package org.apache.james.mailbox.cassandra.mail.migration;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -152,7 +154,7 @@ class AttachmentV2MigrationTest {
when(attachmentDAO.retrieveAll()).thenReturn(Flux.just(
attachment1,
attachment2));
- when(blobsStore.save(any(BucketName.class), any(byte[].class))).thenThrow(new RuntimeException());
+ when(blobsStore.save(any(BucketName.class), any(byte[].class), eq(LOW_COST))).thenThrow(new RuntimeException());
assertThat(migration.asTask().run()).isEqualTo(Task.Result.PARTIAL);
}
@@ -167,9 +169,9 @@ class AttachmentV2MigrationTest {
when(attachmentDAO.retrieveAll()).thenReturn(Flux.just(
attachment1,
attachment2));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes()))
+ when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes(), LOW_COST))
.thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment1.getBytes())));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes()))
+ when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes(), LOW_COST))
.thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment2.getBytes())));
when(attachmentDAOV2.storeAttachment(any())).thenThrow(new RuntimeException());
@@ -186,9 +188,9 @@ class AttachmentV2MigrationTest {
when(attachmentDAO.retrieveAll()).thenReturn(Flux.just(
attachment1,
attachment2));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes()))
+ when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes(), LOW_COST))
.thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment1.getBytes())));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes()))
+ when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes(), LOW_COST))
.thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment2.getBytes())));
when(attachmentDAOV2.storeAttachment(any())).thenReturn(Mono.empty());
when(attachmentDAO.deleteAttachment(any())).thenThrow(new RuntimeException());
@@ -206,9 +208,9 @@ class AttachmentV2MigrationTest {
when(attachmentDAO.retrieveAll()).thenReturn(Flux.just(
attachment1,
attachment2));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes()))
+ when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment1.getBytes(), LOW_COST))
.thenReturn(Mono.just(BLOB_ID_FACTORY.forPayload(attachment1.getBytes())));
- when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes()))
+ when(blobsStore.save(blobsStore.getDefaultBucketName(), attachment2.getBytes(), LOW_COST))
.thenThrow(new RuntimeException());
when(attachmentDAOV2.storeAttachment(any())).thenReturn(Mono.empty());
when(attachmentDAO.deleteAttachment(any())).thenReturn(Mono.empty());
diff --git a/mailbox/plugin/deleted-messages-vault/src/main/java/org/apache/james/vault/blob/BlobStoreDeletedMessageVault.java b/mailbox/plugin/deleted-messages-vault/src/main/java/org/apache/james/vault/blob/BlobStoreDeletedMessageVault.java
index 7cc3905..e545e53 100644
--- a/mailbox/plugin/deleted-messages-vault/src/main/java/org/apache/james/vault/blob/BlobStoreDeletedMessageVault.java
+++ b/mailbox/plugin/deleted-messages-vault/src/main/java/org/apache/james/vault/blob/BlobStoreDeletedMessageVault.java
@@ -19,6 +19,8 @@
package org.apache.james.vault.blob;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+
import java.io.InputStream;
import java.time.Clock;
import java.time.ZonedDateTime;
@@ -47,6 +49,7 @@ import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
@@ -95,7 +98,7 @@ public class BlobStoreDeletedMessageVault implements DeletedMessageVault {
}
private Mono<Void> appendMessage(DeletedMessage deletedMessage, InputStream mimeMessage, BucketName bucketName) {
- return blobStore.save(bucketName, mimeMessage)
+ return blobStore.save(bucketName, mimeMessage, LOW_COST)
.map(blobId -> StorageInformation.builder()
.bucketName(bucketName)
.blobId(blobId))
diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStore.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStore.java
index 9c4f650..59414d5 100644
--- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStore.java
+++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStore.java
@@ -25,16 +25,22 @@ import reactor.core.publisher.Mono;
public interface BlobStore {
- Mono<BlobId> save(BucketName bucketName, byte[] data);
+ enum StoragePolicy {
+ SIZE_BASED,
+ LOW_COST,
+ HIGH_PERFORMANCE
+ }
+
+ Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy);
- Mono<BlobId> save(BucketName bucketName, InputStream data);
+ Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy);
Mono<byte[]> readBytes(BucketName bucketName, BlobId blobId);
InputStream read(BucketName bucketName, BlobId blobId);
- default Mono<BlobId> save(BucketName bucketName, String data) {
- return save(bucketName, data.getBytes(StandardCharsets.UTF_8));
+ default Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
+ return save(bucketName, data.getBytes(StandardCharsets.UTF_8), storagePolicy);
}
BucketName getDefaultBucketName();
diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/MetricableBlobStore.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/MetricableBlobStore.java
index f2a6d60..a081cec 100644
--- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/MetricableBlobStore.java
+++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/MetricableBlobStore.java
@@ -50,15 +50,15 @@ public class MetricableBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data) {
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
return metricFactory
- .runPublishingTimerMetric(SAVE_BYTES_TIMER_NAME, blobStoreImpl.save(bucketName, data));
+ .runPublishingTimerMetric(SAVE_BYTES_TIMER_NAME, blobStoreImpl.save(bucketName, data, storagePolicy));
}
@Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data) {
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
return metricFactory
- .runPublishingTimerMetric(SAVE_INPUT_STREAM_TIMER_NAME, blobStoreImpl.save(bucketName, data));
+ .runPublishingTimerMetric(SAVE_INPUT_STREAM_TIMER_NAME, blobStoreImpl.save(bucketName, data, storagePolicy));
}
@Override
diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/Store.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/Store.java
index 5feef59..f37605c 100644
--- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/Store.java
+++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/Store.java
@@ -71,14 +71,16 @@ public interface Store<T, I> {
public static class BytesToSave implements ValueToSave {
private final byte[] bytes;
+ private final BlobStore.StoragePolicy storagePolicy;
- public BytesToSave(byte[] bytes) {
+ public BytesToSave(byte[] bytes, BlobStore.StoragePolicy storagePolicy) {
this.bytes = bytes;
+ this.storagePolicy = storagePolicy;
}
@Override
public Mono<BlobId> saveIn(BucketName bucketName, BlobStore blobStore) {
- return blobStore.save(bucketName, bytes);
+ return blobStore.save(bucketName, bytes, storagePolicy);
}
}
diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
index 004c44f..df822b1 100644
--- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
+++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
@@ -19,6 +19,7 @@
package org.apache.james.blob.api;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -47,7 +48,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- assertThatThrownBy(() -> store.save(defaultBucketName, (byte[]) null).block())
+ assertThatThrownBy(() -> store.save(defaultBucketName, (byte[]) null, LOW_COST).block())
.isInstanceOf(NullPointerException.class);
}
@@ -56,7 +57,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- assertThatThrownBy(() -> store.save(defaultBucketName, (String) null).block())
+ assertThatThrownBy(() -> store.save(defaultBucketName, (String) null, LOW_COST).block())
.isInstanceOf(NullPointerException.class);
}
@@ -65,7 +66,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- assertThatThrownBy(() -> store.save(defaultBucketName, (InputStream) null).block())
+ assertThatThrownBy(() -> store.save(defaultBucketName, (InputStream) null, LOW_COST).block())
.isInstanceOf(NullPointerException.class);
}
@@ -74,7 +75,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, EMPTY_BYTEARRAY).block();
+ BlobId blobId = store.save(defaultBucketName, EMPTY_BYTEARRAY, LOW_COST).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
@@ -86,7 +87,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, new String()).block();
+ BlobId blobId = store.save(defaultBucketName, new String(), LOW_COST).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
@@ -98,7 +99,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, new ByteArrayInputStream(EMPTY_BYTEARRAY)).block();
+ BlobId blobId = store.save(defaultBucketName, new ByteArrayInputStream(EMPTY_BYTEARRAY), LOW_COST).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
@@ -110,7 +111,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
assertThat(blobId).isEqualTo(blobIdFactory().from("31f7a65e315586ac198bd798b6629ce4903d0899476d5741a9f32e2e521b6a66"));
}
@@ -120,7 +121,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_STRING).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_STRING, LOW_COST).block();
assertThat(blobId).isEqualTo(blobIdFactory().from("31f7a65e315586ac198bd798b6629ce4903d0899476d5741a9f32e2e521b6a66"));
}
@@ -130,7 +131,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, new ByteArrayInputStream(SHORT_BYTEARRAY)).block();
+ BlobId blobId = store.save(defaultBucketName, new ByteArrayInputStream(SHORT_BYTEARRAY), LOW_COST).block();
assertThat(blobId).isEqualTo(blobIdFactory().from("31f7a65e315586ac198bd798b6629ce4903d0899476d5741a9f32e2e521b6a66"));
}
@@ -149,7 +150,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
@@ -161,7 +162,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, ELEVEN_KILOBYTES).block();
+ BlobId blobId = store.save(defaultBucketName, ELEVEN_KILOBYTES, LOW_COST).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
@@ -173,7 +174,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES).block();
+ BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, LOW_COST).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
@@ -194,7 +195,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
InputStream read = store.read(defaultBucketName, blobId);
@@ -206,7 +207,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, ELEVEN_KILOBYTES).block();
+ BlobId blobId = store.save(defaultBucketName, ELEVEN_KILOBYTES, LOW_COST).block();
InputStream read = store.read(defaultBucketName, blobId);
@@ -219,7 +220,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BucketName defaultBucketName = store.getDefaultBucketName();
// 12 MB of text
- BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES).block();
+ BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, LOW_COST).block();
InputStream read = store.read(defaultBucketName, blobId);
diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BucketBlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BucketBlobStoreContract.java
index 5a62c14..a52240e 100644
--- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BucketBlobStoreContract.java
+++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BucketBlobStoreContract.java
@@ -19,6 +19,7 @@
package org.apache.james.blob.api;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -51,7 +52,7 @@ public interface BucketBlobStoreContract {
default void deleteBucketShouldDeleteExistingBucketWithItsData() {
BlobStore store = testee();
- BlobId blobId = store.save(CUSTOM, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(CUSTOM, SHORT_BYTEARRAY, LOW_COST).block();
store.deleteBucket(CUSTOM).block();
assertThatThrownBy(() -> store.read(CUSTOM, blobId))
@@ -62,7 +63,7 @@ public interface BucketBlobStoreContract {
default void deleteBucketShouldBeIdempotent() {
BlobStore store = testee();
- store.save(CUSTOM, SHORT_BYTEARRAY).block();
+ store.save(CUSTOM, SHORT_BYTEARRAY, LOW_COST).block();
store.deleteBucket(CUSTOM).block();
assertThatCode(() -> store.deleteBucket(CUSTOM).block())
@@ -73,7 +74,7 @@ public interface BucketBlobStoreContract {
default void saveBytesShouldThrowWhenNullBucketName() {
BlobStore store = testee();
- assertThatThrownBy(() -> store.save(null, SHORT_BYTEARRAY).block())
+ assertThatThrownBy(() -> store.save(null, SHORT_BYTEARRAY, LOW_COST).block())
.isInstanceOf(NullPointerException.class);
}
@@ -81,7 +82,7 @@ public interface BucketBlobStoreContract {
default void saveStringShouldThrowWhenNullBucketName() {
BlobStore store = testee();
- assertThatThrownBy(() -> store.save(null, SHORT_STRING).block())
+ assertThatThrownBy(() -> store.save(null, SHORT_STRING, LOW_COST).block())
.isInstanceOf(NullPointerException.class);
}
@@ -89,7 +90,7 @@ public interface BucketBlobStoreContract {
default void saveInputStreamShouldThrowWhenNullBucketName() {
BlobStore store = testee();
- assertThatThrownBy(() -> store.save(null, new ByteArrayInputStream(SHORT_BYTEARRAY)).block())
+ assertThatThrownBy(() -> store.save(null, new ByteArrayInputStream(SHORT_BYTEARRAY), LOW_COST).block())
.isInstanceOf(NullPointerException.class);
}
@@ -97,7 +98,7 @@ public interface BucketBlobStoreContract {
default void readShouldThrowWhenNullBucketName() {
BlobStore store = testee();
- BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY, LOW_COST).block();
assertThatThrownBy(() -> store.read(null, blobId))
.isInstanceOf(NullPointerException.class);
}
@@ -106,7 +107,7 @@ public interface BucketBlobStoreContract {
default void readBytesStreamShouldThrowWhenNullBucketName() {
BlobStore store = testee();
- BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY, LOW_COST).block();
assertThatThrownBy(() -> store.readBytes(null, blobId).block())
.isInstanceOf(NullPointerException.class);
}
@@ -115,7 +116,7 @@ public interface BucketBlobStoreContract {
default void readStringShouldThrowWhenBucketDoesNotExist() {
BlobStore store = testee();
- BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY, LOW_COST).block();
assertThatThrownBy(() -> store.read(CUSTOM, blobId))
.isInstanceOf(ObjectStoreException.class);
}
@@ -124,7 +125,7 @@ public interface BucketBlobStoreContract {
default void readBytesStreamShouldThrowWhenBucketDoesNotExist() {
BlobStore store = testee();
- BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY, LOW_COST).block();
assertThatThrownBy(() -> store.readBytes(CUSTOM, blobId).block())
.isInstanceOf(ObjectStoreException.class);
}
@@ -133,8 +134,8 @@ public interface BucketBlobStoreContract {
default void shouldBeAbleToSaveDataInMultipleBuckets() {
BlobStore store = testee();
- BlobId blobIdDefault = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY).block();
- BlobId blobIdCustom = store.save(CUSTOM, SHORT_BYTEARRAY).block();
+ BlobId blobIdDefault = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY, LOW_COST).block();
+ BlobId blobIdCustom = store.save(CUSTOM, SHORT_BYTEARRAY, LOW_COST).block();
byte[] bytesDefault = store.readBytes(BucketName.DEFAULT, blobIdDefault).block();
byte[] bytesCustom = store.readBytes(CUSTOM, blobIdCustom).block();
@@ -147,7 +148,7 @@ public interface BucketBlobStoreContract {
BlobStore store = testee();
ConcurrentTestRunner.builder()
- .operation(((threadNumber, step) -> store.save(CUSTOM, SHORT_STRING + threadNumber + step).block()))
+ .operation(((threadNumber, step) -> store.save(CUSTOM, SHORT_STRING + threadNumber + step, LOW_COST).block()))
.threadCount(10)
.operationCount(10)
.runSuccessfullyWithin(Duration.ofMinutes(1));
@@ -157,7 +158,7 @@ public interface BucketBlobStoreContract {
default void deleteBucketConcurrentlyShouldNotFail() throws Exception {
BlobStore store = testee();
- store.save(CUSTOM, SHORT_BYTEARRAY).block();
+ store.save(CUSTOM, SHORT_BYTEARRAY, LOW_COST).block();
ConcurrentTestRunner.builder()
.operation(((threadNumber, step) -> store.deleteBucket(CUSTOM).block()))
diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java
index 35f7df9..616f355 100644
--- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java
+++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java
@@ -19,6 +19,7 @@
package org.apache.james.blob.api;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -61,7 +62,7 @@ public interface DeleteBlobStoreContract {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
store.delete(defaultBucketName, blobId).block();
assertThatThrownBy(() -> store.read(defaultBucketName, blobId))
@@ -73,7 +74,7 @@ public interface DeleteBlobStoreContract {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
store.delete(defaultBucketName, blobId).block();
assertThatCode(() -> store.delete(defaultBucketName, blobId).block())
@@ -85,8 +86,8 @@ public interface DeleteBlobStoreContract {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobIdToDelete = store.save(defaultBucketName, SHORT_BYTEARRAY).block();
- BlobId otherBlobId = store.save(defaultBucketName, ELEVEN_KILOBYTES).block();
+ BlobId blobIdToDelete = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
+ BlobId otherBlobId = store.save(defaultBucketName, ELEVEN_KILOBYTES, LOW_COST).block();
store.delete(defaultBucketName, blobIdToDelete).block();
@@ -100,7 +101,7 @@ public interface DeleteBlobStoreContract {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES).block();
+ BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, LOW_COST).block();
ConcurrentTestRunner.builder()
.operation(((threadNumber, step) -> store.delete(defaultBucketName, blobId).block()))
@@ -121,8 +122,8 @@ public interface DeleteBlobStoreContract {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId customBlobId = store.save(CUSTOM, "custom_string").block();
- BlobId defaultBlobId = store.save(defaultBucketName, SHORT_BYTEARRAY).block();
+ BlobId customBlobId = store.save(CUSTOM, "custom_string", LOW_COST).block();
+ BlobId defaultBlobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
store.delete(CUSTOM, customBlobId).block();
@@ -136,8 +137,8 @@ public interface DeleteBlobStoreContract {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- store.save(CUSTOM, SHORT_BYTEARRAY).block();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY).block();
+ store.save(CUSTOM, SHORT_BYTEARRAY, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
store.delete(defaultBucketName, blobId).block();
@@ -151,7 +152,7 @@ public interface DeleteBlobStoreContract {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES).block();
+ BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, LOW_COST).block();
ConcurrentTestRunner.builder()
.operation(((threadNumber, step) -> {
@@ -178,7 +179,7 @@ public interface DeleteBlobStoreContract {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES).block();
+ BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, LOW_COST).block();
ConcurrentTestRunner.builder()
.operation(((threadNumber, step) -> {
diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/MetricableBlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/MetricableBlobStoreContract.java
index 6468328..a6aed4a 100644
--- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/MetricableBlobStoreContract.java
+++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/MetricableBlobStoreContract.java
@@ -19,6 +19,7 @@
package org.apache.james.blob.api;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.apache.james.blob.api.MetricableBlobStore.DELETE_BUCKET_TIMER_NAME;
import static org.apache.james.blob.api.MetricableBlobStore.DELETE_TIMER_NAME;
import static org.apache.james.blob.api.MetricableBlobStore.READ_BYTES_TIMER_NAME;
@@ -61,8 +62,8 @@ public interface MetricableBlobStoreContract extends BlobStoreContract {
default void saveBytesShouldPublishSaveBytesTimerMetrics() {
BlobStore store = testee();
- store.save(store.getDefaultBucketName(), BYTES_CONTENT).block();
- store.save(store.getDefaultBucketName(), BYTES_CONTENT).block();
+ store.save(store.getDefaultBucketName(), BYTES_CONTENT, LOW_COST).block();
+ store.save(store.getDefaultBucketName(), BYTES_CONTENT, LOW_COST).block();
assertThat(metricsTestExtension.getMetricFactory().executionTimesFor(SAVE_BYTES_TIMER_NAME))
.hasSize(2);
@@ -72,8 +73,8 @@ public interface MetricableBlobStoreContract extends BlobStoreContract {
default void saveStringShouldPublishSaveBytesTimerMetrics() {
BlobStore store = testee();
- store.save(store.getDefaultBucketName(), STRING_CONTENT).block();
- store.save(store.getDefaultBucketName(), STRING_CONTENT).block();
+ store.save(store.getDefaultBucketName(), STRING_CONTENT, LOW_COST).block();
+ store.save(store.getDefaultBucketName(), STRING_CONTENT, LOW_COST).block();
assertThat(metricsTestExtension.getMetricFactory().executionTimesFor(SAVE_BYTES_TIMER_NAME))
.hasSize(2);
@@ -83,8 +84,8 @@ public interface MetricableBlobStoreContract extends BlobStoreContract {
default void saveInputStreamShouldPublishSaveInputStreamTimerMetrics() {
BlobStore store = testee();
- store.save(store.getDefaultBucketName(), new ByteArrayInputStream(BYTES_CONTENT)).block();
- store.save(store.getDefaultBucketName(), new ByteArrayInputStream(BYTES_CONTENT)).block();
+ store.save(store.getDefaultBucketName(), new ByteArrayInputStream(BYTES_CONTENT), LOW_COST).block();
+ store.save(store.getDefaultBucketName(), new ByteArrayInputStream(BYTES_CONTENT), LOW_COST).block();
assertThat(metricsTestExtension.getMetricFactory().executionTimesFor(SAVE_INPUT_STREAM_TIMER_NAME))
.hasSize(2);
@@ -94,7 +95,7 @@ public interface MetricableBlobStoreContract extends BlobStoreContract {
default void readBytesShouldPublishReadBytesTimerMetrics() {
BlobStore store = testee();
- BlobId blobId = store.save(store.getDefaultBucketName(), BYTES_CONTENT).block();
+ BlobId blobId = store.save(store.getDefaultBucketName(), BYTES_CONTENT, LOW_COST).block();
store.readBytes(store.getDefaultBucketName(), blobId).block();
store.readBytes(store.getDefaultBucketName(), blobId).block();
@@ -106,7 +107,7 @@ public interface MetricableBlobStoreContract extends BlobStoreContract {
default void readShouldPublishReadTimerMetrics() {
BlobStore store = testee();
- BlobId blobId = store.save(store.getDefaultBucketName(), BYTES_CONTENT).block();
+ BlobId blobId = store.save(store.getDefaultBucketName(), BYTES_CONTENT, LOW_COST).block();
store.read(store.getDefaultBucketName(), blobId);
store.read(store.getDefaultBucketName(), blobId);
@@ -119,8 +120,8 @@ public interface MetricableBlobStoreContract extends BlobStoreContract {
BlobStore store = testee();
BucketName bucketName = BucketName.of("custom");
- store.save(BucketName.DEFAULT, BYTES_CONTENT).block();
- store.save(bucketName, BYTES_CONTENT).block();
+ store.save(BucketName.DEFAULT, BYTES_CONTENT, LOW_COST).block();
+ store.save(bucketName, BYTES_CONTENT, LOW_COST).block();
store.deleteBucket(bucketName).block();
@@ -132,8 +133,8 @@ public interface MetricableBlobStoreContract extends BlobStoreContract {
default void deleteShouldPublishDeleteTimerMetrics() {
BlobStore store = testee();
- BlobId blobId1 = store.save(store.getDefaultBucketName(), BYTES_CONTENT).block();
- BlobId blobId2 = store.save(store.getDefaultBucketName(), BYTES_CONTENT).block();
+ BlobId blobId1 = store.save(store.getDefaultBucketName(), BYTES_CONTENT, LOW_COST).block();
+ BlobId blobId2 = store.save(store.getDefaultBucketName(), BYTES_CONTENT, LOW_COST).block();
store.delete(BucketName.DEFAULT, blobId1).block();
store.delete(BucketName.DEFAULT, blobId2).block();
diff --git a/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/CassandraBlobStore.java b/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/CassandraBlobStore.java
index 7f7efe3..cd179f4 100644
--- a/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/CassandraBlobStore.java
+++ b/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/CassandraBlobStore.java
@@ -76,7 +76,7 @@ public class CassandraBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data) {
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
Preconditions.checkNotNull(data);
return saveAsMono(bucketName, data);
@@ -144,7 +144,7 @@ public class CassandraBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data) {
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
Preconditions.checkNotNull(data);
return Mono.fromCallable(() -> IOUtils.toByteArray(data))
.flatMap(bytes -> saveAsMono(bucketName, bytes));
diff --git a/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/CassandraBlobStoreTest.java b/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/CassandraBlobStoreTest.java
index 86b0c69..63c774e 100644
--- a/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/CassandraBlobStoreTest.java
+++ b/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/CassandraBlobStoreTest.java
@@ -19,6 +19,7 @@
package org.apache.james.blob.cassandra;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.spy;
@@ -87,7 +88,7 @@ public class CassandraBlobStoreTest implements MetricableBlobStoreContract {
@Test
void readBytesShouldReturnSplitSavedDataByChunk() {
String longString = Strings.repeat("0123456789\n", MULTIPLE_CHUNK_SIZE);
- BlobId blobId = testee.save(testee.getDefaultBucketName(), longString).block();
+ BlobId blobId = testee.save(testee.getDefaultBucketName(), longString, LOW_COST).block();
byte[] bytes = testee.readBytes(testee.getDefaultBucketName(), blobId).block();
@@ -98,7 +99,7 @@ public class CassandraBlobStoreTest implements MetricableBlobStoreContract {
void readBytesShouldNotReturnInvalidResultsWhenPartialDataPresent() {
int repeatCount = MULTIPLE_CHUNK_SIZE * CHUNK_SIZE;
String longString = Strings.repeat("0123456789\n", repeatCount);
- BlobId blobId = testee.save(testee.getDefaultBucketName(), longString).block();
+ BlobId blobId = testee.save(testee.getDefaultBucketName(), longString, LOW_COST).block();
when(defaultBucketDAO.readPart(blobId, 1)).thenReturn(Mono.empty());
@@ -111,7 +112,7 @@ public class CassandraBlobStoreTest implements MetricableBlobStoreContract {
void readShouldNotReturnInvalidResultsWhenPartialDataPresent() {
int repeatCount = MULTIPLE_CHUNK_SIZE * CHUNK_SIZE;
String longString = Strings.repeat("0123456789\n", repeatCount);
- BlobId blobId = testee.save(testee.getDefaultBucketName(), longString).block();
+ BlobId blobId = testee.save(testee.getDefaultBucketName(), longString, LOW_COST).block();
when(defaultBucketDAO.readPart(blobId, 1)).thenReturn(Mono.empty());
@@ -131,7 +132,7 @@ public class CassandraBlobStoreTest implements MetricableBlobStoreContract {
void blobStoreShouldSupport100MBBlob() throws IOException {
ZeroedInputStream data = new ZeroedInputStream(100_000_000);
HashingInputStream writeHash = new HashingInputStream(Hashing.sha256(), data);
- BlobId blobId = testee.save(testee.getDefaultBucketName(), writeHash).block();
+ BlobId blobId = testee.save(testee.getDefaultBucketName(), writeHash, LOW_COST).block();
InputStream bytes = testee.read(testee.getDefaultBucketName(), blobId);
HashingInputStream readHash = new HashingInputStream(Hashing.sha256(), bytes);
diff --git a/server/blob/blob-export-file/src/test/java/org/apache/james/blob/export/file/LocalFileBlobExportMechanismTest.java b/server/blob/blob-export-file/src/test/java/org/apache/james/blob/export/file/LocalFileBlobExportMechanismTest.java
index ca0cce5..174b2c3 100644
--- a/server/blob/blob-export-file/src/test/java/org/apache/james/blob/export/file/LocalFileBlobExportMechanismTest.java
+++ b/server/blob/blob-export-file/src/test/java/org/apache/james/blob/export/file/LocalFileBlobExportMechanismTest.java
@@ -19,6 +19,7 @@
package org.apache.james.blob.export.file;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
@@ -79,7 +80,7 @@ class LocalFileBlobExportMechanismTest {
@Test
void exportingBlobShouldSendAMail() {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
String explanation = "The content of a deleted message vault had been shared with you.";
testee.blobId(blobId)
@@ -112,7 +113,7 @@ class LocalFileBlobExportMechanismTest {
@Test
void exportingBlobShouldCreateAFileWithTheCorrespondingContent(FileSystem fileSystem) {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
testee.blobId(blobId)
.with(MailAddressFixture.RECIPIENT1)
@@ -150,7 +151,7 @@ class LocalFileBlobExportMechanismTest {
@Test
void exportingBlobShouldCreateAFileWithoutExtensionWhenNotDeclaringExtension() {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
testee.blobId(blobId)
.with(MailAddressFixture.RECIPIENT1)
@@ -174,7 +175,7 @@ class LocalFileBlobExportMechanismTest {
@Test
void exportingBlobShouldCreateAFileWithExtensionWhenDeclaringExtension() {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
testee.blobId(blobId)
.with(MailAddressFixture.RECIPIENT1)
@@ -199,7 +200,7 @@ class LocalFileBlobExportMechanismTest {
@Test
void exportingBlobShouldCreateAFileWithPrefixWhenDeclaringPrefix() {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
String filePrefix = "deleted-message-of-bob@james.org";
testee.blobId(blobId)
diff --git a/server/blob/blob-memory/src/main/java/org/apache/james/blob/memory/MemoryBlobStore.java b/server/blob/blob-memory/src/main/java/org/apache/james/blob/memory/MemoryBlobStore.java
index 6d107ac..bd2615c 100644
--- a/server/blob/blob-memory/src/main/java/org/apache/james/blob/memory/MemoryBlobStore.java
+++ b/server/blob/blob-memory/src/main/java/org/apache/james/blob/memory/MemoryBlobStore.java
@@ -57,7 +57,7 @@ public class MemoryBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data) {
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
Preconditions.checkNotNull(bucketName);
Preconditions.checkNotNull(data);
@@ -72,12 +72,12 @@ public class MemoryBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data) {
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
Preconditions.checkNotNull(bucketName);
Preconditions.checkNotNull(data);
try {
byte[] bytes = IOUtils.toByteArray(data);
- return save(bucketName, bytes);
+ return save(bucketName, bytes, storagePolicy);
} catch (IOException e) {
throw new RuntimeException(e);
}
diff --git a/server/blob/blob-objectstorage/src/main/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStore.java b/server/blob/blob-objectstorage/src/main/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStore.java
index fba4fa1..3c45fe8 100644
--- a/server/blob/blob-objectstorage/src/main/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStore.java
+++ b/server/blob/blob-objectstorage/src/main/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStore.java
@@ -97,7 +97,7 @@ public class ObjectStorageBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data) {
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
Preconditions.checkNotNull(data);
ObjectStorageBucketName resolvedBucketName = bucketNameResolver.resolve(bucketName);
@@ -116,19 +116,19 @@ public class ObjectStorageBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data) {
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
Preconditions.checkNotNull(data);
- return Mono.defer(() -> savingStrategySelection(bucketName, data));
+ return Mono.defer(() -> savingStrategySelection(bucketName, data, storagePolicy));
}
- private Mono<BlobId> savingStrategySelection(BucketName bucketName, InputStream data) {
+ private Mono<BlobId> savingStrategySelection(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
InputStream bufferedData = new BufferedInputStream(data, BUFFERED_SIZE + 1);
try {
if (isItABigStream(bufferedData)) {
return saveBigStream(bucketName, bufferedData);
} else {
- return save(bucketName, IOUtils.toByteArray(bufferedData));
+ return save(bucketName, IOUtils.toByteArray(bufferedData), storagePolicy);
}
} catch (IOException e) {
throw new RuntimeException(e);
diff --git a/server/blob/blob-objectstorage/src/test/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStoreContract.java b/server/blob/blob-objectstorage/src/test/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStoreContract.java
index 2e96f52..a5b1fec 100644
--- a/server/blob/blob-objectstorage/src/test/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStoreContract.java
+++ b/server/blob/blob-objectstorage/src/test/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStoreContract.java
@@ -19,6 +19,7 @@
package org.apache.james.blob.objectstorage;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import java.io.InputStream;
@@ -38,7 +39,7 @@ public interface ObjectStorageBlobStoreContract {
default void assertBlobStoreCanStoreAndRetrieve(ObjectStorageBlobStoreBuilder.ReadyToBuild builder) {
ObjectStorageBlobStore blobStore = builder.build();
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), CONTENT, LOW_COST).block();
InputStream inputStream = blobStore.read(blobStore.getDefaultBucketName(), blobId);
assertThat(inputStream).hasSameContentAs(IOUtils.toInputStream(CONTENT, StandardCharsets.UTF_8));
diff --git a/server/blob/blob-objectstorage/src/test/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStoreTest.java b/server/blob/blob-objectstorage/src/test/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStoreTest.java
index e342525..263e105 100644
--- a/server/blob/blob-objectstorage/src/test/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStoreTest.java
+++ b/server/blob/blob-objectstorage/src/test/java/org/apache/james/blob/objectstorage/ObjectStorageBlobStoreTest.java
@@ -19,6 +19,7 @@
package org.apache.james.blob.objectstorage;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThat;
import java.io.ByteArrayInputStream;
@@ -119,7 +120,7 @@ public class ObjectStorageBlobStoreTest implements MetricableBlobStoreContract {
.namespace(defaultBucketName)
.build();
String content = "James is the best!";
- BlobId blobId = encryptedBlobStore.save(encryptedBlobStore.getDefaultBucketName(), content).block();
+ BlobId blobId = encryptedBlobStore.save(encryptedBlobStore.getDefaultBucketName(), content, LOW_COST).block();
InputStream read = encryptedBlobStore.read(encryptedBlobStore.getDefaultBucketName(), blobId);
String expectedContent = IOUtils.toString(read, Charsets.UTF_8);
@@ -135,7 +136,7 @@ public class ObjectStorageBlobStoreTest implements MetricableBlobStoreContract {
.namespace(defaultBucketName)
.build();
String content = "James is the best!";
- BlobId blobId = encryptedBlobStore.save(encryptedBlobStore.getDefaultBucketName(), content).block();
+ BlobId blobId = encryptedBlobStore.save(encryptedBlobStore.getDefaultBucketName(), content, LOW_COST).block();
InputStream encryptedIs = testee.read(encryptedBlobStore.getDefaultBucketName(), blobId);
assertThat(encryptedIs).isNotNull();
@@ -150,7 +151,7 @@ public class ObjectStorageBlobStoreTest implements MetricableBlobStoreContract {
@Test
void deleteBucketShouldDeleteSwiftContainer() {
BucketName bucketName = BucketName.of("azerty");
- objectStorageBlobStore.save(bucketName, "data").block();
+ objectStorageBlobStore.save(bucketName, "data", LOW_COST).block();
objectStorageBlobStore.deleteBucket(bucketName).block();
@@ -177,7 +178,7 @@ public class ObjectStorageBlobStoreTest implements MetricableBlobStoreContract {
void saveBytesShouldNotCompleteWhenDoesNotAwait() {
// String need to be big enough to get async thread busy hence could not return result instantly
Mono<BlobId> blobIdFuture = testee
- .save(testee.getDefaultBucketName(), BIG_STRING.getBytes(StandardCharsets.UTF_8))
+ .save(testee.getDefaultBucketName(), BIG_STRING.getBytes(StandardCharsets.UTF_8), LOW_COST)
.subscribeOn(Schedulers.elastic());
assertThat(blobIdFuture.toFuture()).isNotCompleted();
}
@@ -185,7 +186,7 @@ public class ObjectStorageBlobStoreTest implements MetricableBlobStoreContract {
@Test
void saveStringShouldNotCompleteWhenDoesNotAwait() {
Mono<BlobId> blobIdFuture = testee
- .save(testee.getDefaultBucketName(), BIG_STRING)
+ .save(testee.getDefaultBucketName(), BIG_STRING, LOW_COST)
.subscribeOn(Schedulers.elastic());
assertThat(blobIdFuture.toFuture()).isNotCompleted();
}
@@ -193,14 +194,14 @@ public class ObjectStorageBlobStoreTest implements MetricableBlobStoreContract {
@Test
void saveInputStreamShouldNotCompleteWhenDoesNotAwait() {
Mono<BlobId> blobIdFuture = testee
- .save(testee.getDefaultBucketName(), new ByteArrayInputStream(BIG_STRING.getBytes(StandardCharsets.UTF_8)))
+ .save(testee.getDefaultBucketName(), new ByteArrayInputStream(BIG_STRING.getBytes(StandardCharsets.UTF_8)), LOW_COST)
.subscribeOn(Schedulers.elastic());
assertThat(blobIdFuture.toFuture()).isNotCompleted();
}
@Test
void readBytesShouldNotCompleteWhenDoesNotAwait() {
- BlobId blobId = testee().save(testee.getDefaultBucketName(), BIG_STRING).block();
+ BlobId blobId = testee().save(testee.getDefaultBucketName(), BIG_STRING, LOW_COST).block();
Mono<byte[]> resultFuture = testee.readBytes(testee.getDefaultBucketName(), blobId).subscribeOn(Schedulers.elastic());
assertThat(resultFuture.toFuture()).isNotCompleted();
}
diff --git a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/UnionBlobStore.java b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/UnionBlobStore.java
index b822b02..37f7c8a 100644
--- a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/UnionBlobStore.java
+++ b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/UnionBlobStore.java
@@ -23,7 +23,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.PushbackInputStream;
import java.util.Optional;
-import java.util.function.BiFunction;
import java.util.function.Function;
import org.apache.james.blob.api.BlobId;
@@ -42,6 +41,11 @@ import reactor.core.publisher.Mono;
public class UnionBlobStore implements BlobStore {
@FunctionalInterface
+ public interface StorageOperation<T> {
+ Mono<BlobId> save(BucketName bucketName, T data, StoragePolicy storagePolicy);
+ }
+
+ @FunctionalInterface
public interface RequireCurrent {
RequireLegacy current(BlobStore blobStore);
}
@@ -83,26 +87,26 @@ public class UnionBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data) {
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
try {
- return saveToCurrentFallbackIfFails(bucketName, data,
+ return saveToCurrentFallbackIfFails(bucketName, data, storagePolicy,
currentBlobStore::save,
legacyBlobStore::save);
} catch (Exception e) {
LOGGER.error("exception directly happens while saving bytes data, fall back to legacy blob store", e);
- return legacyBlobStore.save(bucketName, data);
+ return legacyBlobStore.save(bucketName, data, storagePolicy);
}
}
@Override
- public Mono<BlobId> save(BucketName bucketName, String data) {
+ public Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
try {
- return saveToCurrentFallbackIfFails(bucketName, data,
+ return saveToCurrentFallbackIfFails(bucketName, data, storagePolicy,
currentBlobStore::save,
legacyBlobStore::save);
} catch (Exception e) {
LOGGER.error("exception directly happens while saving String data, fall back to legacy blob store", e);
- return legacyBlobStore.save(bucketName, data);
+ return legacyBlobStore.save(bucketName, data, storagePolicy);
}
}
@@ -118,14 +122,14 @@ public class UnionBlobStore implements BlobStore {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data) {
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
try {
- return saveToCurrentFallbackIfFails(bucketName, data,
+ return saveToCurrentFallbackIfFails(bucketName, data, storagePolicy,
currentBlobStore::save,
legacyBlobStore::save);
} catch (Exception e) {
LOGGER.error("exception directly happens while saving InputStream data, fall back to legacy blob store", e);
- return legacyBlobStore.save(bucketName, data);
+ return legacyBlobStore.save(bucketName, data, storagePolicy);
}
}
@@ -190,12 +194,13 @@ public class UnionBlobStore implements BlobStore {
private <T> Mono<BlobId> saveToCurrentFallbackIfFails(
BucketName bucketName,
T data,
- BiFunction<BucketName, T, Mono<BlobId>> currentSavingOperation,
- BiFunction<BucketName, T, Mono<BlobId>> fallbackSavingOperationSupplier) {
+ StoragePolicy storagePolicy,
+ StorageOperation<T> currentSavingOperation,
+ StorageOperation<T> fallbackSavingOperationSupplier) {
- return Mono.defer(() -> currentSavingOperation.apply(bucketName, data))
+ return Mono.defer(() -> currentSavingOperation.save(bucketName, data, storagePolicy))
.onErrorResume(this::logAndReturnEmpty)
- .switchIfEmpty(Mono.defer(() -> fallbackSavingOperationSupplier.apply(bucketName, data)));
+ .switchIfEmpty(Mono.defer(() -> fallbackSavingOperationSupplier.save(bucketName, data, storagePolicy)));
}
private <T> Mono<T> logAndReturnEmpty(Throwable throwable) {
diff --git a/server/blob/blob-union/src/test/java/org/apache/james/blob/union/UnionBlobStoreTest.java b/server/blob/blob-union/src/test/java/org/apache/james/blob/union/UnionBlobStoreTest.java
index e981edf..e8ccd81 100644
--- a/server/blob/blob-union/src/test/java/org/apache/james/blob/union/UnionBlobStoreTest.java
+++ b/server/blob/blob-union/src/test/java/org/apache/james/blob/union/UnionBlobStoreTest.java
@@ -19,6 +19,8 @@
package org.apache.james.blob.union;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LowCost;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LowCost;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -55,25 +57,24 @@ import reactor.core.publisher.Mono;
class UnionBlobStoreTest implements BlobStoreContract {
private static class FailingBlobStore implements BlobStore {
-
@Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data) {
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
return Mono.error(new RuntimeException("broken everywhere"));
}
@Override
- public Mono<BlobId> save(BucketName bucketName, String data) {
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
return Mono.error(new RuntimeException("broken everywhere"));
}
@Override
- public BucketName getDefaultBucketName() {
- return BucketName.DEFAULT;
+ public Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
+ return Mono.error(new RuntimeException("broken everywhere"));
}
@Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data) {
- return Mono.error(new RuntimeException("broken everywhere"));
+ public BucketName getDefaultBucketName() {
+ return BucketName.DEFAULT;
}
@Override
@@ -106,12 +107,12 @@ class UnionBlobStoreTest implements BlobStoreContract {
private static class ThrowingBlobStore implements BlobStore {
@Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data) {
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
throw new RuntimeException("broken everywhere");
}
@Override
- public Mono<BlobId> save(BucketName bucketName, String data) {
+ public Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
throw new RuntimeException("broken everywhere");
}
@@ -121,7 +122,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
}
@Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data) {
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
throw new RuntimeException("broken everywhere");
}
@@ -190,7 +191,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
.current(new ThrowingBlobStore())
.legacy(legacyBlobStore)
.build();
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
SoftAssertions.assertSoftly(softly -> {
softly.assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
@@ -207,7 +208,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
.current(new ThrowingBlobStore())
.legacy(legacyBlobStore)
.build();
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT)).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost).block();
SoftAssertions.assertSoftly(softly -> {
softly.assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
@@ -228,7 +229,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
.current(new FailingBlobStore())
.legacy(legacyBlobStore)
.build();
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
SoftAssertions.assertSoftly(softly -> {
softly.assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
@@ -245,7 +246,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
.current(new FailingBlobStore())
.legacy(legacyBlobStore)
.build();
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT)).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost).block();
SoftAssertions.assertSoftly(softly -> {
softly.assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
@@ -267,7 +268,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
.current(new ThrowingBlobStore())
.legacy(legacyBlobStore)
.build();
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
.hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
@@ -281,7 +282,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
.current(new ThrowingBlobStore())
.legacy(legacyBlobStore)
.build();
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(unionBlobStore.readBytes(unionBlobStore.getDefaultBucketName(), blobId).block())
.isEqualTo(BLOB_CONTENT);
@@ -299,7 +300,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
.current(new FailingBlobStore())
.legacy(legacyBlobStore)
.build();
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
.hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
@@ -312,7 +313,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
.current(new FailingBlobStore())
.legacy(legacyBlobStore)
.build();
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(unionBlobStore.readBytes(unionBlobStore.getDefaultBucketName(), blobId).block())
.isEqualTo(BLOB_CONTENT);
@@ -326,9 +327,9 @@ class UnionBlobStoreTest implements BlobStoreContract {
Stream<Function<UnionBlobStore, Mono<?>>> blobStoreOperationsReturnFutures() {
return Stream.of(
- blobStore -> blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT),
- blobStore -> blobStore.save(blobStore.getDefaultBucketName(), STRING_CONTENT),
- blobStore -> blobStore.save(blobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT)),
+ blobStore -> blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost),
+ blobStore -> blobStore.save(blobStore.getDefaultBucketName(), STRING_CONTENT, LowCost),
+ blobStore -> blobStore.save(blobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost),
blobStore -> blobStore.readBytes(blobStore.getDefaultBucketName(), BLOB_ID_FACTORY.randomId()));
}
@@ -390,7 +391,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void readShouldReturnFromCurrentWhenAvailable() {
- BlobId blobId = currentBlobStore.save(currentBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = currentBlobStore.save(currentBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
.hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
@@ -398,7 +399,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void readShouldReturnFromLegacyWhenCurrentNotAvailable() {
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
.hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
@@ -406,7 +407,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void readBytesShouldReturnFromCurrentWhenAvailable() {
- BlobId blobId = currentBlobStore.save(currentBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = currentBlobStore.save(currentBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(unionBlobStore.readBytes(currentBlobStore.getDefaultBucketName(), blobId).block())
.isEqualTo(BLOB_CONTENT);
@@ -414,7 +415,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void readBytesShouldReturnFromLegacyWhenCurrentNotAvailable() {
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(unionBlobStore.readBytes(unionBlobStore.getDefaultBucketName(), blobId).block())
.isEqualTo(BLOB_CONTENT);
@@ -422,7 +423,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void saveShouldWriteToCurrent() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThat(currentBlobStore.readBytes(currentBlobStore.getDefaultBucketName(), blobId).block())
.isEqualTo(BLOB_CONTENT);
@@ -430,7 +431,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void saveShouldNotWriteToLegacy() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
assertThatThrownBy(() -> legacyBlobStore.readBytes(legacyBlobStore.getDefaultBucketName(), blobId).block())
.isInstanceOf(ObjectStoreException.class);
@@ -438,7 +439,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void saveStringShouldWriteToCurrent() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), STRING_CONTENT).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), STRING_CONTENT, LowCost).block();
assertThat(currentBlobStore.readBytes(currentBlobStore.getDefaultBucketName(), blobId).block())
.isEqualTo(BLOB_CONTENT);
@@ -446,7 +447,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void saveStringShouldNotWriteToLegacy() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), STRING_CONTENT).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), STRING_CONTENT, LowCost).block();
assertThatThrownBy(() -> legacyBlobStore.readBytes(legacyBlobStore.getDefaultBucketName(), blobId).block())
.isInstanceOf(ObjectStoreException.class);
@@ -454,7 +455,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void saveInputStreamShouldWriteToCurrent() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT)).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost).block();
assertThat(currentBlobStore.readBytes(currentBlobStore.getDefaultBucketName(), blobId).block())
.isEqualTo(BLOB_CONTENT);
@@ -462,7 +463,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void saveInputStreamShouldNotWriteToLegacy() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT)).block();
+ BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost).block();
assertThatThrownBy(() -> legacyBlobStore.readBytes(legacyBlobStore.getDefaultBucketName(), blobId).block())
.isInstanceOf(ObjectStoreException.class);
@@ -512,8 +513,8 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void deleteBucketShouldDeleteBothCurrentAndLegacyBuckets() {
- BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT).block();
- BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT).block();
+ BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
+ BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
unionBlobStore.deleteBucket(BucketName.DEFAULT).block();
@@ -525,7 +526,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void deleteBucketShouldDeleteCurrentBucketEvenWhenLegacyDoesNotExist() {
- BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT).block();
+ BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
unionBlobStore.deleteBucket(BucketName.DEFAULT).block();
@@ -535,7 +536,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void deleteBucketShouldDeleteLegacyBucketEvenWhenCurrentDoesNotExist() {
- BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT).block();
+ BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
unionBlobStore.deleteBucket(BucketName.DEFAULT).block();
@@ -564,8 +565,8 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void deleteShouldDeleteBothCurrentAndLegacyBlob() {
- BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT).block();
- BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT).block();
+ BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
+ BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
unionBlobStore.delete(BucketName.DEFAULT, currentBlobId).block();
@@ -577,7 +578,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void deleteShouldDeleteCurrentBlobEvenWhenLegacyDoesNotExist() {
- BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT).block();
+ BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
unionBlobStore.delete(BucketName.DEFAULT, currentBlobId).block();
@@ -587,7 +588,7 @@ class UnionBlobStoreTest implements BlobStoreContract {
@Test
void deleteShouldDeleteLegacyBlobEvenWhenCurrentDoesNotExist() {
- BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT).block();
+ BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
unionBlobStore.delete(BucketName.DEFAULT, legacyBlobId).block();
diff --git a/server/blob/mail-store/src/main/java/org/apache/james/blob/mail/MimeMessageStore.java b/server/blob/mail-store/src/main/java/org/apache/james/blob/mail/MimeMessageStore.java
index f0c2480..ad11196 100644
--- a/server/blob/mail-store/src/main/java/org/apache/james/blob/mail/MimeMessageStore.java
+++ b/server/blob/mail-store/src/main/java/org/apache/james/blob/mail/MimeMessageStore.java
@@ -20,6 +20,8 @@
package org.apache.james.blob.mail;
import static org.apache.commons.io.output.NullOutputStream.NULL_OUTPUT_STREAM;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED;
import static org.apache.james.blob.mail.MimeMessagePartsId.BODY_BLOB_TYPE;
import static org.apache.james.blob.mail.MimeMessagePartsId.HEADER_BLOB_TYPE;
@@ -76,8 +78,8 @@ public class MimeMessageStore {
byte[] headerBytes = getHeaderBytes(messageAsArray, bodyStartOctet);
byte[] bodyBytes = getBodyBytes(messageAsArray, bodyStartOctet);
return Stream.of(
- Pair.of(HEADER_BLOB_TYPE, new Store.Impl.BytesToSave(headerBytes)),
- Pair.of(BODY_BLOB_TYPE, new Store.Impl.BytesToSave(bodyBytes)));
+ Pair.of(HEADER_BLOB_TYPE, new Store.Impl.BytesToSave(headerBytes, SIZE_BASED)),
+ Pair.of(BODY_BLOB_TYPE, new Store.Impl.BytesToSave(bodyBytes, LOW_COST)));
} catch (MessagingException | IOException e) {
throw new RuntimeException(e);
}
diff --git a/server/container/guice/blob-objectstorage-guice/src/test/java/org/apache/james/modules/objectstorage/swift/ObjectStorageBlobStoreModuleTest.java b/server/container/guice/blob-objectstorage-guice/src/test/java/org/apache/james/modules/objectstorage/swift/ObjectStorageBlobStoreModuleTest.java
index 7911d16..05a386a 100644
--- a/server/container/guice/blob-objectstorage-guice/src/test/java/org/apache/james/modules/objectstorage/swift/ObjectStorageBlobStoreModuleTest.java
+++ b/server/container/guice/blob-objectstorage-guice/src/test/java/org/apache/james/modules/objectstorage/swift/ObjectStorageBlobStoreModuleTest.java
@@ -19,6 +19,7 @@
package org.apache.james.modules.objectstorage.swift;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.assertj.core.api.Assertions.assertThatCode;
import java.util.Optional;
@@ -137,7 +138,7 @@ class ObjectStorageBlobStoreModuleTest {
BlobStore blobStore = injector.getInstance(Key.get(BlobStore.class, Names.named(MetricableBlobStore.BLOB_STORE_IMPLEMENTATION)));
- assertThatCode(() -> blobStore.save(blobStore.getDefaultBucketName(), new byte[] {0x00})).doesNotThrowAnyException();
+ assertThatCode(() -> blobStore.save(blobStore.getDefaultBucketName(), new byte[] {0x00}, LOW_COST)).doesNotThrowAnyException();
}
}
diff --git a/server/protocols/webadmin/webadmin-mailbox-deleted-message-vault/src/main/java/org/apache/james/webadmin/vault/routes/ExportService.java b/server/protocols/webadmin/webadmin-mailbox-deleted-message-vault/src/main/java/org/apache/james/webadmin/vault/routes/ExportService.java
index 1c03fdf..34d55f1 100644
--- a/server/protocols/webadmin/webadmin-mailbox-deleted-message-vault/src/main/java/org/apache/james/webadmin/vault/routes/ExportService.java
+++ b/server/protocols/webadmin/webadmin-mailbox-deleted-message-vault/src/main/java/org/apache/james/webadmin/vault/routes/ExportService.java
@@ -19,6 +19,8 @@
package org.apache.james.webadmin.vault.routes;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+
import java.io.IOException;
import java.util.Optional;
import java.util.function.Predicate;
@@ -85,7 +87,7 @@ class ExportService {
try (FileBackedOutputStream fileOutputStream = new FileBackedOutputStream(FileUtils.ONE_MB_BI.intValue())) {
zipper.zip(contentLoader(username), messages.toStream(), fileOutputStream);
ByteSource byteSource = fileOutputStream.asByteSource();
- return blobStore.save(blobStore.getDefaultBucketName(), byteSource.openStream()).block();
+ return blobStore.save(blobStore.getDefaultBucketName(), byteSource.openStream(), LOW_COST).block();
}
}
diff --git a/third-party/linshare/src/test/java/org/apache/james/linshare/LinshareBlobExportMechanismTest.java b/third-party/linshare/src/test/java/org/apache/james/linshare/LinshareBlobExportMechanismTest.java
index 542900a..729bbeb 100644
--- a/third-party/linshare/src/test/java/org/apache/james/linshare/LinshareBlobExportMechanismTest.java
+++ b/third-party/linshare/src/test/java/org/apache/james/linshare/LinshareBlobExportMechanismTest.java
@@ -20,6 +20,7 @@
package org.apache.james.linshare;
import static io.restassured.RestAssured.given;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.apache.james.linshare.LinshareFixture.USER_1;
import static org.apache.james.linshare.LinshareFixture.USER_2;
import static org.assertj.core.api.Assertions.assertThat;
@@ -74,7 +75,7 @@ class LinshareBlobExportMechanismTest {
@Test
void exportShouldShareTheDocumentViaLinshare() throws Exception {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), FILE_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), FILE_CONTENT, LOW_COST).block();
String filePrefix = "deleted-message-of-bob@james.org-";
testee.blobId(blobId)
@@ -93,7 +94,7 @@ class LinshareBlobExportMechanismTest {
@Test
void exportShouldSendAnEmailToSharee() throws Exception {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), FILE_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), FILE_CONTENT, LOW_COST).block();
testee.blobId(blobId)
.with(new MailAddress(USER_2.getUsername()))
@@ -122,7 +123,7 @@ class LinshareBlobExportMechanismTest {
@Test
void exportShouldShareTheDocumentAndAllowDownloadViaLinshare() throws Exception {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), FILE_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), FILE_CONTENT, LOW_COST).block();
testee.blobId(blobId)
.with(new MailAddress(USER_2.getUsername()))
@@ -152,7 +153,7 @@ class LinshareBlobExportMechanismTest {
@Test
void exportWithFilePrefixShouldCreateFileWithCustomPrefix() throws Exception {
- BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), FILE_CONTENT).block();
+ BlobId blobId = blobStore.save(blobStore.getDefaultBucketName(), FILE_CONTENT, LOW_COST).block();
String filePrefix = "deleted-message-of-bob@james.org";
testee.blobId(blobId)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 05/10: JAMES-2921 Documentation changes
Posted by bt...@apache.org.
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 e0508a7084e90065f79744bcfa3ac31f7f06972d
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jan 7 14:11:31 2020 +0700
JAMES-2921 Documentation changes
---
src/site/xdoc/server/config-blobstore.xml | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
diff --git a/src/site/xdoc/server/config-blobstore.xml b/src/site/xdoc/server/config-blobstore.xml
index 95af133..b173956 100644
--- a/src/site/xdoc/server/config-blobstore.xml
+++ b/src/site/xdoc/server/config-blobstore.xml
@@ -46,9 +46,16 @@
<dt><strong>implementation</strong></dt>
<dd>cassandra: use cassandra based BlobStore</dd>
<dd>objectstorage: use Swift/AWS S3 based BlobStore</dd>
- <dd>union: Using both objectstorage as the current BlobStore and cassandra as the legacy BlobStore</dd>
+ <dd>hybrid: Using both objectstorage for unfrequently read or big blobs & cassandra for small, often read blobs</dd>
</dl>
+ <subsection name="Hybrid BlobStore size threshold">
+ <dl>
+ <dt><strong>hybrid.size.threshold</strong></dt>
+ <dd>DEFAULT: 32768 bytes (32KB), must be positive. Size threshold for considering a blob as 'big', causing it to be saved in the low cost blobStore.</dd>
+ </dl>
+ </subsection>
+
<subsection name="ObjectStorage BlobStore Codec Configuration">
<dl>
<dt><strong>objectstorage.payload.codec</strong></dt>
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 06/10: JAMES-2921 BlobStoreContract should test all
storage policies basic behaviour
Posted by bt...@apache.org.
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 700dfec5e5f29304275ddf0803da6f9fa6ae7ba2
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jan 7 14:19:25 2020 +0700
JAMES-2921 BlobStoreContract should test all storage policies basic behaviour
---
server/blob/blob-api/pom.xml | 5 +
.../apache/james/blob/api/BlobStoreContract.java | 118 +++++++++++++--------
server/blob/blob-cassandra/pom.xml | 5 +
server/blob/blob-memory/pom.xml | 5 +
4 files changed, 88 insertions(+), 45 deletions(-)
diff --git a/server/blob/blob-api/pom.xml b/server/blob/blob-api/pom.xml
index f4ebecc..9a8c946 100644
--- a/server/blob/blob-api/pom.xml
+++ b/server/blob/blob-api/pom.xml
@@ -65,6 +65,11 @@
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
index df822b1..2067d70 100644
--- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
+++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
@@ -19,20 +19,33 @@
package org.apache.james.blob.api;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.HIGH_PERFORMANCE;
import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
+import java.util.stream.Stream;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import com.google.common.base.Strings;
public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobStoreContract {
+ static Stream<Arguments> storagePolicies() {
+ return Stream.of(
+ Arguments.arguments(LOW_COST),
+ Arguments.arguments(SIZE_BASED),
+ Arguments.arguments(HIGH_PERFORMANCE));
+ }
+
String SHORT_STRING = "toto";
byte[] EMPTY_BYTEARRAY = {};
byte[] SHORT_BYTEARRAY = SHORT_STRING.getBytes(StandardCharsets.UTF_8);
@@ -43,95 +56,104 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobId.Factory blobIdFactory();
- @Test
- default void saveShouldThrowWhenNullData() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldThrowWhenNullData(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- assertThatThrownBy(() -> store.save(defaultBucketName, (byte[]) null, LOW_COST).block())
+ assertThatThrownBy(() -> store.save(defaultBucketName, (byte[]) null, storagePolicy).block())
.isInstanceOf(NullPointerException.class);
}
- @Test
- default void saveShouldThrowWhenNullString() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldThrowWhenNullString(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- assertThatThrownBy(() -> store.save(defaultBucketName, (String) null, LOW_COST).block())
+ assertThatThrownBy(() -> store.save(defaultBucketName, (String) null, storagePolicy).block())
.isInstanceOf(NullPointerException.class);
}
- @Test
- default void saveShouldThrowWhenNullInputStream() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldThrowWhenNullInputStream(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- assertThatThrownBy(() -> store.save(defaultBucketName, (InputStream) null, LOW_COST).block())
+ assertThatThrownBy(() -> store.save(defaultBucketName, (InputStream) null, storagePolicy).block())
.isInstanceOf(NullPointerException.class);
}
- @Test
- default void saveShouldSaveEmptyData() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldSaveEmptyData(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, EMPTY_BYTEARRAY, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, EMPTY_BYTEARRAY, storagePolicy).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
assertThat(new String(bytes, StandardCharsets.UTF_8)).isEmpty();
}
- @Test
- default void saveShouldSaveEmptyString() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldSaveEmptyString(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, new String(), LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, new String(), storagePolicy).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
assertThat(new String(bytes, StandardCharsets.UTF_8)).isEmpty();
}
- @Test
- default void saveShouldSaveEmptyInputStream() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldSaveEmptyInputStream(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, new ByteArrayInputStream(EMPTY_BYTEARRAY), LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, new ByteArrayInputStream(EMPTY_BYTEARRAY), storagePolicy).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
assertThat(new String(bytes, StandardCharsets.UTF_8)).isEmpty();
}
- @Test
- default void saveShouldReturnBlobId() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldReturnBlobId(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, storagePolicy).block();
assertThat(blobId).isEqualTo(blobIdFactory().from("31f7a65e315586ac198bd798b6629ce4903d0899476d5741a9f32e2e521b6a66"));
}
- @Test
- default void saveShouldReturnBlobIdOfString() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldReturnBlobIdOfString(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_STRING, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_STRING, storagePolicy).block();
assertThat(blobId).isEqualTo(blobIdFactory().from("31f7a65e315586ac198bd798b6629ce4903d0899476d5741a9f32e2e521b6a66"));
}
- @Test
- default void saveShouldReturnBlobIdOfInputStream() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void saveShouldReturnBlobIdOfInputStream(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, new ByteArrayInputStream(SHORT_BYTEARRAY), LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, new ByteArrayInputStream(SHORT_BYTEARRAY), storagePolicy).block();
assertThat(blobId).isEqualTo(blobIdFactory().from("31f7a65e315586ac198bd798b6629ce4903d0899476d5741a9f32e2e521b6a66"));
}
@@ -145,36 +167,39 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
.isExactlyInstanceOf(ObjectNotFoundException.class);
}
- @Test
- default void readBytesShouldReturnSavedData() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void readBytesShouldReturnSavedData(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, storagePolicy).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
assertThat(bytes).isEqualTo(SHORT_BYTEARRAY);
}
- @Test
- default void readBytesShouldReturnLongSavedData() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void readBytesShouldReturnLongSavedData(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, ELEVEN_KILOBYTES, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, ELEVEN_KILOBYTES, storagePolicy).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
assertThat(bytes).isEqualTo(ELEVEN_KILOBYTES);
}
- @Test
- default void readBytesShouldReturnBigSavedData() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void readBytesShouldReturnBigSavedData(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, storagePolicy).block();
byte[] bytes = store.readBytes(defaultBucketName, blobId).block();
@@ -190,37 +215,40 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
.isInstanceOf(ObjectNotFoundException.class);
}
- @Test
- default void readShouldReturnSavedData() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void readShouldReturnSavedData(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, storagePolicy).block();
InputStream read = store.read(defaultBucketName, blobId);
assertThat(read).hasSameContentAs(new ByteArrayInputStream(SHORT_BYTEARRAY));
}
- @Test
- default void readShouldReturnLongSavedData() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void readShouldReturnLongSavedData(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- BlobId blobId = store.save(defaultBucketName, ELEVEN_KILOBYTES, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, ELEVEN_KILOBYTES, storagePolicy).block();
InputStream read = store.read(defaultBucketName, blobId);
assertThat(read).hasSameContentAs(new ByteArrayInputStream(ELEVEN_KILOBYTES));
}
- @Test
- default void readShouldReturnBigSavedData() {
+ @ParameterizedTest
+ @MethodSource("storagePolicies")
+ default void readShouldReturnBigSavedData(BlobStore.StoragePolicy storagePolicy) {
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
// 12 MB of text
- BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, LOW_COST).block();
+ BlobId blobId = store.save(defaultBucketName, TWELVE_MEGABYTES, storagePolicy).block();
InputStream read = store.read(defaultBucketName, blobId);
diff --git a/server/blob/blob-cassandra/pom.xml b/server/blob/blob-cassandra/pom.xml
index dcaf57c..6f4e2ff 100644
--- a/server/blob/blob-cassandra/pom.xml
+++ b/server/blob/blob-cassandra/pom.xml
@@ -72,6 +72,11 @@
<artifactId>guava</artifactId>
</dependency>
<dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<scope>test</scope>
diff --git a/server/blob/blob-memory/pom.xml b/server/blob/blob-memory/pom.xml
index 7d5e709..243f456 100644
--- a/server/blob/blob-memory/pom.xml
+++ b/server/blob/blob-memory/pom.xml
@@ -57,6 +57,11 @@
<artifactId>commons-io</artifactId>
</dependency>
<dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 08/10: JAMES-2921
s/swiftBlobStoreProvider/objectStorageBlobStoreProvider
Posted by bt...@apache.org.
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 f22e18fb6b9fef81f90c6f9698bdd31f19569b8e
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jan 7 16:52:40 2020 +0700
JAMES-2921 s/swiftBlobStoreProvider/objectStorageBlobStoreProvider
Because this codePath is also used with S3
---
.../org/apache/james/modules/blobstore/BlobStoreChoosingModule.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
index 82e74cf..4e79e50 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
@@ -75,17 +75,17 @@ public class BlobStoreChoosingModule extends AbstractModule {
@Singleton
BlobStore provideBlobStore(BlobStoreChoosingConfiguration choosingConfiguration,
Provider<CassandraBlobStore> cassandraBlobStoreProvider,
- Provider<ObjectStorageBlobStore> swiftBlobStoreProvider,
+ Provider<ObjectStorageBlobStore> objectStorageBlobStoreProvider,
HybridBlobStore.Configuration hybridBlobStoreConfiguration) {
switch (choosingConfiguration.getImplementation()) {
case OBJECTSTORAGE:
- return swiftBlobStoreProvider.get();
+ return objectStorageBlobStoreProvider.get();
case CASSANDRA:
return cassandraBlobStoreProvider.get();
case HYBRID:
return HybridBlobStore.builder()
- .lowCost(swiftBlobStoreProvider.get())
+ .lowCost(objectStorageBlobStoreProvider.get())
.highPerformance(cassandraBlobStoreProvider.get())
.configuration(hybridBlobStoreConfiguration)
.build();
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/10: JAMES-2921 configure hybrid blobStore
threshold
Posted by bt...@apache.org.
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 23d311db5a5adfaee4a4a4b0490250907e820be0
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jan 7 14:05:18 2020 +0700
JAMES-2921 configure hybrid blobStore threshold
---
.../destination/conf/blob.properties | 6 ++
.../destination/conf/blob.properties | 6 ++
server/blob/blob-union/pom.xml | 4 ++
.../apache/james/blob/union/HybridBlobStore.java | 67 +++++++++++++++++++---
.../james/blob/union/HybridBlobStoreTest.java | 22 ++++++-
.../modules/blobstore/BlobStoreChoosingModule.java | 15 ++++-
.../blobstore/BlobStoreChoosingModuleTest.java | 58 ++++++++++++++++++-
7 files changed, 164 insertions(+), 14 deletions(-)
diff --git a/dockerfiles/run/guice/cassandra-rabbitmq-ldap/destination/conf/blob.properties b/dockerfiles/run/guice/cassandra-rabbitmq-ldap/destination/conf/blob.properties
index f00fb5d..ea41c7f 100644
--- a/dockerfiles/run/guice/cassandra-rabbitmq-ldap/destination/conf/blob.properties
+++ b/dockerfiles/run/guice/cassandra-rabbitmq-ldap/destination/conf/blob.properties
@@ -6,6 +6,12 @@
# hybrid is using both objectstorage for unfrequently read or big blobs & cassandra for small, often read blobs
implementation=objectstorage
+# ========================================= Hybrid BlobStore ======================================
+# hybrid is using both objectstorage for unfrequently read or big blobs & cassandra for small, often read blobs
+# Size threshold for considering a blob as 'big', causing it to be saved in the low cost blobStore
+# Optional, defaults to 32768 bytes (32KB), must be positive
+hybrid.size.threshold=32768
+
# ============================================== ObjectStorage ============================================
# ========================================= ObjectStorage Codec ======================================
diff --git a/dockerfiles/run/guice/cassandra-rabbitmq/destination/conf/blob.properties b/dockerfiles/run/guice/cassandra-rabbitmq/destination/conf/blob.properties
index f00fb5d..ea41c7f 100644
--- a/dockerfiles/run/guice/cassandra-rabbitmq/destination/conf/blob.properties
+++ b/dockerfiles/run/guice/cassandra-rabbitmq/destination/conf/blob.properties
@@ -6,6 +6,12 @@
# hybrid is using both objectstorage for unfrequently read or big blobs & cassandra for small, often read blobs
implementation=objectstorage
+# ========================================= Hybrid BlobStore ======================================
+# hybrid is using both objectstorage for unfrequently read or big blobs & cassandra for small, often read blobs
+# Size threshold for considering a blob as 'big', causing it to be saved in the low cost blobStore
+# Optional, defaults to 32768 bytes (32KB), must be positive
+hybrid.size.threshold=32768
+
# ============================================== ObjectStorage ============================================
# ========================================= ObjectStorage Codec ======================================
diff --git a/server/blob/blob-union/pom.xml b/server/blob/blob-union/pom.xml
index 9df5cee..0e868fa 100644
--- a/server/blob/blob-union/pom.xml
+++ b/server/blob/blob-union/pom.xml
@@ -67,6 +67,10 @@
<scope>test</scope>
</dependency>
<dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-configuration2</artifactId>
+ </dependency>
+ <dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-params</artifactId>
<scope>test</scope>
diff --git a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
index 45d1813..03328a0 100644
--- a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
+++ b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
@@ -22,6 +22,8 @@ package org.apache.james.blob.union;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.util.Objects;
+import java.util.Optional;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.BlobStore;
@@ -43,43 +45,90 @@ public class HybridBlobStore implements BlobStore {
@FunctionalInterface
public interface RequirePerforming {
- Builder highPerformance(BlobStore blobStore);
+ RequireConfiguration highPerformance(BlobStore blobStore);
+ }
+
+ @FunctionalInterface
+ public interface RequireConfiguration {
+ Builder configuration(Configuration configuration);
}
public static class Builder {
private final BlobStore lowCostBlobStore;
private final BlobStore highPerformanceBlobStore;
+ private final Configuration configuration;
- Builder(BlobStore lowCostBlobStore, BlobStore highPerformanceBlobStore) {
+ Builder(BlobStore lowCostBlobStore, BlobStore highPerformanceBlobStore, Configuration configuration) {
this.lowCostBlobStore = lowCostBlobStore;
this.highPerformanceBlobStore = highPerformanceBlobStore;
+ this.configuration = configuration;
}
public HybridBlobStore build() {
return new HybridBlobStore(
lowCostBlobStore,
- highPerformanceBlobStore);
+ highPerformanceBlobStore,
+ configuration);
+ }
+ }
+
+ public static class Configuration {
+ public static final int DEFAULT_SIZE_THREASHOLD = 32 * 1024;
+ public static final Configuration DEFAULT = new Configuration(DEFAULT_SIZE_THREASHOLD);
+ private static final String PROPERTY_NAME = "hybrid.size.threshold";
+
+ public static Configuration from(org.apache.commons.configuration2.Configuration propertiesConfiguration) {
+ return new Configuration(Optional.ofNullable(propertiesConfiguration.getInteger(PROPERTY_NAME, null))
+ .orElse(DEFAULT_SIZE_THREASHOLD));
+ }
+
+ private final int sizeThreshold;
+
+ public Configuration(int sizeThreshold) {
+ Preconditions.checkArgument(sizeThreshold >= 0, "'" + PROPERTY_NAME + "' needs to be positive");
+
+ this.sizeThreshold = sizeThreshold;
+ }
+
+ public int getSizeThreshold() {
+ return sizeThreshold;
+ }
+
+ @Override
+ public final boolean equals(Object o) {
+ if (o instanceof Configuration) {
+ Configuration that = (Configuration) o;
+
+ return Objects.equals(this.sizeThreshold, that.sizeThreshold);
+ }
+ return false;
+ }
+
+ @Override
+ public final int hashCode() {
+ return Objects.hash(sizeThreshold);
}
}
private static final Logger LOGGER = LoggerFactory.getLogger(HybridBlobStore.class);
- private static final int SIZE_THRESHOLD = 32 * 1024;
public static RequireLowCost builder() {
- return lowCost -> highPerformance -> new Builder(lowCost, highPerformance);
+ return lowCost -> highPerformance -> configuration -> new Builder(lowCost, highPerformance, configuration);
}
private final BlobStore lowCostBlobStore;
private final BlobStore highPerformanceBlobStore;
+ private final Configuration configuration;
- private HybridBlobStore(BlobStore lowCostBlobStore, BlobStore highPerformanceBlobStore) {
+ private HybridBlobStore(BlobStore lowCostBlobStore, BlobStore highPerformanceBlobStore, Configuration configuration) {
this.lowCostBlobStore = lowCostBlobStore;
this.highPerformanceBlobStore = highPerformanceBlobStore;
+ this.configuration = configuration;
}
@Override
public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
- return selectBlobStore(storagePolicy, Mono.just(data.length > SIZE_THRESHOLD))
+ return selectBlobStore(storagePolicy, Mono.just(data.length > configuration.getSizeThreshold()))
.flatMap(blobStore -> blobStore.save(bucketName, data, storagePolicy));
}
@@ -87,7 +136,7 @@ public class HybridBlobStore implements BlobStore {
public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
Preconditions.checkNotNull(data);
- BufferedInputStream bufferedInputStream = new BufferedInputStream(data, SIZE_THRESHOLD + 1);
+ BufferedInputStream bufferedInputStream = new BufferedInputStream(data, configuration.getSizeThreshold() + 1);
return selectBlobStore(storagePolicy, Mono.fromCallable(() -> isItABigStream(bufferedInputStream)))
.flatMap(blobStore -> blobStore.save(bucketName, bufferedInputStream, storagePolicy));
}
@@ -112,7 +161,7 @@ public class HybridBlobStore implements BlobStore {
private boolean isItABigStream(InputStream bufferedData) throws IOException {
bufferedData.mark(0);
- bufferedData.skip(SIZE_THRESHOLD);
+ bufferedData.skip(configuration.getSizeThreshold());
boolean isItABigStream = bufferedData.read() != -1;
bufferedData.reset();
return isItABigStream;
diff --git a/server/blob/blob-union/src/test/java/org/apache/james/blob/union/HybridBlobStoreTest.java b/server/blob/blob-union/src/test/java/org/apache/james/blob/union/HybridBlobStoreTest.java
index 97bb738..ad9d38f 100644
--- a/server/blob/blob-union/src/test/java/org/apache/james/blob/union/HybridBlobStoreTest.java
+++ b/server/blob/blob-union/src/test/java/org/apache/james/blob/union/HybridBlobStoreTest.java
@@ -19,8 +19,8 @@
package org.apache.james.blob.union;
-import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.apache.james.blob.api.BlobStore.StoragePolicy.HIGH_PERFORMANCE;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
@@ -45,6 +45,7 @@ import org.junit.jupiter.api.Test;
import com.github.fge.lambdas.Throwing;
import com.google.common.base.MoreObjects;
+import nl.jqno.equalsverifier.EqualsVerifier;
import reactor.core.publisher.Mono;
class HybridBlobStoreTest implements BlobStoreContract {
@@ -161,6 +162,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
hybridBlobStore = HybridBlobStore.builder()
.lowCost(lowCostBlobStore)
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
}
@@ -281,6 +283,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
.lowCost(new ThrowingBlobStore())
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
assertThatThrownBy(() -> hybridBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block())
@@ -293,6 +296,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
.lowCost(new ThrowingBlobStore())
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
assertThatThrownBy(() -> hybridBlobStore.save(hybridBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LOW_COST).block())
@@ -309,6 +313,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
.lowCost(new FailingBlobStore())
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
assertThatThrownBy(() -> hybridBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block())
@@ -321,6 +326,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
.lowCost(new FailingBlobStore())
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
assertThatThrownBy(() -> hybridBlobStore.save(hybridBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LOW_COST).block())
@@ -338,6 +344,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
.lowCost(new ThrowingBlobStore())
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
@@ -352,6 +359,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
.lowCost(new ThrowingBlobStore())
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
@@ -370,6 +378,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
.lowCost(new FailingBlobStore())
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
@@ -383,6 +392,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
.lowCost(new FailingBlobStore())
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
@@ -469,6 +479,7 @@ class HybridBlobStoreTest implements BlobStoreContract {
hybridBlobStore = HybridBlobStore.builder()
.lowCost(lowCostBlobStore)
.highPerformance(highPerformanceBlobStore)
+ .configuration(HybridBlobStore.Configuration.DEFAULT)
.build();
assertThatThrownBy(() -> hybridBlobStore.getDefaultBucketName())
@@ -513,4 +524,13 @@ class HybridBlobStoreTest implements BlobStoreContract {
assertThatCode(() -> hybridBlobStore.delete(BucketName.DEFAULT, blobIdFactory().randomId()).block())
.doesNotThrowAnyException();
}
+
+ @Nested
+ class ConfigurationTest {
+ @Test
+ void shouldMatchBeanContract() {
+ EqualsVerifier.forClass(HybridBlobStore.Configuration.class)
+ .verify();
+ }
+ }
}
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
index 2ab354d..82e74cf 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
@@ -75,7 +75,8 @@ public class BlobStoreChoosingModule extends AbstractModule {
@Singleton
BlobStore provideBlobStore(BlobStoreChoosingConfiguration choosingConfiguration,
Provider<CassandraBlobStore> cassandraBlobStoreProvider,
- Provider<ObjectStorageBlobStore> swiftBlobStoreProvider) {
+ Provider<ObjectStorageBlobStore> swiftBlobStoreProvider,
+ HybridBlobStore.Configuration hybridBlobStoreConfiguration) {
switch (choosingConfiguration.getImplementation()) {
case OBJECTSTORAGE:
@@ -86,6 +87,7 @@ public class BlobStoreChoosingModule extends AbstractModule {
return HybridBlobStore.builder()
.lowCost(swiftBlobStoreProvider.get())
.highPerformance(cassandraBlobStoreProvider.get())
+ .configuration(hybridBlobStoreConfiguration)
.build();
default:
throw new RuntimeException(String.format("can not get the right blobstore provider with configuration %s",
@@ -93,4 +95,15 @@ public class BlobStoreChoosingModule extends AbstractModule {
}
}
+ @Provides
+ @Singleton
+ @VisibleForTesting
+ HybridBlobStore.Configuration providesHybridBlobStoreConfiguration(PropertiesProvider propertiesProvider) {
+ try {
+ Configuration configuration = propertiesProvider.getConfigurations(ConfigurationComponent.NAMES);
+ return HybridBlobStore.Configuration.from(configuration);
+ } catch (FileNotFoundException | ConfigurationException e) {
+ return HybridBlobStore.Configuration.DEFAULT;
+ }
+ }
}
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingModuleTest.java b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingModuleTest.java
index 4e04a5b..810cf50 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingModuleTest.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingModuleTest.java
@@ -131,11 +131,63 @@ class BlobStoreChoosingModuleTest {
}
@Test
+ void providesHybridBlobStoreConfigurationShouldThrowWhenNegative() {
+ BlobStoreChoosingModule module = new BlobStoreChoosingModule();
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.addProperty("hybrid.size.threshold", -1);
+ FakePropertiesProvider propertyProvider = FakePropertiesProvider.builder()
+ .register(ConfigurationComponent.NAME, configuration)
+ .build();
+
+ assertThatThrownBy(() -> module.providesHybridBlobStoreConfiguration(propertyProvider))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void providesHybridBlobStoreConfigurationShouldNotThrowWhenZero() {
+ BlobStoreChoosingModule module = new BlobStoreChoosingModule();
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.addProperty("hybrid.size.threshold", 0);
+ FakePropertiesProvider propertyProvider = FakePropertiesProvider.builder()
+ .register(ConfigurationComponent.NAME, configuration)
+ .build();
+
+ assertThat(module.providesHybridBlobStoreConfiguration(propertyProvider))
+ .isEqualTo(new HybridBlobStore.Configuration(0));
+ }
+
+ @Test
+ void providesHybridBlobStoreConfigurationShouldReturnConfiguration() {
+ BlobStoreChoosingModule module = new BlobStoreChoosingModule();
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.addProperty("hybrid.size.threshold", 36);
+ FakePropertiesProvider propertyProvider = FakePropertiesProvider.builder()
+ .register(ConfigurationComponent.NAME, configuration)
+ .build();
+
+ assertThat(module.providesHybridBlobStoreConfiguration(propertyProvider))
+ .isEqualTo(new HybridBlobStore.Configuration(36));
+ }
+
+ @Test
+ void providesHybridBlobStoreConfigurationShouldReturnConfigurationWhenLegacyFile() {
+ BlobStoreChoosingModule module = new BlobStoreChoosingModule();
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.addProperty("hybrid.size.threshold", 36);
+ FakePropertiesProvider propertyProvider = FakePropertiesProvider.builder()
+ .register(ConfigurationComponent.LEGACY, configuration)
+ .build();
+
+ assertThat(module.providesHybridBlobStoreConfiguration(propertyProvider))
+ .isEqualTo(new HybridBlobStore.Configuration(36));
+ }
+
+ @Test
void provideBlobStoreShouldReturnCassandraBlobStoreWhenCassandraConfigured() {
BlobStoreChoosingModule module = new BlobStoreChoosingModule();
assertThat(module.provideBlobStore(BlobStoreChoosingConfiguration.cassandra(),
- CASSANDRA_BLOBSTORE_PROVIDER, OBJECT_STORAGE_BLOBSTORE_PROVIDER))
+ CASSANDRA_BLOBSTORE_PROVIDER, OBJECT_STORAGE_BLOBSTORE_PROVIDER, HybridBlobStore.Configuration.DEFAULT))
.isEqualTo(CASSANDRA_BLOBSTORE);
}
@@ -144,7 +196,7 @@ class BlobStoreChoosingModuleTest {
BlobStoreChoosingModule module = new BlobStoreChoosingModule();
assertThat(module.provideBlobStore(BlobStoreChoosingConfiguration.cassandra(),
- CASSANDRA_BLOBSTORE_PROVIDER, OBJECT_STORAGE_BLOBSTORE_PROVIDER))
+ CASSANDRA_BLOBSTORE_PROVIDER, OBJECT_STORAGE_BLOBSTORE_PROVIDER, HybridBlobStore.Configuration.DEFAULT))
.isEqualTo(CASSANDRA_BLOBSTORE);
}
@@ -153,7 +205,7 @@ class BlobStoreChoosingModuleTest {
BlobStoreChoosingModule module = new BlobStoreChoosingModule();
assertThat(module.provideBlobStore(BlobStoreChoosingConfiguration.hybrid(),
- CASSANDRA_BLOBSTORE_PROVIDER, OBJECT_STORAGE_BLOBSTORE_PROVIDER))
+ CASSANDRA_BLOBSTORE_PROVIDER, OBJECT_STORAGE_BLOBSTORE_PROVIDER, HybridBlobStore.Configuration.DEFAULT))
.isInstanceOf(HybridBlobStore.class);
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 09/10: [Refactoring] throwing tests for BlobStore
should actually read() to ensure lazy streams are evaluated
Posted by bt...@apache.org.
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 694d36a2f14b788743e3ffe8c91caef7abc71af0
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Mon Jan 6 15:12:20 2020 +0100
[Refactoring] throwing tests for BlobStore should actually read() to ensure lazy streams are evaluated
---
.../src/test/java/org/apache/james/blob/api/BlobStoreContract.java | 2 +-
.../test/java/org/apache/james/blob/api/BucketBlobStoreContract.java | 4 ++--
.../test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
index 2067d70..0fd6961 100644
--- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
+++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreContract.java
@@ -211,7 +211,7 @@ public interface BlobStoreContract extends DeleteBlobStoreContract, BucketBlobSt
BlobStore store = testee();
BucketName defaultBucketName = store.getDefaultBucketName();
- assertThatThrownBy(() -> store.read(defaultBucketName, blobIdFactory().from("unknown")))
+ assertThatThrownBy(() -> store.read(defaultBucketName, blobIdFactory().from("unknown")).read())
.isInstanceOf(ObjectNotFoundException.class);
}
diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BucketBlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BucketBlobStoreContract.java
index a52240e..0d993bd 100644
--- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BucketBlobStoreContract.java
+++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BucketBlobStoreContract.java
@@ -55,7 +55,7 @@ public interface BucketBlobStoreContract {
BlobId blobId = store.save(CUSTOM, SHORT_BYTEARRAY, LOW_COST).block();
store.deleteBucket(CUSTOM).block();
- assertThatThrownBy(() -> store.read(CUSTOM, blobId))
+ assertThatThrownBy(() -> store.read(CUSTOM, blobId).read())
.isInstanceOf(ObjectStoreException.class);
}
@@ -117,7 +117,7 @@ public interface BucketBlobStoreContract {
BlobStore store = testee();
BlobId blobId = store.save(BucketName.DEFAULT, SHORT_BYTEARRAY, LOW_COST).block();
- assertThatThrownBy(() -> store.read(CUSTOM, blobId))
+ assertThatThrownBy(() -> store.read(CUSTOM, blobId).read())
.isInstanceOf(ObjectStoreException.class);
}
diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java
index 616f355..b0cbeb9 100644
--- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java
+++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeleteBlobStoreContract.java
@@ -65,7 +65,7 @@ public interface DeleteBlobStoreContract {
BlobId blobId = store.save(defaultBucketName, SHORT_BYTEARRAY, LOW_COST).block();
store.delete(defaultBucketName, blobId).block();
- assertThatThrownBy(() -> store.read(defaultBucketName, blobId))
+ assertThatThrownBy(() -> store.read(defaultBucketName, blobId).read())
.isInstanceOf(ObjectStoreException.class);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/10: JAMES-2921 HybridBlobStore upgrade
instructions
Posted by bt...@apache.org.
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 bf918b61e58dccec0a6aebd0e30eb6fb7fd8b10b
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon Jan 6 10:56:11 2020 +0700
JAMES-2921 HybridBlobStore upgrade instructions
Users of the Union blobStore needs to be using the Hybrid blobStore
---
.../destination/conf/blob.properties | 4 ++--
.../destination/conf/blob.properties | 4 ++--
upgrade-instructions.md | 22 +++++++++++++++++++++-
3 files changed, 25 insertions(+), 5 deletions(-)
diff --git a/dockerfiles/run/guice/cassandra-rabbitmq-ldap/destination/conf/blob.properties b/dockerfiles/run/guice/cassandra-rabbitmq-ldap/destination/conf/blob.properties
index b7085a1..f00fb5d 100644
--- a/dockerfiles/run/guice/cassandra-rabbitmq-ldap/destination/conf/blob.properties
+++ b/dockerfiles/run/guice/cassandra-rabbitmq-ldap/destination/conf/blob.properties
@@ -2,8 +2,8 @@
# Read https://james.apache.org/server/config-blobstore.html for further details
# Choose your BlobStore implementation
-# Mandatory, allowed values are: cassandra, objectstorage, union
-# union is using both objectstorage as the current BlobStore & cassandra as the legacy BlobStore
+# Mandatory, allowed values are: cassandra, objectstorage, hybrid
+# hybrid is using both objectstorage for unfrequently read or big blobs & cassandra for small, often read blobs
implementation=objectstorage
# ============================================== ObjectStorage ============================================
diff --git a/dockerfiles/run/guice/cassandra-rabbitmq/destination/conf/blob.properties b/dockerfiles/run/guice/cassandra-rabbitmq/destination/conf/blob.properties
index b7085a1..f00fb5d 100644
--- a/dockerfiles/run/guice/cassandra-rabbitmq/destination/conf/blob.properties
+++ b/dockerfiles/run/guice/cassandra-rabbitmq/destination/conf/blob.properties
@@ -2,8 +2,8 @@
# Read https://james.apache.org/server/config-blobstore.html for further details
# Choose your BlobStore implementation
-# Mandatory, allowed values are: cassandra, objectstorage, union
-# union is using both objectstorage as the current BlobStore & cassandra as the legacy BlobStore
+# Mandatory, allowed values are: cassandra, objectstorage, hybrid
+# hybrid is using both objectstorage for unfrequently read or big blobs & cassandra for small, often read blobs
implementation=objectstorage
# ============================================== ObjectStorage ============================================
diff --git a/upgrade-instructions.md b/upgrade-instructions.md
index f647ae3..dd9c470 100644
--- a/upgrade-instructions.md
+++ b/upgrade-instructions.md
@@ -24,7 +24,27 @@ Change list:
- [JAMES-2703 Post 3.4.0 release removals](#james-2703-post-340-release-removals)
- [Health checks routes return code changes](#health-checks-routes-return-code-changes)
- [User mailboxes reIndexing endpoint change](#user-mailboxes-reindexing-endpoint-change)
-
+ - [Hybrid blobStore replaces Union blobStore](#hybrid-blobstore-replaces-union-blobstore)
+
+### Hybrid blobStore replaces Union blobStore
+
+Date 6/01/2020
+
+SHA-1 XXX
+
+Concerned products: Guice distributed James server
+
+Union blobStore, allowing to store older blobs within Cassandra while storing new blobs into object storage, had been removed.
+
+Hybrid blobStore had been replacing it, allowing to store blobs either in a low cost blobStore or in a high performance blobStore, allowing thus some performance
+improvement for small, often read blobs while big or unfrequently read blobs keeps being stored cheaply.
+
+Users relying on the Union blobStore will need to adopt Hybrid blobStore. Please adjust "blob.properties" accordingly:
+
+```
+implementation=hybrid
+```
+
### User mailboxes reIndexing endpoint change
Date 16/12/2019
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/10: JAMES-2921 Propose an Hybrid blobStore
Posted by bt...@apache.org.
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 7cfa6fe9d5909b414ee61ac10e460f1024087884
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jan 3 18:06:59 2020 +0700
JAMES-2921 Propose an Hybrid blobStore
---
.../apache/james/blob/union/HybridBlobStore.java | 182 +++++++
.../apache/james/blob/union/UnionBlobStore.java | 223 --------
.../james/blob/union/HybridBlobStoreTest.java | 516 ++++++++++++++++++
.../james/blob/union/UnionBlobStoreTest.java | 604 ---------------------
.../blobstore/BlobStoreChoosingConfiguration.java | 6 +-
.../modules/blobstore/BlobStoreChoosingModule.java | 10 +-
.../BlobStoreChoosingConfigurationTest.java | 14 +-
.../blobstore/BlobStoreChoosingModuleTest.java | 14 +-
8 files changed, 720 insertions(+), 849 deletions(-)
diff --git a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
new file mode 100644
index 0000000..45d1813
--- /dev/null
+++ b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/HybridBlobStore.java
@@ -0,0 +1,182 @@
+/****************************************************************
+ * 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.blob.union;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.james.blob.api.BlobId;
+import org.apache.james.blob.api.BlobStore;
+import org.apache.james.blob.api.BucketName;
+import org.apache.james.blob.api.ObjectNotFoundException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Preconditions;
+
+import reactor.core.publisher.Mono;
+
+public class HybridBlobStore implements BlobStore {
+ @FunctionalInterface
+ public interface RequireLowCost {
+ RequirePerforming lowCost(BlobStore blobStore);
+ }
+
+ @FunctionalInterface
+ public interface RequirePerforming {
+ Builder highPerformance(BlobStore blobStore);
+ }
+
+ public static class Builder {
+ private final BlobStore lowCostBlobStore;
+ private final BlobStore highPerformanceBlobStore;
+
+ Builder(BlobStore lowCostBlobStore, BlobStore highPerformanceBlobStore) {
+ this.lowCostBlobStore = lowCostBlobStore;
+ this.highPerformanceBlobStore = highPerformanceBlobStore;
+ }
+
+ public HybridBlobStore build() {
+ return new HybridBlobStore(
+ lowCostBlobStore,
+ highPerformanceBlobStore);
+ }
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(HybridBlobStore.class);
+ private static final int SIZE_THRESHOLD = 32 * 1024;
+
+ public static RequireLowCost builder() {
+ return lowCost -> highPerformance -> new Builder(lowCost, highPerformance);
+ }
+
+ private final BlobStore lowCostBlobStore;
+ private final BlobStore highPerformanceBlobStore;
+
+ private HybridBlobStore(BlobStore lowCostBlobStore, BlobStore highPerformanceBlobStore) {
+ this.lowCostBlobStore = lowCostBlobStore;
+ this.highPerformanceBlobStore = highPerformanceBlobStore;
+ }
+
+ @Override
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
+ return selectBlobStore(storagePolicy, Mono.just(data.length > SIZE_THRESHOLD))
+ .flatMap(blobStore -> blobStore.save(bucketName, data, storagePolicy));
+ }
+
+ @Override
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
+ Preconditions.checkNotNull(data);
+
+ BufferedInputStream bufferedInputStream = new BufferedInputStream(data, SIZE_THRESHOLD + 1);
+ return selectBlobStore(storagePolicy, Mono.fromCallable(() -> isItABigStream(bufferedInputStream)))
+ .flatMap(blobStore -> blobStore.save(bucketName, bufferedInputStream, storagePolicy));
+ }
+
+ private Mono<BlobStore> selectBlobStore(StoragePolicy storagePolicy, Mono<Boolean> largeData) {
+ switch (storagePolicy) {
+ case LOW_COST:
+ return Mono.just(lowCostBlobStore);
+ case SIZE_BASED:
+ return largeData.map(isLarge -> {
+ if (isLarge) {
+ return lowCostBlobStore;
+ }
+ return highPerformanceBlobStore;
+ });
+ case HIGH_PERFORMANCE:
+ return Mono.just(highPerformanceBlobStore);
+ default:
+ throw new RuntimeException("Unknown storage policy: " + storagePolicy);
+ }
+ }
+
+ private boolean isItABigStream(InputStream bufferedData) throws IOException {
+ bufferedData.mark(0);
+ bufferedData.skip(SIZE_THRESHOLD);
+ boolean isItABigStream = bufferedData.read() != -1;
+ bufferedData.reset();
+ return isItABigStream;
+ }
+
+ @Override
+ public BucketName getDefaultBucketName() {
+ Preconditions.checkState(
+ lowCostBlobStore.getDefaultBucketName()
+ .equals(highPerformanceBlobStore.getDefaultBucketName()),
+ "lowCostBlobStore and highPerformanceBlobStore doen't have same defaultBucketName which could lead to " +
+ "unexpected result when interact with other APIs");
+
+ return lowCostBlobStore.getDefaultBucketName();
+ }
+
+ @Override
+ public Mono<byte[]> readBytes(BucketName bucketName, BlobId blobId) {
+ return Mono.defer(() -> highPerformanceBlobStore.readBytes(bucketName, blobId))
+ .onErrorResume(this::logAndReturnEmpty)
+ .switchIfEmpty(Mono.defer(() -> lowCostBlobStore.readBytes(bucketName, blobId)));
+ }
+
+ @Override
+ public InputStream read(BucketName bucketName, BlobId blobId) {
+ try {
+ return highPerformanceBlobStore.read(bucketName, blobId);
+ } catch (ObjectNotFoundException e) {
+ return lowCostBlobStore.read(bucketName, blobId);
+ } catch (Exception e) {
+ LOGGER.error("Error reading {} {} in {}, falling back to {}", bucketName, blobId, highPerformanceBlobStore, lowCostBlobStore);
+ return lowCostBlobStore.read(bucketName, blobId);
+ }
+ }
+
+ @Override
+ public Mono<Void> deleteBucket(BucketName bucketName) {
+ return Mono.defer(() -> lowCostBlobStore.deleteBucket(bucketName))
+ .and(highPerformanceBlobStore.deleteBucket(bucketName))
+ .onErrorResume(this::logDeleteFailureAndReturnEmpty);
+ }
+
+ @Override
+ public Mono<Void> delete(BucketName bucketName, BlobId blobId) {
+ return Mono.defer(() -> lowCostBlobStore.delete(bucketName, blobId))
+ .and(highPerformanceBlobStore.delete(bucketName, blobId))
+ .onErrorResume(this::logDeleteFailureAndReturnEmpty);
+ }
+
+ private <T> Mono<T> logAndReturnEmpty(Throwable throwable) {
+ LOGGER.error("error happens from current blob store, fall back to lowCost blob store", throwable);
+ return Mono.empty();
+ }
+
+ private <T> Mono<T> logDeleteFailureAndReturnEmpty(Throwable throwable) {
+ LOGGER.error("Cannot delete from either lowCost or highPerformance blob store", throwable);
+ return Mono.empty();
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .add("lowCostBlobStore", lowCostBlobStore)
+ .add("highPerformanceBlobStore", highPerformanceBlobStore)
+ .toString();
+ }
+}
diff --git a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/UnionBlobStore.java b/server/blob/blob-union/src/main/java/org/apache/james/blob/union/UnionBlobStore.java
deleted file mode 100644
index 37f7c8a..0000000
--- a/server/blob/blob-union/src/main/java/org/apache/james/blob/union/UnionBlobStore.java
+++ /dev/null
@@ -1,223 +0,0 @@
-/****************************************************************
- * 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.blob.union;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.PushbackInputStream;
-import java.util.Optional;
-import java.util.function.Function;
-
-import org.apache.james.blob.api.BlobId;
-import org.apache.james.blob.api.BlobStore;
-import org.apache.james.blob.api.BucketName;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.github.fge.lambdas.Throwing;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.MoreObjects;
-import com.google.common.base.Preconditions;
-
-import reactor.core.publisher.Mono;
-
-public class UnionBlobStore implements BlobStore {
-
- @FunctionalInterface
- public interface StorageOperation<T> {
- Mono<BlobId> save(BucketName bucketName, T data, StoragePolicy storagePolicy);
- }
-
- @FunctionalInterface
- public interface RequireCurrent {
- RequireLegacy current(BlobStore blobStore);
- }
-
- @FunctionalInterface
- public interface RequireLegacy {
- Builder legacy(BlobStore blobStore);
- }
-
- public static class Builder {
- private final BlobStore currentBlobStore;
- private final BlobStore legacyBlobStore;
-
- Builder(BlobStore currentBlobStore, BlobStore legacyBlobStore) {
- this.currentBlobStore = currentBlobStore;
- this.legacyBlobStore = legacyBlobStore;
- }
-
- public UnionBlobStore build() {
- return new UnionBlobStore(
- currentBlobStore,
- legacyBlobStore);
- }
- }
-
- private static final Logger LOGGER = LoggerFactory.getLogger(UnionBlobStore.class);
- private static final int UNAVAILABLE = -1;
-
- public static RequireCurrent builder() {
- return current -> legacy -> new Builder(current, legacy);
- }
-
- private final BlobStore currentBlobStore;
- private final BlobStore legacyBlobStore;
-
- private UnionBlobStore(BlobStore currentBlobStore, BlobStore legacyBlobStore) {
- this.currentBlobStore = currentBlobStore;
- this.legacyBlobStore = legacyBlobStore;
- }
-
- @Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
- try {
- return saveToCurrentFallbackIfFails(bucketName, data, storagePolicy,
- currentBlobStore::save,
- legacyBlobStore::save);
- } catch (Exception e) {
- LOGGER.error("exception directly happens while saving bytes data, fall back to legacy blob store", e);
- return legacyBlobStore.save(bucketName, data, storagePolicy);
- }
- }
-
- @Override
- public Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
- try {
- return saveToCurrentFallbackIfFails(bucketName, data, storagePolicy,
- currentBlobStore::save,
- legacyBlobStore::save);
- } catch (Exception e) {
- LOGGER.error("exception directly happens while saving String data, fall back to legacy blob store", e);
- return legacyBlobStore.save(bucketName, data, storagePolicy);
- }
- }
-
- @Override
- public BucketName getDefaultBucketName() {
- Preconditions.checkState(
- currentBlobStore.getDefaultBucketName()
- .equals(legacyBlobStore.getDefaultBucketName()),
- "currentBlobStore and legacyBlobStore doen't have same defaultBucketName which could lead to " +
- "unexpected result when interact with other APIs");
-
- return currentBlobStore.getDefaultBucketName();
- }
-
- @Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
- try {
- return saveToCurrentFallbackIfFails(bucketName, data, storagePolicy,
- currentBlobStore::save,
- legacyBlobStore::save);
- } catch (Exception e) {
- LOGGER.error("exception directly happens while saving InputStream data, fall back to legacy blob store", e);
- return legacyBlobStore.save(bucketName, data, storagePolicy);
- }
- }
-
- @Override
- public Mono<byte[]> readBytes(BucketName bucketName, BlobId blobId) {
- try {
- return readBytesFallBackIfFailsOrEmptyResult(bucketName, blobId);
- } catch (Exception e) {
- LOGGER.error("exception directly happens while readBytes, fall back to legacy blob store", e);
- return Mono.defer(() -> legacyBlobStore.readBytes(bucketName, blobId));
- }
- }
-
- @Override
- public InputStream read(BucketName bucketName, BlobId blobId) {
- try {
- return readFallBackIfEmptyResult(bucketName, blobId);
- } catch (Exception e) {
- LOGGER.error("exception directly happens while read, fall back to legacy blob store", e);
- return legacyBlobStore.read(bucketName, blobId);
- }
- }
-
- @Override
- public Mono<Void> deleteBucket(BucketName bucketName) {
- return Mono.defer(() -> currentBlobStore.deleteBucket(bucketName))
- .and(legacyBlobStore.deleteBucket(bucketName))
- .onErrorResume(this::logDeleteFailureAndReturnEmpty);
- }
-
- @Override
- public Mono<Void> delete(BucketName bucketName, BlobId blobId) {
- return Mono.defer(() -> currentBlobStore.delete(bucketName, blobId))
- .and(legacyBlobStore.delete(bucketName, blobId))
- .onErrorResume(this::logDeleteFailureAndReturnEmpty);
- }
-
- private InputStream readFallBackIfEmptyResult(BucketName bucketName, BlobId blobId) {
- return Optional.ofNullable(currentBlobStore.read(bucketName, blobId))
- .map(PushbackInputStream::new)
- .filter(Throwing.predicate(this::streamHasContent).sneakyThrow())
- .<InputStream>map(Function.identity())
- .orElseGet(() -> legacyBlobStore.read(bucketName, blobId));
- }
-
- @VisibleForTesting
- boolean streamHasContent(PushbackInputStream pushBackIS) throws IOException {
- int byteRead = pushBackIS.read();
- if (byteRead != UNAVAILABLE) {
- pushBackIS.unread(byteRead);
- return true;
- }
- return false;
- }
-
- private Mono<byte[]> readBytesFallBackIfFailsOrEmptyResult(BucketName bucketName, BlobId blobId) {
- return Mono.defer(() -> currentBlobStore.readBytes(bucketName, blobId))
- .onErrorResume(this::logAndReturnEmpty)
- .switchIfEmpty(legacyBlobStore.readBytes(bucketName, blobId));
- }
-
- private <T> Mono<BlobId> saveToCurrentFallbackIfFails(
- BucketName bucketName,
- T data,
- StoragePolicy storagePolicy,
- StorageOperation<T> currentSavingOperation,
- StorageOperation<T> fallbackSavingOperationSupplier) {
-
- return Mono.defer(() -> currentSavingOperation.save(bucketName, data, storagePolicy))
- .onErrorResume(this::logAndReturnEmpty)
- .switchIfEmpty(Mono.defer(() -> fallbackSavingOperationSupplier.save(bucketName, data, storagePolicy)));
- }
-
- private <T> Mono<T> logAndReturnEmpty(Throwable throwable) {
- LOGGER.error("error happens from current blob store, fall back to legacy blob store", throwable);
- return Mono.empty();
- }
-
- private <T> Mono<T> logDeleteFailureAndReturnEmpty(Throwable throwable) {
- LOGGER.error("Cannot delete from either legacy or current blob store", throwable);
- return Mono.empty();
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(this)
- .add("currentBlobStore", currentBlobStore)
- .add("legacyBlobStore", legacyBlobStore)
- .toString();
- }
-}
diff --git a/server/blob/blob-union/src/test/java/org/apache/james/blob/union/HybridBlobStoreTest.java b/server/blob/blob-union/src/test/java/org/apache/james/blob/union/HybridBlobStoreTest.java
new file mode 100644
index 0000000..97bb738
--- /dev/null
+++ b/server/blob/blob-union/src/test/java/org/apache/james/blob/union/HybridBlobStoreTest.java
@@ -0,0 +1,516 @@
+/****************************************************************
+ * 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.blob.union;
+
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.HIGH_PERFORMANCE;
+import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+
+import org.apache.james.blob.api.BlobId;
+import org.apache.james.blob.api.BlobStore;
+import org.apache.james.blob.api.BlobStoreContract;
+import org.apache.james.blob.api.BucketName;
+import org.apache.james.blob.api.HashBlobId;
+import org.apache.james.blob.api.ObjectNotFoundException;
+import org.apache.james.blob.api.ObjectStoreException;
+import org.apache.james.blob.memory.MemoryBlobStore;
+import org.assertj.core.api.SoftAssertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import com.github.fge.lambdas.Throwing;
+import com.google.common.base.MoreObjects;
+
+import reactor.core.publisher.Mono;
+
+class HybridBlobStoreTest implements BlobStoreContract {
+
+ private static class FailingBlobStore implements BlobStore {
+ @Override
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
+ return Mono.error(new RuntimeException("broken everywhere"));
+ }
+
+ @Override
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
+ return Mono.error(new RuntimeException("broken everywhere"));
+ }
+
+ @Override
+ public Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
+ return Mono.error(new RuntimeException("broken everywhere"));
+ }
+
+ @Override
+ public BucketName getDefaultBucketName() {
+ return BucketName.DEFAULT;
+ }
+
+ @Override
+ public Mono<byte[]> readBytes(BucketName bucketName, BlobId blobId) {
+ return Mono.error(new RuntimeException("broken everywhere"));
+ }
+
+ @Override
+ public InputStream read(BucketName bucketName, BlobId blobId) {
+ throw new RuntimeException("broken everywhere");
+ }
+
+ @Override
+ public Mono<Void> deleteBucket(BucketName bucketName) {
+ return Mono.error(new RuntimeException("broken everywhere"));
+ }
+
+ @Override
+ public Mono<Void> delete(BucketName bucketName, BlobId blobId) {
+ return Mono.error(new RuntimeException("broken everywhere"));
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .toString();
+ }
+ }
+
+ private static class ThrowingBlobStore implements BlobStore {
+
+ @Override
+ public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
+ throw new RuntimeException("broken everywhere");
+ }
+
+ @Override
+ public Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
+ throw new RuntimeException("broken everywhere");
+ }
+
+ @Override
+ public BucketName getDefaultBucketName() {
+ return BucketName.DEFAULT;
+ }
+
+ @Override
+ public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
+ throw new RuntimeException("broken everywhere");
+ }
+
+ @Override
+ public Mono<byte[]> readBytes(BucketName bucketName, BlobId blobId) {
+ throw new RuntimeException("broken everywhere");
+ }
+
+ @Override
+ public InputStream read(BucketName bucketName, BlobId blobId) {
+ throw new RuntimeException("broken everywhere");
+ }
+
+ @Override
+ public Mono<Void> deleteBucket(BucketName bucketName) {
+ return Mono.error(new RuntimeException("broken everywhere"));
+ }
+
+ @Override
+ public Mono<Void> delete(BucketName bucketName, BlobId blobId) {
+ return Mono.error(new RuntimeException("broken everywhere"));
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .toString();
+ }
+ }
+
+ private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
+ private static final String STRING_CONTENT = "blob content";
+ private static final byte [] BLOB_CONTENT = STRING_CONTENT.getBytes();
+
+ private MemoryBlobStore lowCostBlobStore;
+ private MemoryBlobStore highPerformanceBlobStore;
+ private HybridBlobStore hybridBlobStore;
+
+ @BeforeEach
+ void setup() {
+ lowCostBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(lowCostBlobStore)
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+ }
+
+ @Override
+ public BlobStore testee() {
+ return hybridBlobStore;
+ }
+
+ @Override
+ public BlobId.Factory blobIdFactory() {
+ return BLOB_ID_FACTORY;
+ }
+
+ @Nested
+ class StoragePolicyTests {
+ @Test
+ void saveShouldRelyOnLowCostWhenLowCost() {
+ BlobId blobId = hybridBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LOW_COST).block();
+
+ SoftAssertions.assertSoftly(softly -> {
+ softly.assertThat(lowCostBlobStore.read(BucketName.DEFAULT, blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ softly.assertThatThrownBy(() -> highPerformanceBlobStore.read(BucketName.DEFAULT, blobId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ });
+ }
+
+ @Test
+ void saveShouldRelyOnPerformingWhenPerforming() {
+ BlobId blobId = hybridBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, HIGH_PERFORMANCE).block();
+
+ SoftAssertions.assertSoftly(softly -> {
+ softly.assertThat(highPerformanceBlobStore.read(BucketName.DEFAULT, blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ softly.assertThatThrownBy(() -> lowCostBlobStore.read(BucketName.DEFAULT, blobId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ });
+ }
+
+ @Test
+ void saveShouldRelyOnPerformingWhenSizeBasedAndSmall() {
+ BlobId blobId = hybridBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, SIZE_BASED).block();
+
+ SoftAssertions.assertSoftly(softly -> {
+ softly.assertThat(highPerformanceBlobStore.read(BucketName.DEFAULT, blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ softly.assertThatThrownBy(() -> lowCostBlobStore.read(BucketName.DEFAULT, blobId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ });
+ }
+
+ @Test
+ void saveShouldRelyOnLowCostWhenSizeBasedAndBig() {
+ BlobId blobId = hybridBlobStore.save(BucketName.DEFAULT, TWELVE_MEGABYTES, SIZE_BASED).block();
+
+ SoftAssertions.assertSoftly(softly -> {
+ softly.assertThat(lowCostBlobStore.read(BucketName.DEFAULT, blobId))
+ .satisfies(Throwing.consumer(inputStream -> assertThat(inputStream.read()).isGreaterThan(0)));
+ softly.assertThatThrownBy(() -> highPerformanceBlobStore.read(BucketName.DEFAULT, blobId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ });
+ }
+
+ @Test
+ void saveInputStreamShouldRelyOnLowCostWhenLowCost() {
+ BlobId blobId = hybridBlobStore.save(BucketName.DEFAULT, new ByteArrayInputStream(BLOB_CONTENT), LOW_COST).block();
+
+ SoftAssertions.assertSoftly(softly -> {
+ softly.assertThat(lowCostBlobStore.read(BucketName.DEFAULT, blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ softly.assertThatThrownBy(() -> highPerformanceBlobStore.read(BucketName.DEFAULT, blobId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ });
+ }
+
+ @Test
+ void saveInputStreamShouldRelyOnPerformingWhenPerforming() {
+ BlobId blobId = hybridBlobStore.save(BucketName.DEFAULT, new ByteArrayInputStream(BLOB_CONTENT), HIGH_PERFORMANCE).block();
+
+ SoftAssertions.assertSoftly(softly -> {
+ softly.assertThat(highPerformanceBlobStore.read(BucketName.DEFAULT, blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ softly.assertThatThrownBy(() -> lowCostBlobStore.read(BucketName.DEFAULT, blobId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ });
+ }
+
+ @Test
+ void saveInputStreamShouldRelyOnPerformingWhenSizeBasedAndSmall() {
+ BlobId blobId = hybridBlobStore.save(BucketName.DEFAULT, new ByteArrayInputStream(BLOB_CONTENT), SIZE_BASED).block();
+
+ SoftAssertions.assertSoftly(softly -> {
+ softly.assertThat(highPerformanceBlobStore.read(BucketName.DEFAULT, blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ softly.assertThatThrownBy(() -> lowCostBlobStore.read(BucketName.DEFAULT, blobId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ });
+ }
+
+ @Test
+ void saveInputStreamShouldRelyOnLowCostWhenSizeBasedAndBig() {
+ BlobId blobId = hybridBlobStore.save(BucketName.DEFAULT, new ByteArrayInputStream(TWELVE_MEGABYTES), SIZE_BASED).block();
+
+ SoftAssertions.assertSoftly(softly -> {
+ softly.assertThat(lowCostBlobStore.read(BucketName.DEFAULT, blobId))
+ .satisfies(Throwing.consumer(inputStream -> assertThat(inputStream.read()).isGreaterThan(0)));
+ softly.assertThatThrownBy(() -> highPerformanceBlobStore.read(BucketName.DEFAULT, blobId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ });
+ }
+ }
+
+ @Nested
+ class LowCostSaveThrowsExceptionDirectly {
+ @Test
+ void saveShouldFailWhenException() {
+ MemoryBlobStore highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(new ThrowingBlobStore())
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+
+ assertThatThrownBy(() -> hybridBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block())
+ .isInstanceOf(RuntimeException.class);
+ }
+
+ @Test
+ void saveInputStreamShouldFailWhenException() {
+ MemoryBlobStore highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(new ThrowingBlobStore())
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+
+ assertThatThrownBy(() -> hybridBlobStore.save(hybridBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LOW_COST).block())
+ .isInstanceOf(RuntimeException.class);
+ }
+ }
+
+ @Nested
+ class LowCostSaveCompletesExceptionally {
+
+ @Test
+ void saveShouldFailWhenLowCostCompletedExceptionally() {
+ MemoryBlobStore highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(new FailingBlobStore())
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+
+ assertThatThrownBy(() -> hybridBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block())
+ .isInstanceOf(RuntimeException.class);
+ }
+
+ @Test
+ void saveInputStreamShouldFallBackToPerformingWhenLowCostCompletedExceptionally() {
+ MemoryBlobStore highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(new FailingBlobStore())
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+
+ assertThatThrownBy(() -> hybridBlobStore.save(hybridBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LOW_COST).block())
+ .isInstanceOf(RuntimeException.class);
+ }
+
+ }
+
+ @Nested
+ class LowCostReadThrowsExceptionDirectly {
+
+ @Test
+ void readShouldReturnFallbackToPerformingWhenLowCostGotException() {
+ MemoryBlobStore highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(new ThrowingBlobStore())
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+ BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
+
+ assertThat(hybridBlobStore.read(hybridBlobStore.getDefaultBucketName(), blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ }
+
+ @Test
+ void readBytesShouldReturnFallbackToPerformingWhenLowCostGotException() {
+ MemoryBlobStore highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+
+ HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(new ThrowingBlobStore())
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+ BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
+
+ assertThat(hybridBlobStore.readBytes(hybridBlobStore.getDefaultBucketName(), blobId).block())
+ .isEqualTo(BLOB_CONTENT);
+ }
+
+ }
+
+ @Nested
+ class LowCostReadCompletesExceptionally {
+
+ @Test
+ void readShouldReturnFallbackToPerformingWhenLowCostCompletedExceptionally() {
+ MemoryBlobStore highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(new FailingBlobStore())
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+ BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
+
+ assertThat(hybridBlobStore.read(hybridBlobStore.getDefaultBucketName(), blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ }
+
+ @Test
+ void readBytesShouldReturnFallbackToPerformingWhenLowCostCompletedExceptionally() {
+ MemoryBlobStore highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
+ HybridBlobStore hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(new FailingBlobStore())
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+ BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
+
+ assertThat(hybridBlobStore.readBytes(hybridBlobStore.getDefaultBucketName(), blobId).block())
+ .isEqualTo(BLOB_CONTENT);
+ }
+ }
+
+ @Test
+ void readShouldReturnFromLowCostWhenAvailable() {
+ BlobId blobId = lowCostBlobStore.save(lowCostBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
+
+ assertThat(hybridBlobStore.read(hybridBlobStore.getDefaultBucketName(), blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ }
+
+ @Test
+ void readShouldReturnFromPerformingWhenLowCostNotAvailable() {
+ BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
+
+ assertThat(hybridBlobStore.read(hybridBlobStore.getDefaultBucketName(), blobId))
+ .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
+ }
+
+ @Test
+ void readBytesShouldReturnFromLowCostWhenAvailable() {
+ BlobId blobId = lowCostBlobStore.save(lowCostBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
+
+ assertThat(hybridBlobStore.readBytes(lowCostBlobStore.getDefaultBucketName(), blobId).block())
+ .isEqualTo(BLOB_CONTENT);
+ }
+
+ @Test
+ void readBytesShouldReturnFromPerformingWhenLowCostNotAvailable() {
+ BlobId blobId = highPerformanceBlobStore.save(hybridBlobStore.getDefaultBucketName(), BLOB_CONTENT, LOW_COST).block();
+
+ assertThat(hybridBlobStore.readBytes(hybridBlobStore.getDefaultBucketName(), blobId).block())
+ .isEqualTo(BLOB_CONTENT);
+ }
+
+ @Test
+ void deleteBucketShouldDeleteBothLowCostAndPerformingBuckets() {
+ BlobId blobId1 = highPerformanceBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LOW_COST).block();
+ BlobId blobId2 = lowCostBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LOW_COST).block();
+
+ hybridBlobStore.deleteBucket(BucketName.DEFAULT).block();
+
+ assertThatThrownBy(() -> highPerformanceBlobStore.readBytes(BucketName.DEFAULT, blobId1).block())
+ .isInstanceOf(ObjectStoreException.class);
+ assertThatThrownBy(() -> lowCostBlobStore.readBytes(BucketName.DEFAULT, blobId2).block())
+ .isInstanceOf(ObjectStoreException.class);
+ }
+
+ @Test
+ void deleteBucketShouldDeleteLowCostBucketEvenWhenPerformingDoesNotExist() {
+ BlobId blobId = lowCostBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LOW_COST).block();
+
+ hybridBlobStore.deleteBucket(BucketName.DEFAULT).block();
+
+ assertThatThrownBy(() -> lowCostBlobStore.readBytes(BucketName.DEFAULT, blobId).block())
+ .isInstanceOf(ObjectStoreException.class);
+ }
+
+ @Test
+ void deleteBucketShouldDeletePerformingBucketEvenWhenLowCostDoesNotExist() {
+ BlobId blobId = highPerformanceBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LOW_COST).block();
+
+ hybridBlobStore.deleteBucket(BucketName.DEFAULT).block();
+
+ assertThatThrownBy(() -> highPerformanceBlobStore.readBytes(BucketName.DEFAULT, blobId).block())
+ .isInstanceOf(ObjectStoreException.class);
+ }
+
+ @Test
+ void deleteBucketShouldNotThrowWhenLowCostAndPerformingBucketsDoNotExist() {
+ assertThatCode(() -> hybridBlobStore.deleteBucket(BucketName.DEFAULT).block())
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void getDefaultBucketNameShouldThrowWhenBlobStoreDontShareTheSameDefaultBucketName() {
+ lowCostBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY, BucketName.of("lowCost"));
+ highPerformanceBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY, BucketName.of("highPerformance"));
+ hybridBlobStore = HybridBlobStore.builder()
+ .lowCost(lowCostBlobStore)
+ .highPerformance(highPerformanceBlobStore)
+ .build();
+
+ assertThatThrownBy(() -> hybridBlobStore.getDefaultBucketName())
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void deleteShouldDeleteBothLowCostAndPerformingBlob() {
+ BlobId blobId1 = hybridBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LOW_COST).block();
+ BlobId blobId2 = hybridBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, HIGH_PERFORMANCE).block();
+
+ hybridBlobStore.delete(BucketName.DEFAULT, blobId1).block();
+
+ assertThatThrownBy(() -> highPerformanceBlobStore.readBytes(BucketName.DEFAULT, blobId1).block())
+ .isInstanceOf(ObjectStoreException.class);
+ assertThatThrownBy(() -> lowCostBlobStore.readBytes(BucketName.DEFAULT, blobId2).block())
+ .isInstanceOf(ObjectStoreException.class);
+ }
+
+ @Test
+ void deleteShouldDeleteLowCostBlobEvenWhenPerformingDoesNotExist() {
+ BlobId blobId = lowCostBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LOW_COST).block();
+
+ hybridBlobStore.delete(BucketName.DEFAULT, blobId).block();
+
+ assertThatThrownBy(() -> lowCostBlobStore.readBytes(BucketName.DEFAULT, blobId).block())
+ .isInstanceOf(ObjectStoreException.class);
+ }
+
+ @Test
+ void deleteShouldDeletePerformingBlobEvenWhenLowCostDoesNotExist() {
+ BlobId blobId = highPerformanceBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LOW_COST).block();
+
+ hybridBlobStore.delete(BucketName.DEFAULT, blobId).block();
+
+ assertThatThrownBy(() -> highPerformanceBlobStore.readBytes(BucketName.DEFAULT, blobId).block())
+ .isInstanceOf(ObjectStoreException.class);
+ }
+
+ @Test
+ void deleteShouldNotThrowWhenLowCostAndPerformingBlobsDoNotExist() {
+ assertThatCode(() -> hybridBlobStore.delete(BucketName.DEFAULT, blobIdFactory().randomId()).block())
+ .doesNotThrowAnyException();
+ }
+}
diff --git a/server/blob/blob-union/src/test/java/org/apache/james/blob/union/UnionBlobStoreTest.java b/server/blob/blob-union/src/test/java/org/apache/james/blob/union/UnionBlobStoreTest.java
deleted file mode 100644
index e8ccd81..0000000
--- a/server/blob/blob-union/src/test/java/org/apache/james/blob/union/UnionBlobStoreTest.java
+++ /dev/null
@@ -1,604 +0,0 @@
-/****************************************************************
- * 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.blob.union;
-
-import static org.apache.james.blob.api.BlobStore.StoragePolicy.LowCost;
-import static org.apache.james.blob.api.BlobStore.StoragePolicy.LowCost;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatCode;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-import java.io.ByteArrayInputStream;
-import java.io.InputStream;
-import java.io.PushbackInputStream;
-import java.util.List;
-import java.util.function.Function;
-import java.util.stream.Stream;
-
-import org.apache.james.blob.api.BlobId;
-import org.apache.james.blob.api.BlobStore;
-import org.apache.james.blob.api.BlobStoreContract;
-import org.apache.james.blob.api.BucketName;
-import org.apache.james.blob.api.HashBlobId;
-import org.apache.james.blob.api.ObjectStoreException;
-import org.apache.james.blob.memory.MemoryBlobStore;
-import org.apache.james.util.StreamUtils;
-import org.assertj.core.api.SoftAssertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Nested;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestInstance;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
-
-import com.google.common.base.MoreObjects;
-import com.google.common.collect.ImmutableList;
-
-import reactor.core.publisher.Mono;
-
-class UnionBlobStoreTest implements BlobStoreContract {
-
- private static class FailingBlobStore implements BlobStore {
- @Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
- return Mono.error(new RuntimeException("broken everywhere"));
- }
-
- @Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
- return Mono.error(new RuntimeException("broken everywhere"));
- }
-
- @Override
- public Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
- return Mono.error(new RuntimeException("broken everywhere"));
- }
-
- @Override
- public BucketName getDefaultBucketName() {
- return BucketName.DEFAULT;
- }
-
- @Override
- public Mono<byte[]> readBytes(BucketName bucketName, BlobId blobId) {
- return Mono.error(new RuntimeException("broken everywhere"));
- }
-
- @Override
- public InputStream read(BucketName bucketName, BlobId blobId) {
- throw new RuntimeException("broken everywhere");
- }
-
- @Override
- public Mono<Void> deleteBucket(BucketName bucketName) {
- return Mono.error(new RuntimeException("broken everywhere"));
- }
-
- @Override
- public Mono<Void> delete(BucketName bucketName, BlobId blobId) {
- return Mono.error(new RuntimeException("broken everywhere"));
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(this)
- .toString();
- }
- }
-
- private static class ThrowingBlobStore implements BlobStore {
-
- @Override
- public Mono<BlobId> save(BucketName bucketName, byte[] data, StoragePolicy storagePolicy) {
- throw new RuntimeException("broken everywhere");
- }
-
- @Override
- public Mono<BlobId> save(BucketName bucketName, String data, StoragePolicy storagePolicy) {
- throw new RuntimeException("broken everywhere");
- }
-
- @Override
- public BucketName getDefaultBucketName() {
- return BucketName.DEFAULT;
- }
-
- @Override
- public Mono<BlobId> save(BucketName bucketName, InputStream data, StoragePolicy storagePolicy) {
- throw new RuntimeException("broken everywhere");
- }
-
- @Override
- public Mono<byte[]> readBytes(BucketName bucketName, BlobId blobId) {
- throw new RuntimeException("broken everywhere");
- }
-
- @Override
- public InputStream read(BucketName bucketName, BlobId blobId) {
- throw new RuntimeException("broken everywhere");
- }
-
- @Override
- public Mono<Void> deleteBucket(BucketName bucketName) {
- return Mono.error(new RuntimeException("broken everywhere"));
- }
-
- @Override
- public Mono<Void> delete(BucketName bucketName, BlobId blobId) {
- return Mono.error(new RuntimeException("broken everywhere"));
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(this)
- .toString();
- }
- }
-
- private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
- private static final String STRING_CONTENT = "blob content";
- private static final byte [] BLOB_CONTENT = STRING_CONTENT.getBytes();
-
- private MemoryBlobStore currentBlobStore;
- private MemoryBlobStore legacyBlobStore;
- private UnionBlobStore unionBlobStore;
-
- @BeforeEach
- void setup() {
- currentBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- unionBlobStore = UnionBlobStore.builder()
- .current(currentBlobStore)
- .legacy(legacyBlobStore)
- .build();
- }
-
- @Override
- public BlobStore testee() {
- return unionBlobStore;
- }
-
- @Override
- public BlobId.Factory blobIdFactory() {
- return BLOB_ID_FACTORY;
- }
-
- @Nested
- class CurrentSaveThrowsExceptionDirectly {
-
- @Test
- void saveShouldFallBackToLegacyWhenCurrentGotException() {
- MemoryBlobStore legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- UnionBlobStore unionBlobStore = UnionBlobStore.builder()
- .current(new ThrowingBlobStore())
- .legacy(legacyBlobStore)
- .build();
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- SoftAssertions.assertSoftly(softly -> {
- softly.assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- softly.assertThat(legacyBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- });
- }
-
- @Test
- void saveInputStreamShouldFallBackToLegacyWhenCurrentGotException() {
- MemoryBlobStore legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- UnionBlobStore unionBlobStore = UnionBlobStore.builder()
- .current(new ThrowingBlobStore())
- .legacy(legacyBlobStore)
- .build();
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost).block();
-
- SoftAssertions.assertSoftly(softly -> {
- softly.assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- softly.assertThat(legacyBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- });
- }
- }
-
- @Nested
- class CurrentSaveCompletesExceptionally {
-
- @Test
- void saveShouldFallBackToLegacyWhenCurrentCompletedExceptionally() {
- MemoryBlobStore legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- UnionBlobStore unionBlobStore = UnionBlobStore.builder()
- .current(new FailingBlobStore())
- .legacy(legacyBlobStore)
- .build();
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- SoftAssertions.assertSoftly(softly -> {
- softly.assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- softly.assertThat(legacyBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- });
- }
-
- @Test
- void saveInputStreamShouldFallBackToLegacyWhenCurrentCompletedExceptionally() {
- MemoryBlobStore legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- UnionBlobStore unionBlobStore = UnionBlobStore.builder()
- .current(new FailingBlobStore())
- .legacy(legacyBlobStore)
- .build();
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost).block();
-
- SoftAssertions.assertSoftly(softly -> {
- softly.assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- softly.assertThat(legacyBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- });
- }
-
- }
-
- @Nested
- class CurrentReadThrowsExceptionDirectly {
-
- @Test
- void readShouldReturnFallbackToLegacyWhenCurrentGotException() {
- MemoryBlobStore legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- UnionBlobStore unionBlobStore = UnionBlobStore.builder()
- .current(new ThrowingBlobStore())
- .legacy(legacyBlobStore)
- .build();
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- }
-
- @Test
- void readBytesShouldReturnFallbackToLegacyWhenCurrentGotException() {
- MemoryBlobStore legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
-
- UnionBlobStore unionBlobStore = UnionBlobStore.builder()
- .current(new ThrowingBlobStore())
- .legacy(legacyBlobStore)
- .build();
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(unionBlobStore.readBytes(unionBlobStore.getDefaultBucketName(), blobId).block())
- .isEqualTo(BLOB_CONTENT);
- }
-
- }
-
- @Nested
- class CurrentReadCompletesExceptionally {
-
- @Test
- void readShouldReturnFallbackToLegacyWhenCurrentCompletedExceptionally() {
- MemoryBlobStore legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- UnionBlobStore unionBlobStore = UnionBlobStore.builder()
- .current(new FailingBlobStore())
- .legacy(legacyBlobStore)
- .build();
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- }
-
- @Test
- void readBytesShouldReturnFallbackToLegacyWhenCurrentCompletedExceptionally() {
- MemoryBlobStore legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY);
- UnionBlobStore unionBlobStore = UnionBlobStore.builder()
- .current(new FailingBlobStore())
- .legacy(legacyBlobStore)
- .build();
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(unionBlobStore.readBytes(unionBlobStore.getDefaultBucketName(), blobId).block())
- .isEqualTo(BLOB_CONTENT);
- }
- }
-
- @TestInstance(TestInstance.Lifecycle.PER_CLASS)
- @Nested
- class CurrentAndLegacyCouldNotComplete {
-
-
- Stream<Function<UnionBlobStore, Mono<?>>> blobStoreOperationsReturnFutures() {
- return Stream.of(
- blobStore -> blobStore.save(blobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost),
- blobStore -> blobStore.save(blobStore.getDefaultBucketName(), STRING_CONTENT, LowCost),
- blobStore -> blobStore.save(blobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost),
- blobStore -> blobStore.readBytes(blobStore.getDefaultBucketName(), BLOB_ID_FACTORY.randomId()));
- }
-
- Stream<Function<UnionBlobStore, InputStream>> blobStoreOperationsNotReturnFutures() {
- return Stream.of(
- blobStore -> blobStore.read(blobStore.getDefaultBucketName(), BLOB_ID_FACTORY.randomId()));
- }
-
- Stream<Arguments> blobStoresCauseReturnExceptionallyFutures() {
- List<UnionBlobStore> futureThrowingUnionBlobStores = ImmutableList.of(
- UnionBlobStore.builder()
- .current(new ThrowingBlobStore())
- .legacy(new FailingBlobStore())
- .build(),
- UnionBlobStore.builder()
- .current(new FailingBlobStore())
- .legacy(new ThrowingBlobStore())
- .build(),
- UnionBlobStore.builder()
- .current(new FailingBlobStore())
- .legacy(new FailingBlobStore())
- .build());
-
- return blobStoreOperationsReturnFutures()
- .flatMap(blobStoreFunction -> futureThrowingUnionBlobStores
- .stream()
- .map(blobStore -> Arguments.of(blobStore, blobStoreFunction)));
- }
-
- Stream<Arguments> blobStoresCauseThrowExceptions() {
- UnionBlobStore throwingUnionBlobStore = UnionBlobStore.builder()
- .current(new ThrowingBlobStore())
- .legacy(new ThrowingBlobStore())
- .build();
-
- return StreamUtils.flatten(
- blobStoreOperationsReturnFutures()
- .map(blobStoreFunction -> Arguments.of(throwingUnionBlobStore, blobStoreFunction)),
- blobStoreOperationsNotReturnFutures()
- .map(blobStoreFunction -> Arguments.of(throwingUnionBlobStore, blobStoreFunction)));
- }
-
- @ParameterizedTest
- @MethodSource("blobStoresCauseThrowExceptions")
- void operationShouldThrow(UnionBlobStore blobStoreThrowsException,
- Function<UnionBlobStore, Mono<?>> blobStoreOperation) {
- assertThatThrownBy(() -> blobStoreOperation.apply(blobStoreThrowsException).block())
- .isInstanceOf(RuntimeException.class);
- }
-
- @ParameterizedTest
- @MethodSource("blobStoresCauseReturnExceptionallyFutures")
- void operationShouldReturnExceptionallyFuture(UnionBlobStore blobStoreReturnsExceptionallyFuture,
- Function<UnionBlobStore, Mono<?>> blobStoreOperation) {
- Mono<?> mono = blobStoreOperation.apply(blobStoreReturnsExceptionallyFuture);
- assertThatThrownBy(mono::block).isInstanceOf(RuntimeException.class);
- }
- }
-
- @Test
- void readShouldReturnFromCurrentWhenAvailable() {
- BlobId blobId = currentBlobStore.save(currentBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- }
-
- @Test
- void readShouldReturnFromLegacyWhenCurrentNotAvailable() {
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(unionBlobStore.read(unionBlobStore.getDefaultBucketName(), blobId))
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- }
-
- @Test
- void readBytesShouldReturnFromCurrentWhenAvailable() {
- BlobId blobId = currentBlobStore.save(currentBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(unionBlobStore.readBytes(currentBlobStore.getDefaultBucketName(), blobId).block())
- .isEqualTo(BLOB_CONTENT);
- }
-
- @Test
- void readBytesShouldReturnFromLegacyWhenCurrentNotAvailable() {
- BlobId blobId = legacyBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(unionBlobStore.readBytes(unionBlobStore.getDefaultBucketName(), blobId).block())
- .isEqualTo(BLOB_CONTENT);
- }
-
- @Test
- void saveShouldWriteToCurrent() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThat(currentBlobStore.readBytes(currentBlobStore.getDefaultBucketName(), blobId).block())
- .isEqualTo(BLOB_CONTENT);
- }
-
- @Test
- void saveShouldNotWriteToLegacy() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), BLOB_CONTENT, LowCost).block();
-
- assertThatThrownBy(() -> legacyBlobStore.readBytes(legacyBlobStore.getDefaultBucketName(), blobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void saveStringShouldWriteToCurrent() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), STRING_CONTENT, LowCost).block();
-
- assertThat(currentBlobStore.readBytes(currentBlobStore.getDefaultBucketName(), blobId).block())
- .isEqualTo(BLOB_CONTENT);
- }
-
- @Test
- void saveStringShouldNotWriteToLegacy() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), STRING_CONTENT, LowCost).block();
-
- assertThatThrownBy(() -> legacyBlobStore.readBytes(legacyBlobStore.getDefaultBucketName(), blobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void saveInputStreamShouldWriteToCurrent() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost).block();
-
- assertThat(currentBlobStore.readBytes(currentBlobStore.getDefaultBucketName(), blobId).block())
- .isEqualTo(BLOB_CONTENT);
- }
-
- @Test
- void saveInputStreamShouldNotWriteToLegacy() {
- BlobId blobId = unionBlobStore.save(unionBlobStore.getDefaultBucketName(), new ByteArrayInputStream(BLOB_CONTENT), LowCost).block();
-
- assertThatThrownBy(() -> legacyBlobStore.readBytes(legacyBlobStore.getDefaultBucketName(), blobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void streamHasContentShouldReturnTrueWhenStreamHasContent() throws Exception {
- PushbackInputStream pushBackIS = new PushbackInputStream(new ByteArrayInputStream(BLOB_CONTENT));
-
- assertThat(unionBlobStore.streamHasContent(pushBackIS))
- .isTrue();
- }
-
- @Test
- void streamHasContentShouldReturnFalseWhenStreamHasNoContent() throws Exception {
- PushbackInputStream pushBackIS = new PushbackInputStream(new ByteArrayInputStream(new byte[0]));
-
- assertThat(unionBlobStore.streamHasContent(pushBackIS))
- .isFalse();
- }
-
- @Test
- void streamHasContentShouldNotThrowWhenStreamHasNoContent() {
- PushbackInputStream pushBackIS = new PushbackInputStream(new ByteArrayInputStream(new byte[0]));
-
- assertThatCode(() -> unionBlobStore.streamHasContent(pushBackIS))
- .doesNotThrowAnyException();
- }
-
- @Test
- void streamHasContentShouldNotDrainPushBackStreamContent() throws Exception {
- PushbackInputStream pushBackIS = new PushbackInputStream(new ByteArrayInputStream(BLOB_CONTENT));
- unionBlobStore.streamHasContent(pushBackIS);
-
- assertThat(pushBackIS)
- .hasSameContentAs(new ByteArrayInputStream(BLOB_CONTENT));
- }
-
- @Test
- void streamHasContentShouldKeepStreamEmptyWhenStreamIsEmpty() throws Exception {
- PushbackInputStream pushBackIS = new PushbackInputStream(new ByteArrayInputStream(new byte[0]));
- unionBlobStore.streamHasContent(pushBackIS);
-
- assertThat(pushBackIS)
- .hasSameContentAs(new ByteArrayInputStream(new byte[0]));
- }
-
- @Test
- void deleteBucketShouldDeleteBothCurrentAndLegacyBuckets() {
- BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
- BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
-
- unionBlobStore.deleteBucket(BucketName.DEFAULT).block();
-
- assertThatThrownBy(() -> legacyBlobStore.readBytes(BucketName.DEFAULT, legacyBlobId).block())
- .isInstanceOf(ObjectStoreException.class);
- assertThatThrownBy(() -> currentBlobStore.readBytes(BucketName.DEFAULT, currentBlobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void deleteBucketShouldDeleteCurrentBucketEvenWhenLegacyDoesNotExist() {
- BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
-
- unionBlobStore.deleteBucket(BucketName.DEFAULT).block();
-
- assertThatThrownBy(() -> currentBlobStore.readBytes(BucketName.DEFAULT, currentBlobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void deleteBucketShouldDeleteLegacyBucketEvenWhenCurrentDoesNotExist() {
- BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
-
- unionBlobStore.deleteBucket(BucketName.DEFAULT).block();
-
- assertThatThrownBy(() -> legacyBlobStore.readBytes(BucketName.DEFAULT, legacyBlobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void deleteBucketShouldNotThrowWhenCurrentAndLegacyBucketsDoNotExist() {
- assertThatCode(() -> unionBlobStore.deleteBucket(BucketName.DEFAULT).block())
- .doesNotThrowAnyException();
- }
-
- @Test
- void getDefaultBucketNameShouldThrowWhenBlobStoreDontShareTheSameDefaultBucketName() {
- currentBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY, BucketName.of("current"));
- legacyBlobStore = new MemoryBlobStore(BLOB_ID_FACTORY, BucketName.of("legacy"));
- unionBlobStore = UnionBlobStore.builder()
- .current(currentBlobStore)
- .legacy(legacyBlobStore)
- .build();
-
- assertThatThrownBy(() -> unionBlobStore.getDefaultBucketName())
- .isInstanceOf(IllegalStateException.class);
- }
-
- @Test
- void deleteShouldDeleteBothCurrentAndLegacyBlob() {
- BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
- BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
-
- unionBlobStore.delete(BucketName.DEFAULT, currentBlobId).block();
-
- assertThatThrownBy(() -> legacyBlobStore.readBytes(BucketName.DEFAULT, legacyBlobId).block())
- .isInstanceOf(ObjectStoreException.class);
- assertThatThrownBy(() -> currentBlobStore.readBytes(BucketName.DEFAULT, currentBlobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void deleteShouldDeleteCurrentBlobEvenWhenLegacyDoesNotExist() {
- BlobId currentBlobId = currentBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
-
- unionBlobStore.delete(BucketName.DEFAULT, currentBlobId).block();
-
- assertThatThrownBy(() -> currentBlobStore.readBytes(BucketName.DEFAULT, currentBlobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void deleteShouldDeleteLegacyBlobEvenWhenCurrentDoesNotExist() {
- BlobId legacyBlobId = legacyBlobStore.save(BucketName.DEFAULT, BLOB_CONTENT, LowCost).block();
-
- unionBlobStore.delete(BucketName.DEFAULT, legacyBlobId).block();
-
- assertThatThrownBy(() -> legacyBlobStore.readBytes(BucketName.DEFAULT, legacyBlobId).block())
- .isInstanceOf(ObjectStoreException.class);
- }
-
- @Test
- void deleteShouldNotThrowWhenCurrentAndLegacyBlobsDoNotExist() {
- assertThatCode(() -> unionBlobStore.delete(BucketName.DEFAULT, blobIdFactory().randomId()).block())
- .doesNotThrowAnyException();
- }
-}
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingConfiguration.java b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingConfiguration.java
index bde669c..84bc7fb 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingConfiguration.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingConfiguration.java
@@ -34,7 +34,7 @@ public class BlobStoreChoosingConfiguration {
public enum BlobStoreImplName {
CASSANDRA("cassandra"),
OBJECTSTORAGE("objectstorage"),
- UNION("union");
+ HYBRID("hybrid");
static String supportedImplNames() {
return Stream.of(BlobStoreImplName.values())
@@ -82,8 +82,8 @@ public class BlobStoreChoosingConfiguration {
return new BlobStoreChoosingConfiguration(BlobStoreImplName.OBJECTSTORAGE);
}
- public static BlobStoreChoosingConfiguration union() {
- return new BlobStoreChoosingConfiguration(BlobStoreImplName.UNION);
+ public static BlobStoreChoosingConfiguration hybrid() {
+ return new BlobStoreChoosingConfiguration(BlobStoreImplName.HYBRID);
}
private final BlobStoreImplName implementation;
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
index 7011d1e..2ab354d 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/blobstore/BlobStoreChoosingModule.java
@@ -33,7 +33,7 @@ import org.apache.james.blob.api.MetricableBlobStore;
import org.apache.james.blob.cassandra.CassandraBlobModule;
import org.apache.james.blob.cassandra.CassandraBlobStore;
import org.apache.james.blob.objectstorage.ObjectStorageBlobStore;
-import org.apache.james.blob.union.UnionBlobStore;
+import org.apache.james.blob.union.HybridBlobStore;
import org.apache.james.modules.mailbox.ConfigurationComponent;
import org.apache.james.modules.objectstorage.ObjectStorageDependenciesModule;
import org.apache.james.utils.PropertiesProvider;
@@ -82,10 +82,10 @@ public class BlobStoreChoosingModule extends AbstractModule {
return swiftBlobStoreProvider.get();
case CASSANDRA:
return cassandraBlobStoreProvider.get();
- case UNION:
- return UnionBlobStore.builder()
- .current(swiftBlobStoreProvider.get())
- .legacy(cassandraBlobStoreProvider.get())
+ case HYBRID:
+ return HybridBlobStore.builder()
+ .lowCost(swiftBlobStoreProvider.get())
+ .highPerformance(cassandraBlobStoreProvider.get())
.build();
default:
throw new RuntimeException(String.format("can not get the right blobstore provider with configuration %s",
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingConfigurationTest.java b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingConfigurationTest.java
index a5e1eb1..2ef718d 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingConfigurationTest.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingConfigurationTest.java
@@ -31,7 +31,7 @@ class BlobStoreChoosingConfigurationTest {
private static final String OBJECT_STORAGE = "objectstorage";
private static final String CASSANDRA = "cassandra";
- private static final String UNION = "union";
+ private static final String HYBRID = "hybrid";
@Test
void shouldMatchBeanContract() {
@@ -45,7 +45,7 @@ class BlobStoreChoosingConfigurationTest {
assertThatThrownBy(() -> BlobStoreChoosingConfiguration.from(configuration))
.isInstanceOf(IllegalStateException.class)
- .hasMessage("implementation property is missing please use one of supported values in: cassandra, objectstorage, union");
+ .hasMessage("implementation property is missing please use one of supported values in: cassandra, objectstorage, hybrid");
}
@Test
@@ -55,7 +55,7 @@ class BlobStoreChoosingConfigurationTest {
assertThatThrownBy(() -> BlobStoreChoosingConfiguration.from(configuration))
.isInstanceOf(IllegalStateException.class)
- .hasMessage("implementation property is missing please use one of supported values in: cassandra, objectstorage, union");
+ .hasMessage("implementation property is missing please use one of supported values in: cassandra, objectstorage, hybrid");
}
@Test
@@ -65,7 +65,7 @@ class BlobStoreChoosingConfigurationTest {
assertThatThrownBy(() -> BlobStoreChoosingConfiguration.from(configuration))
.isInstanceOf(IllegalStateException.class)
- .hasMessage("implementation property is missing please use one of supported values in: cassandra, objectstorage, union");
+ .hasMessage("implementation property is missing please use one of supported values in: cassandra, objectstorage, hybrid");
}
@Test
@@ -75,7 +75,7 @@ class BlobStoreChoosingConfigurationTest {
assertThatThrownBy(() -> BlobStoreChoosingConfiguration.from(configuration))
.isInstanceOf(IllegalArgumentException.class)
- .hasMessage("un_supported is not a valid name of BlobStores, please use one of supported values in: cassandra, objectstorage, union");
+ .hasMessage("un_supported is not a valid name of BlobStores, please use one of supported values in: cassandra, objectstorage, hybrid");
}
@Test
@@ -93,13 +93,13 @@ class BlobStoreChoosingConfigurationTest {
@Test
void fromShouldReturnConfigurationWhenBlobStoreImplIsUnion() {
PropertiesConfiguration configuration = new PropertiesConfiguration();
- configuration.addProperty("implementation", UNION);
+ configuration.addProperty("implementation", HYBRID);
assertThat(
BlobStoreChoosingConfiguration.from(configuration)
.getImplementation()
.getName())
- .isEqualTo(UNION);
+ .isEqualTo(HYBRID);
}
@Test
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingModuleTest.java b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingModuleTest.java
index ce4354b..4e04a5b 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingModuleTest.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/modules/blobstore/BlobStoreChoosingModuleTest.java
@@ -27,7 +27,7 @@ import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.james.FakePropertiesProvider;
import org.apache.james.blob.cassandra.CassandraBlobStore;
import org.apache.james.blob.objectstorage.ObjectStorageBlobStore;
-import org.apache.james.blob.union.UnionBlobStore;
+import org.apache.james.blob.union.HybridBlobStore;
import org.apache.james.modules.blobstore.BlobStoreChoosingConfiguration.BlobStoreImplName;
import org.apache.james.modules.mailbox.ConfigurationComponent;
import org.junit.jupiter.api.Test;
@@ -105,16 +105,16 @@ class BlobStoreChoosingModuleTest {
}
@Test
- void provideChoosingConfigurationShouldReturnUnionConfigurationWhenConfigurationImplIsUnion() throws Exception {
+ void provideChoosingConfigurationShouldReturnHybridConfigurationWhenConfigurationImplIsHybrid() throws Exception {
BlobStoreChoosingModule module = new BlobStoreChoosingModule();
PropertiesConfiguration configuration = new PropertiesConfiguration();
- configuration.addProperty("implementation", BlobStoreImplName.UNION.getName());
+ configuration.addProperty("implementation", BlobStoreImplName.HYBRID.getName());
FakePropertiesProvider propertyProvider = FakePropertiesProvider.builder()
.register(ConfigurationComponent.NAME, configuration)
.build();
assertThat(module.provideChoosingConfiguration(propertyProvider))
- .isEqualTo(BlobStoreChoosingConfiguration.union());
+ .isEqualTo(BlobStoreChoosingConfiguration.hybrid());
}
@Test
@@ -149,11 +149,11 @@ class BlobStoreChoosingModuleTest {
}
@Test
- void provideBlobStoreShouldReturnUnionBlobStoreWhenUnionConfigured() {
+ void provideBlobStoreShouldReturnHybridBlobStoreWhenHybridConfigured() {
BlobStoreChoosingModule module = new BlobStoreChoosingModule();
- assertThat(module.provideBlobStore(BlobStoreChoosingConfiguration.union(),
+ assertThat(module.provideBlobStore(BlobStoreChoosingConfiguration.hybrid(),
CASSANDRA_BLOBSTORE_PROVIDER, OBJECT_STORAGE_BLOBSTORE_PROVIDER))
- .isInstanceOf(UnionBlobStore.class);
+ .isInstanceOf(HybridBlobStore.class);
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org