Skip to content

JAMES-4182 Proposal for changes in BlobStoreDAO interface #2960

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chibenwa merged 17 commits into from
Apr 13, 2026
Merged

JAMES-4182 Proposal for changes in BlobStoreDAO interface #2960

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
8a43b7c
Remove unused ReactiveByteSource
chibenwa Mar 3, 2026
7cc2509
BlobStoreDAO: saveBlob
chibenwa Mar 3, 2026
68c6f22
BlobStoreDAO: readBytesBlob
chibenwa Mar 3, 2026
77f5a2f
BlobStoreDAO: readBlobReactive
chibenwa Mar 3, 2026
cbaaeb8
BlobStoreDAO: readBlob
chibenwa Mar 3, 2026
f3432ef
BlobStoreDAO: remove blob from method names
chibenwa Mar 3, 2026
3444feb
TODOs
chibenwa Mar 3, 2026
88c77dd
Fix existing contracts
chibenwa Mar 19, 2026
be67d46
Adapt metadata model following review comments
chibenwa Mar 19, 2026
3addef6
Memory: store Blob metadata
chibenwa Mar 19, 2026
3451c8c
JAMES-4182 Fix compilation
quantranhong1999 Apr 7, 2026
ded5e50
JAMES-4182 Implement equals and hashCode for BytesBlob properly
quantranhong1999 Apr 8, 2026
33826d5
JAMES-4182 Fix S3BlobStoreDAOTest::readBytesShouldFallbackToDefinedBu…
quantranhong1999 Apr 8, 2026
40e6383
JAMES-4182 AESBlobStoreDAO should respect metadata information from u…
quantranhong1999 Apr 8, 2026
6110c88
JAMES-4182 Adapt review suggestions
quantranhong1999 Apr 8, 2026
2fd8fec
JAMES-4182 [FIX] InMemoryUploadRepository: do not eagerly calculate i…
quantranhong1999 Apr 9, 2026
a3d5a9d
JAMES-4182 BlobMetadataName should be case-insensitive
quantranhong1999 Apr 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 27 additions & 20 deletions server/blob/blob-aes/src/main/java/org/apache/james/blob/aes/AESBlobStoreDAO.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,59 +93,67 @@ public InputStream decrypt(InputStream ciphertext) throws IOException {
}

@Override
public InputStream read(BucketName bucketName, BlobId blobId) throws ObjectStoreIOException, ObjectNotFoundException {
public InputStreamBlob read(BucketName bucketName, BlobId blobId) throws ObjectStoreIOException, ObjectNotFoundException {
try {
return decrypt(underlying.read(bucketName, blobId));
InputStreamBlob underlyingBlob = underlying.read(bucketName, blobId);
return InputStreamBlob.of(decrypt(underlyingBlob.payload()), underlyingBlob.metadata());
} catch (IOException e) {
throw new ObjectStoreIOException("Error reading blob " + blobId.asString(), e);
}
}

@Override
public Publisher<InputStream> readReactive(BucketName bucketName, BlobId blobId) {
public Publisher<InputStreamBlob> readReactive(BucketName bucketName, BlobId blobId) {
return Mono.from(underlying.readReactive(bucketName, blobId))
.map(Throwing.function(this::decrypt));
.map(Throwing.function(inputStreamBlob -> InputStreamBlob.of(decrypt(inputStreamBlob.payload()), inputStreamBlob.metadata())));
}

@Override
public Publisher<byte[]> readBytes(BucketName bucketName, BlobId blobId) {
public Publisher<BytesBlob> readBytes(BucketName bucketName, BlobId blobId) {
return Mono.from(underlying.readBytes(bucketName, blobId))
.map(Throwing.function(bytes -> {
InputStream inputStream = decrypt(new ByteArrayInputStream(bytes));
.map(Throwing.function(bytesBlob -> {
InputStream inputStream = decrypt(new ByteArrayInputStream(bytesBlob.payload()));
try (UnsynchronizedByteArrayOutputStream outputStream = UnsynchronizedByteArrayOutputStream.builder()
.setBufferSize(bytes.length + PBKDF2StreamingAeadFactory.SEGMENT_SIZE)
.setBufferSize(bytesBlob.payload().length + PBKDF2StreamingAeadFactory.SEGMENT_SIZE)
.get()) {
IOUtils.copy(inputStream, outputStream);
return outputStream.toByteArray();
return BytesBlob.of(outputStream.toByteArray(), bytesBlob.metadata());
}
}));
}

@Override
public Publisher<Void> save(BucketName bucketName, BlobId blobId, byte[] data) {
public Publisher<Void> save(BucketName bucketName, BlobId blobId, Blob blob) {
return switch (blob) {
case BytesBlob bytesBlob -> save(bucketName, blobId, bytesBlob.payload(), bytesBlob.metadata());
case InputStreamBlob inputStreamBlob -> save(bucketName, blobId, inputStreamBlob.payload(), inputStreamBlob.metadata());
case ByteSourceBlob byteSourceBlob -> save(bucketName, blobId, byteSourceBlob.payload(), byteSourceBlob.metadata());
};
}

private Publisher<Void> save(BucketName bucketName, BlobId blobId, byte[] data, BlobMetadata metadata) {
Preconditions.checkNotNull(bucketName);
Preconditions.checkNotNull(blobId);
Preconditions.checkNotNull(data);

return save(bucketName, blobId, new ByteArrayInputStream(data));
return save(bucketName, blobId, new ByteArrayInputStream(data), metadata);
}

@Override
public Publisher<Void> save(BucketName bucketName, BlobId blobId, InputStream inputStream) {
private Publisher<Void> save(BucketName bucketName, BlobId blobId, InputStream inputStream, BlobMetadata metadata) {
Preconditions.checkNotNull(bucketName);
Preconditions.checkNotNull(blobId);
Preconditions.checkNotNull(inputStream);

return Mono.usingWhen(
Mono.fromCallable(() -> encrypt(inputStream)),
pair -> Mono.from(underlying.save(bucketName, blobId, byteSourceWithSize(pair.getLeft().asByteSource(), pair.getRight()))),
pair -> Mono.from(underlying.save(bucketName, blobId, byteSourceWithSize(pair.getLeft().asByteSource(), pair.getRight(), metadata))),
Throwing.function(pair -> Mono.fromRunnable(Throwing.runnable(pair.getLeft()::reset)).subscribeOn(Schedulers.boundedElastic())))
.subscribeOn(Schedulers.boundedElastic())
.onErrorMap(e -> new ObjectStoreIOException("Exception occurred while saving bytearray", e));
}

private ByteSource byteSourceWithSize(ByteSource byteSource, long size) {
return new ByteSource() {
private ByteSourceBlob byteSourceWithSize(ByteSource byteSource, long size, BlobMetadata metadata) {
return ByteSourceBlob.of(new ByteSource() {
@Override
public InputStream openStream() throws IOException {
return byteSource.openStream();
Expand All @@ -160,17 +168,16 @@ public com.google.common.base.Optional<Long> sizeIfKnown() {
public long size() {
return size;
}
};
}, metadata);
}

@Override
public Publisher<Void> save(BucketName bucketName, BlobId blobId, ByteSource content) {
private Publisher<Void> save(BucketName bucketName, BlobId blobId, ByteSource content, BlobMetadata metadata) {
Preconditions.checkNotNull(bucketName);
Preconditions.checkNotNull(blobId);
Preconditions.checkNotNull(content);

return Mono.using(content::openStream,
in -> Mono.from(save(bucketName, blobId, in)),
in -> Mono.from(save(bucketName, blobId, in, metadata)),
Throwing.consumer(InputStream::close))
.subscribeOn(Schedulers.boundedElastic());
}
Expand Down
17 changes: 7 additions & 10 deletions server/blob/blob-aes/src/test/java/org/apache/james/blob/aes/AESBlobStoreDAOTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,20 +24,17 @@
import static org.apache.james.blob.api.BlobStoreDAOFixture.TEST_BUCKET_NAME;
import static org.assertj.core.api.Assertions.assertThat;

import java.io.ByteArrayInputStream;

import org.apache.james.blob.api.BlobStoreDAO;
import org.apache.james.blob.api.BlobStoreDAOContract;
import org.apache.james.blob.api.MetadataAwareBlobStoreDAOContract;
import org.apache.james.blob.memory.MemoryBlobStoreDAO;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;

import com.google.common.io.ByteSource;

import reactor.core.publisher.Mono;

class AESBlobStoreDAOTest implements BlobStoreDAOContract {
class AESBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract {
private static final String SAMPLE_SALT = "c603a7327ee3dcbc031d8d34b1096c605feca5e1";
private static final CryptoConfig CRYPTO_CONFIG = CryptoConfig.builder()
.salt(SAMPLE_SALT)
Expand All @@ -62,25 +59,25 @@ public BlobStoreDAO testee() {
void underlyingDataShouldBeEncrypted() {
Mono.from(testee.save(TEST_BUCKET_NAME, TEST_BLOB_ID, SHORT_BYTEARRAY)).block();

byte[] bytes = Mono.from(underlying.readBytes(TEST_BUCKET_NAME, TEST_BLOB_ID)).block();
byte[] bytes = Mono.from(underlying.readBytes(TEST_BUCKET_NAME, TEST_BLOB_ID)).block().payload();

assertThat(bytes).isNotEqualTo(SHORT_BYTEARRAY);
}

@Test
void underlyingDataShouldBeEncryptedWhenUsingStream() {
Mono.from(testee.save(TEST_BUCKET_NAME, TEST_BLOB_ID, new ByteArrayInputStream(SHORT_BYTEARRAY))).block();
Mono.from(testee.save(TEST_BUCKET_NAME, TEST_BLOB_ID, SHORT_BYTEARRAY.asInputStream())).block();

byte[] bytes = Mono.from(underlying.readBytes(TEST_BUCKET_NAME, TEST_BLOB_ID)).block();
byte[] bytes = Mono.from(underlying.readBytes(TEST_BUCKET_NAME, TEST_BLOB_ID)).block().payload();

assertThat(bytes).isNotEqualTo(SHORT_BYTEARRAY);
}

@Test
void underlyingDataShouldBeEncryptedWhenUsingByteSource() {
Mono.from(testee.save(TEST_BUCKET_NAME, TEST_BLOB_ID, ByteSource.wrap(SHORT_BYTEARRAY))).block();
Mono.from(testee.save(TEST_BUCKET_NAME, TEST_BLOB_ID, SHORT_BYTEARRAY.asByteSource())).block();

byte[] bytes = Mono.from(underlying.readBytes(TEST_BUCKET_NAME, TEST_BLOB_ID)).block();
byte[] bytes = Mono.from(underlying.readBytes(TEST_BUCKET_NAME, TEST_BLOB_ID)).block().payload();

assertThat(bytes).isNotEqualTo(SHORT_BYTEARRAY);
}
Expand Down
Loading