You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2024/04/01 03:00:59 UTC
(james-project) branch master updated: [Fix] AESBlobStore threading
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
The following commit(s) were added to refs/heads/master by this push:
new f35cc07e03 [Fix] AESBlobStore threading
f35cc07e03 is described below
commit f35cc07e0363685cc64430a5b7e22b8a78e28d24
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Fri Mar 29 21:41:52 2024 +0100
[Fix] AESBlobStore threading
Was blocking on the S3 driver Netty EventLoop and on the parrallel scheduler
(verified with printing thread names)
---
.../src/main/java/org/apache/james/blob/aes/AESBlobStoreDAO.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/server/blob/blob-aes/src/main/java/org/apache/james/blob/aes/AESBlobStoreDAO.java b/server/blob/blob-aes/src/main/java/org/apache/james/blob/aes/AESBlobStoreDAO.java
index ca1cbeef1d..a73495e375 100644
--- a/server/blob/blob-aes/src/main/java/org/apache/james/blob/aes/AESBlobStoreDAO.java
+++ b/server/blob/blob-aes/src/main/java/org/apache/james/blob/aes/AESBlobStoreDAO.java
@@ -108,6 +108,7 @@ public class AESBlobStoreDAO implements BlobStoreDAO {
WritableByteChannel channel = Channels.newChannel(encryptedContent);
return Flux.from(ciphertext.getContent())
+ .publishOn(Schedulers.boundedElastic())
.doOnNext(Throwing.consumer(channel::write))
.then(Mono.fromCallable(() -> {
try {
@@ -174,10 +175,10 @@ public class AESBlobStoreDAO implements BlobStoreDAO {
Preconditions.checkNotNull(blobId);
Preconditions.checkNotNull(inputStream);
- return Mono.using(
- () -> encrypt(inputStream),
+ return Mono.usingWhen(
+ Mono.fromCallable(() -> encrypt(inputStream)),
pair -> Mono.from(underlying.save(bucketName, blobId, byteSourceWithSize(pair.getLeft().asByteSource(), pair.getRight()))),
- Throwing.consumer(pair -> pair.getLeft().reset()))
+ 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));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org