You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2024/03/18 10:47:06 UTC
(james-project) branch master updated (7222019643 -> 019dabb517)
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 7222019643 JAMES-4020 Fully comply with RFC-3501 Section 6.4.5 (#2123)
new 77d195f8a5 [FIX] IMAP linearization can be simplified
new f5d1b73559 [FIX] IMAP FETCH backpressure propagate cancel
new a9c9f02ac5 [FIX] FETCH: do not reuse FetchResponseBuilder
new 019dabb517 [FIX] FETCH: flatten reactor operations
The 4 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:
.../james/imap/processor/fetch/FetchProcessor.java | 37 ++++++-----
.../netty/ImapChannelUpstreamHandler.java | 33 ++++++++--
.../james/imapserver/netty/Linearalizer.java | 71 ----------------------
.../james/imapserver/netty/NettyConstants.java | 2 +-
.../james/imapserver/netty/LinearalizerTest.java | 71 ----------------------
5 files changed, 48 insertions(+), 166 deletions(-)
delete mode 100644 server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/Linearalizer.java
delete mode 100644 server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/LinearalizerTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 03/04: [FIX] FETCH: do not reuse FetchResponseBuilder
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 a9c9f02ac5171b3d9f7351ec99e85a87d0eebbff
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Sat Mar 16 22:11:29 2024 +0100
[FIX] FETCH: do not reuse FetchResponseBuilder
---
.../apache/james/imap/processor/fetch/FetchProcessor.java | 13 ++++++-------
1 file changed, 6 insertions(+), 7 deletions(-)
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
index 83578d358b..db2c99fb90 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
@@ -227,14 +227,13 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
* {@link org.apache.james.imap.api.process.ImapProcessor.Responder}
*/
private Mono<Void> processMessageRanges(SelectedMailbox selected, MessageManager mailbox, List<MessageRange> ranges, FetchData fetch, MailboxSession mailboxSession, Responder responder, ImapSession imapSession) {
- FetchResponseBuilder builder = new FetchResponseBuilder(new EnvelopeBuilder());
FetchGroup resultToFetch = FetchDataConverter.getFetchGroup(fetch);
if (fetch.isOnlyFlags()) {
return Flux.fromIterable(consolidate(selected, ranges, fetch))
.concatMap(range -> Flux.from(mailbox.listMessagesMetadata(range, mailboxSession)))
.filter(ids -> !fetch.contains(Item.MODSEQ) || ids.getModSeq().asLong() > fetch.getChangedSince())
- .concatMap(result -> toResponse(mailbox, fetch, mailboxSession, builder, selected, result))
+ .concatMap(result -> toResponse(mailbox, fetch, mailboxSession, selected, result))
.doOnNext(responder::respond)
.then();
} else {
@@ -245,7 +244,7 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
return Flux.from(mailbox.getMessagesReactive(range, resultToFetch, mailboxSession))
.filter(ids -> !fetch.contains(Item.MODSEQ) || ids.getModSeq().asLong() > fetch.getChangedSince())
- .concatMap(result -> toResponse(mailbox, fetch, mailboxSession, builder, selected, result));
+ .concatMap(result -> toResponse(mailbox, fetch, mailboxSession, selected, result));
})
.subscribe(fetchSubscriber);
@@ -266,9 +265,9 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
return MessageRange.toRanges(filter.longStream().mapToObj(MessageUid::of).collect(ImmutableList.toImmutableList()));
}
- private Mono<FetchResponse> toResponse(MessageManager mailbox, FetchData fetch, MailboxSession mailboxSession, FetchResponseBuilder builder, SelectedMailbox selected, org.apache.james.mailbox.model.ComposedMessageIdWithMetaData result) {
+ private Mono<FetchResponse> toResponse(MessageManager mailbox, FetchData fetch, MailboxSession mailboxSession, SelectedMailbox selected, org.apache.james.mailbox.model.ComposedMessageIdWithMetaData result) {
try {
- return builder.build(fetch, result, mailbox, selected, mailboxSession);
+ return new FetchResponseBuilder(new EnvelopeBuilder()).build(fetch, result, mailbox, selected, mailboxSession);
} catch (MessageRangeException e) {
// we can't for whatever reason find the message so
// just skip it and log it to debug
@@ -285,9 +284,9 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
}
}
- private Mono<FetchResponse> toResponse(MessageManager mailbox, FetchData fetch, MailboxSession mailboxSession, FetchResponseBuilder builder, SelectedMailbox selected, MessageResult result) {
+ private Mono<FetchResponse> toResponse(MessageManager mailbox, FetchData fetch, MailboxSession mailboxSession, SelectedMailbox selected, MessageResult result) {
try {
- return builder.build(fetch, result, mailbox, selected, mailboxSession);
+ return new FetchResponseBuilder(new EnvelopeBuilder()).build(fetch, result, mailbox, selected, mailboxSession);
} catch (MessageRangeException e) {
// we can't for whatever reason find the message so
// just skip it and log it to debug
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 01/04: [FIX] IMAP linearization can be simplified
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 77d195f8a552f702370543e6f7da906c2a399aa4
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Sat Mar 16 08:27:43 2024 +0100
[FIX] IMAP linearization can be simplified
Handle the queuing and linearization within netty channel,
not into reactor.
This avoids complex signalling at the REACTOR level...
---
.../netty/ImapChannelUpstreamHandler.java | 33 ++++++++--
.../james/imapserver/netty/Linearalizer.java | 71 ----------------------
.../james/imapserver/netty/NettyConstants.java | 2 +-
.../james/imapserver/netty/LinearalizerTest.java | 71 ----------------------
4 files changed, 30 insertions(+), 147 deletions(-)
diff --git a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ImapChannelUpstreamHandler.java b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ImapChannelUpstreamHandler.java
index 934a5d4483..094d0284e3 100644
--- a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ImapChannelUpstreamHandler.java
+++ b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ImapChannelUpstreamHandler.java
@@ -29,6 +29,8 @@ import java.time.Duration;
import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.net.ssl.SSLHandshakeException;
@@ -156,6 +158,11 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
}
}
+ static class ImapLinerarizer {
+ private final AtomicBoolean isExecutingRequest = new AtomicBoolean(false);
+ private final ConcurrentLinkedQueue<Object> throttled = new ConcurrentLinkedQueue<>();
+ }
+
public static ImapChannelUpstreamHandlerBuilder builder() {
return new ImapChannelUpstreamHandlerBuilder();
}
@@ -200,7 +207,7 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
authenticationConfiguration.isPlainAuthEnabled(), sessionId,
authenticationConfiguration.getOidcSASLConfiguration());
ctx.channel().attr(IMAP_SESSION_ATTRIBUTE_KEY).set(imapsession);
- ctx.channel().attr(LINEARALIZER_ATTRIBUTE_KEY).set(new Linearalizer());
+ ctx.channel().attr(LINEARIZER_ATTRIBUTE_KEY).set(new ImapLinerarizer());
MDCBuilder boundMDC = IMAPMDCContext.boundMDC(ctx)
.addToContext(MDCBuilder.SESSION_ID, sessionId.asString());
imapsession.setAttribute(MDC_KEY, boundMDC);
@@ -375,8 +382,17 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
public void channelRead(ChannelHandlerContext ctx, Object msg) {
imapCommandsMetric.increment();
ImapSession session = ctx.channel().attr(IMAP_SESSION_ATTRIBUTE_KEY).get();
- Linearalizer linearalizer = ctx.channel().attr(LINEARALIZER_ATTRIBUTE_KEY).get();
Attribute<Disposable> disposableAttribute = ctx.channel().attr(REQUEST_IN_FLIGHT_ATTRIBUTE_KEY);
+
+ ImapLinerarizer linearalizer = ctx.channel().attr(LINEARIZER_ATTRIBUTE_KEY).get();
+ synchronized (linearalizer) {
+ if (linearalizer.isExecutingRequest.get()) {
+ linearalizer.throttled.add(msg);
+ return;
+ }
+ linearalizer.isExecutingRequest.set(true);
+ }
+
ChannelImapResponseWriter writer = new ChannelImapResponseWriter(ctx.channel());
ImapResponseComposerImpl response = new ImapResponseComposerImpl(writer);
writer.setFlushCallback(response::flush);
@@ -384,8 +400,7 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
beforeIDLEUponProcessing(ctx);
ResponseEncoder responseEncoder = new ResponseEncoder(encoder, response);
- Disposable disposable = reactiveThrottler.throttle(
- linearalizer.execute(processor.processReactive(message, responseEncoder, session))
+ Disposable disposable = reactiveThrottler.throttle(processor.processReactive(message, responseEncoder, session)
.doOnEach(Throwing.consumer(signal -> {
if (session.getState() == ImapSessionState.LOGOUT) {
// Make sure we close the channel after all the buffers were flushed out
@@ -407,6 +422,11 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
ctx.fireExceptionCaught(failure);
}
}
+ Object waitingMessage;
+ synchronized (linearalizer) {
+ linearalizer.isExecutingRequest.set(false);
+ waitingMessage = linearalizer.throttled.poll();
+ }
if (signal.isOnComplete() || signal.isOnError()) {
afterIDLEUponProcessing(ctx);
}
@@ -416,6 +436,11 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
disposableAttribute.set(null);
response.flush();
ctx.fireChannelReadComplete();
+ if (signal.isOnComplete() || signal.isOnError()) {
+ if (waitingMessage != null && signal.isOnComplete()) {
+ channelRead(ctx, waitingMessage);
+ }
+ }
}))
.contextWrite(ReactorUtils.context("imap", mdc(session))), message)
// Manage throttling errors
diff --git a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/Linearalizer.java b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/Linearalizer.java
deleted file mode 100644
index 05cf3f9e20..0000000000
--- a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/Linearalizer.java
+++ /dev/null
@@ -1,71 +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.imapserver.netty;
-
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.reactivestreams.Publisher;
-
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.Sinks;
-
-public class Linearalizer {
- private final ReentrantLock lock = new ReentrantLock();
- private boolean inFlight = false;
- private final Queue<Publisher<Void>> queue = new ConcurrentLinkedQueue<>();
-
- public Mono<Void> execute(Publisher<Void> task) {
- lock.lock();
- try {
- if (!inFlight) {
- inFlight = true;
- return Mono.from(task)
- .doFinally(any -> onRequestDone());
- } else {
- // Queue the request for later
- Sinks.One<Void> one = Sinks.one();
- queue.add(Mono.from(task)
- .then(Mono.fromRunnable(() -> one.emitEmpty(Sinks.EmitFailureHandler.FAIL_FAST))));
- // Let the caller await task completion
- return one.asMono();
- }
- } finally {
- lock.unlock();
- }
- }
-
- private void onRequestDone() {
- lock.lock();
- try {
- Publisher<Void> throttled = queue.poll();
- if (throttled != null) {
- Mono.from(throttled)
- .doFinally(any -> onRequestDone())
- .subscribe();
- } else {
- inFlight = false;
- }
- } finally {
- lock.unlock();
- }
- }
-}
diff --git a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/NettyConstants.java b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/NettyConstants.java
index 2af87cd415..2aa10f32f4 100644
--- a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/NettyConstants.java
+++ b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/NettyConstants.java
@@ -41,8 +41,8 @@ public interface NettyConstants {
String HEARTBEAT_HANDLER = "heartbeatHandler";
AttributeKey<ImapSession> IMAP_SESSION_ATTRIBUTE_KEY = AttributeKey.valueOf("ImapSession");
- AttributeKey<Linearalizer> LINEARALIZER_ATTRIBUTE_KEY = AttributeKey.valueOf("Linearalizer");
AttributeKey<Disposable> REQUEST_IN_FLIGHT_ATTRIBUTE_KEY = AttributeKey.valueOf("requestInFlight");
+ AttributeKey<ImapChannelUpstreamHandler.ImapLinerarizer> LINEARIZER_ATTRIBUTE_KEY = AttributeKey.valueOf("linearizer");
AttributeKey<Runnable> BACKPRESSURE_CALLBACK = AttributeKey.valueOf("BACKPRESSURE_CALLBACK");
AttributeKey<Map<String, Object>> FRAME_DECODE_ATTACHMENT_ATTRIBUTE_KEY = AttributeKey.valueOf("FrameDecoderMap");
diff --git a/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/LinearalizerTest.java b/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/LinearalizerTest.java
deleted file mode 100644
index 9d745a6188..0000000000
--- a/server/protocols/protocols-imap4/src/test/java/org/apache/james/imapserver/netty/LinearalizerTest.java
+++ /dev/null
@@ -1,71 +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.imapserver.netty;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-import java.time.Duration;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.junit.jupiter.api.Test;
-import org.testcontainers.shaded.org.awaitility.Awaitility;
-
-import reactor.core.publisher.Mono;
-
-class LinearalizerTest {
- @Test
- void shouldExecuteSubmittedTasks() {
- Linearalizer testee = new Linearalizer();
-
- // When I submit a task
- AtomicBoolean executed = new AtomicBoolean(false);
- Mono.from(testee.execute(Mono.delay(Duration.ofMillis(50)).then(Mono.fromRunnable(() -> executed.getAndSet(true))))).block();
-
- // Then that task is executed
- assertThat(executed.get()).isTrue();
- }
-
- @Test
- void shouldNotExecuteQueuedTasksLogicRightAway() {
- Linearalizer testee = new Linearalizer();
-
- // When I submit 2 tasks task
- AtomicBoolean executed = new AtomicBoolean(false);
- Mono.from(testee.execute(Mono.delay(Duration.ofMillis(200)).then())).subscribe();
- Mono.from(testee.execute(Mono.fromRunnable(() -> executed.getAndSet(true)))).subscribe();
-
- // Then the second task is not executed staight away
- assertThat(executed.get()).isFalse();
- }
-
- @Test
- void shouldEventuallyExecuteQueuedTasks() {
- Linearalizer testee = new Linearalizer();
-
- // When I submit 2 tasks task
- AtomicBoolean executed = new AtomicBoolean(false);
- Mono.from(testee.execute(Mono.delay(Duration.ofMillis(200)).then())).subscribe();
- Mono.from(testee.execute(Mono.fromRunnable(() -> executed.getAndSet(true)))).subscribe();
-
- // Then that task is eventually executed
- Awaitility.await().atMost(Duration.ofSeconds(10))
- .untilAsserted(() -> assertThat(executed.get()).isTrue());
- }
-}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 02/04: [FIX] IMAP FETCH backpressure propagate cancel
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 f5d1b7355927cd4b82e4efd09d0e5fd320b1941d
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Sat Mar 16 08:44:44 2024 +0100
[FIX] IMAP FETCH backpressure propagate cancel
---
.../james/imap/processor/fetch/FetchProcessor.java | 21 ++++++++++++---------
1 file changed, 12 insertions(+), 9 deletions(-)
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
index 2caf9a0e7d..83578d358b 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
@@ -121,7 +121,11 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
}
public Mono<Void> completionMono() {
- return sink.asMono();
+ return sink.asMono()
+ .doOnCancel(() -> {
+ Optional.ofNullable(subscription.get()).ifPresent(Subscription::cancel);
+ subscription.set(null);
+ });
}
}
@@ -234,19 +238,18 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
.doOnNext(responder::respond)
.then();
} else {
- return Flux.fromIterable(consolidate(selected, ranges, fetch))
+ FetchSubscriber fetchSubscriber = new FetchSubscriber(imapSession, responder);
+ Flux.fromIterable(consolidate(selected, ranges, fetch))
.concatMap(range -> {
- FetchSubscriber fetchSubscriber = new FetchSubscriber(imapSession, responder);
auditTrail(mailbox, mailboxSession, resultToFetch, range);
- Flux.from(mailbox.getMessagesReactive(range, resultToFetch, mailboxSession))
+ return Flux.from(mailbox.getMessagesReactive(range, resultToFetch, mailboxSession))
.filter(ids -> !fetch.contains(Item.MODSEQ) || ids.getModSeq().asLong() > fetch.getChangedSince())
- .concatMap(result -> toResponse(mailbox, fetch, mailboxSession, builder, selected, result))
- .subscribe(fetchSubscriber);
-
- return fetchSubscriber.completionMono();
+ .concatMap(result -> toResponse(mailbox, fetch, mailboxSession, builder, selected, result));
})
- .then();
+ .subscribe(fetchSubscriber);
+
+ return fetchSubscriber.completionMono();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 04/04: [FIX] FETCH: flatten reactor operations
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 019dabb51789fb1fccc5e342aec77a34b48478e3
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Sat Mar 16 22:12:09 2024 +0100
[FIX] FETCH: flatten reactor operations
---
.../org/apache/james/imap/processor/fetch/FetchProcessor.java | 11 ++++-------
1 file changed, 4 insertions(+), 7 deletions(-)
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
index db2c99fb90..7878c19162 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/FetchProcessor.java
@@ -239,13 +239,10 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
} else {
FetchSubscriber fetchSubscriber = new FetchSubscriber(imapSession, responder);
Flux.fromIterable(consolidate(selected, ranges, fetch))
- .concatMap(range -> {
- auditTrail(mailbox, mailboxSession, resultToFetch, range);
-
- return Flux.from(mailbox.getMessagesReactive(range, resultToFetch, mailboxSession))
- .filter(ids -> !fetch.contains(Item.MODSEQ) || ids.getModSeq().asLong() > fetch.getChangedSince())
- .concatMap(result -> toResponse(mailbox, fetch, mailboxSession, selected, result));
- })
+ .doOnNext(range -> auditTrail(mailbox, mailboxSession, resultToFetch, range))
+ .concatMap(range -> Flux.from(mailbox.getMessagesReactive(range, resultToFetch, mailboxSession)))
+ .filter(ids -> !fetch.contains(Item.MODSEQ) || ids.getModSeq().asLong() > fetch.getChangedSince())
+ .concatMap(result -> toResponse(mailbox, fetch, mailboxSession, selected, result))
.subscribe(fetchSubscriber);
return fetchSubscriber.completionMono();
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org