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/03/29 03:05:14 UTC
(james-project) branch master updated (1bc01eb36e -> 8155100e36)
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
from 1bc01eb36e [FIX] IMAP STATUS should better handle MailboxNotFoundException (#2158)
new d2502daef6 JAMES-3977 Add bytes preservation in manager and mapper contracts
new 487120bf77 JAMES-3977 Improve IMAP FETCH concurrency control
new 7df572d655 JAMES-3977 Specify publishOn prefetch
new 8155100e36 JAMES-3977 IMAP log should embed username where possible
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:
.../apache/james/mailbox/MailboxManagerTest.java | 7 +
.../cassandra/mail/CassandraMessageMetadata.java | 10 ++
.../cassandra/CassandraMailboxManagerTest.java | 31 +++++
.../inmemory/mail/InMemoryMapperProvider.java | 3 +-
.../james/mailbox/store/StoreMessageIdManager.java | 4 +-
.../james/mailbox/store/StoreMessageManager.java | 4 +-
.../store/mail/model/DelegatingMailboxMessage.java | 5 +
.../store/mail/model/impl/SimpleMessage.java | 20 ++-
.../mailbox/store/mail/model/MapperProvider.java | 4 +-
.../store/mail/model/MessageMapperTest.java | 62 ++++++++-
.../james/imap/processor/fetch/FetchProcessor.java | 25 ++--
.../apache/james/WithCassandraBlobStoreTest.java | 149 ++++++++++++++++++++-
.../netty/ImapChannelUpstreamHandler.java | 17 ++-
13 files changed, 312 insertions(+), 29 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 03/04: JAMES-3977 Specify publishOn prefetch
Posted by rc...@apache.org.
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
commit 7df572d655ebcc91aad3164eca4d8ec3bc3f96fc
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Wed Mar 27 17:22:07 2024 +0100
JAMES-3977 Specify publishOn prefetch
Failure to do so results in a prefetch of 256 items.
This is bad for big messages: 256 * 20MB -> 5GB memory used.
Also delaying errors is problematic.
---
.../cassandra/CassandraMailboxManagerTest.java | 31 +++++
.../james/mailbox/store/StoreMessageIdManager.java | 4 +-
.../james/mailbox/store/StoreMessageManager.java | 4 +-
.../apache/james/WithCassandraBlobStoreTest.java | 149 ++++++++++++++++++++-
4 files changed, 185 insertions(+), 3 deletions(-)
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java
index 159a79eeb2..b24b4b8f33 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java
@@ -36,6 +36,7 @@ import java.util.stream.IntStream;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.backends.cassandra.CassandraCluster;
import org.apache.james.backends.cassandra.CassandraClusterExtension;
+import org.apache.james.backends.cassandra.StatementRecorder;
import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
import org.apache.james.blob.api.BlobStore;
import org.apache.james.blob.api.HashBlobId;
@@ -899,5 +900,35 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
.blockLast();
}
+ @Test
+ void shouldNotLoadMoreDataThanPrefetch() throws Exception {
+ MailboxSession session = mailboxManager.createSystemSession(USER_1);
+ MailboxPath inbox = MailboxPath.inbox(session);
+
+ mailboxManager.createMailbox(inbox, session).get();
+ MessageManager inboxManager = mailboxManager.getMailbox(inbox, session);
+
+ MessageManager.AppendCommand appendCommand = MessageManager.AppendCommand.from(Message.Builder.of()
+ .setSubject("Test")
+ .setBody("01234567890\r\n", StandardCharsets.UTF_8));
+
+ IntStream.range(0, 64)
+ .forEach(Throwing.intConsumer(i -> inboxManager.appendMessage(appendCommand, session)));
+
+
+ cassandra.getCassandraCluster().getConf().printStatements();
+ StatementRecorder statementRecorder = cassandra.getCassandraCluster().getConf().recordStatements();
+
+ Flux.from(inboxManager.getMessagesReactive(MessageRange.all(), FetchGroup.FULL_CONTENT, session))
+ .skip(10)
+ .next()
+ .block();
+
+ Thread.sleep(1000);
+ assertThat(statementRecorder.listExecutedStatements(StatementRecorder.Selector.preparedStatement("SELECT * FROM blobs WHERE id=:id")))
+ .hasSizeLessThanOrEqualTo(30); // times 2 for header and blob, 10 skipped 5 prefetch.
+ cassandra.getCassandraCluster().getConf().stopPrintingStatements();
+ }
+
}
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
index c2f930bfcd..d4950bd309 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageIdManager.java
@@ -177,11 +177,13 @@ public class StoreMessageIdManager implements MessageIdManager {
MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession);
MessageMapper.FetchType fetchType = FetchGroupConverter.getFetchType(fetchGroup);
+ boolean delayError = false;
+ int prefetch = 1;
return messageIdMapper.findReactive(messageIds, fetchType)
.groupBy(MailboxMessage::getMailboxId)
.filterWhen(groupedFlux -> hasRightsOnMailboxReactive(mailboxSession, Right.Read).apply(groupedFlux.key()), DEFAULT_CONCURRENCY)
.flatMap(Function.identity(), DEFAULT_CONCURRENCY)
- .publishOn(forFetchType(fetchType))
+ .publishOn(forFetchType(fetchType), delayError, prefetch)
.map(Throwing.function(messageResultConverter(fetchGroup)).sneakyThrow());
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
index d81b4e9a59..09aeb04bc8 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreMessageManager.java
@@ -771,9 +771,11 @@ public class StoreMessageManager implements MessageManager {
@Override
public Publisher<MessageResult> getMessagesReactive(MessageRange set, FetchGroup fetchGroup, MailboxSession mailboxSession) {
FetchType fetchType = FetchGroupConverter.getFetchType(fetchGroup);
+ boolean delayError = false;
+ int prefetch = 1;
return Flux.from(mapperFactory.getMessageMapper(mailboxSession)
.findInMailboxReactive(mailbox, set, fetchType, -1))
- .publishOn(forFetchType(fetchType))
+ .publishOn(forFetchType(fetchType), delayError, prefetch)
.map(Throwing.<MailboxMessage, MessageResult>function(message -> ResultUtils.loadMessageResult(message, fetchGroup)).sneakyThrow());
}
diff --git a/server/apps/cassandra-app/src/test/java/org/apache/james/WithCassandraBlobStoreTest.java b/server/apps/cassandra-app/src/test/java/org/apache/james/WithCassandraBlobStoreTest.java
index b025444765..6f3f26aa6d 100644
--- a/server/apps/cassandra-app/src/test/java/org/apache/james/WithCassandraBlobStoreTest.java
+++ b/server/apps/cassandra-app/src/test/java/org/apache/james/WithCassandraBlobStoreTest.java
@@ -19,11 +19,158 @@
package org.apache.james;
+import static org.apache.james.jmap.JMAPTestingConstants.LOCALHOST_IP;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
+
+import org.apache.james.backends.cassandra.StatementRecorder;
+import org.apache.james.backends.cassandra.TestingSession;
+import org.apache.james.backends.cassandra.init.SessionWithInitializedTablesFactory;
+import org.apache.james.core.Domain;
+import org.apache.james.mailbox.DefaultMailboxes;
+import org.apache.james.mailbox.store.BatchSizes;
+import org.apache.james.modules.MailboxProbeImpl;
+import org.apache.james.modules.TestJMAPServerModule;
+import org.apache.james.util.Port;
+import org.apache.james.utils.DataProbeImpl;
+import org.apache.james.utils.GuiceProbe;
+import org.apache.james.utils.SMTPMessageSender;
+import org.apache.james.utils.SpoolerProbe;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.Statement;
+import com.datastax.oss.protocol.internal.request.Batch;
+import com.github.fge.lambdas.Throwing;
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.Resources;
+import com.google.inject.AbstractModule;
+import com.google.inject.Inject;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.multibindings.Multibinder;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
class WithCassandraBlobStoreTest implements MailsShouldBeWellReceivedConcreteContract {
+ private static class TestingSessionProbe implements GuiceProbe {
+ private final TestingSession testingSession;
+
+ @Inject
+ private TestingSessionProbe(TestingSession testingSession) {
+ this.testingSession = testingSession;
+ }
+
+ public TestingSession getTestingSession() {
+ return testingSession;
+ }
+ }
+
+ private static class TestingSessionModule extends AbstractModule {
+ @Override
+ protected void configure() {
+ Multibinder.newSetBinder(binder(), GuiceProbe.class)
+ .addBinding()
+ .to(TestingSessionProbe.class);
+
+ bind(CqlSession.class).to(TestingSession.class);
+ }
+
+ @Provides
+ @Singleton
+ TestingSession provideSession(SessionWithInitializedTablesFactory factory) {
+ return new TestingSession(factory.get());
+ }
+ }
+
@RegisterExtension
- static JamesServerExtension jamesServerExtension = WithCassandraBlobStoreImmutableTest.baseExtensionBuilder()
+ static JamesServerExtension jamesServerExtension = TestingDistributedJamesServerBuilder.withSearchConfiguration(SearchConfiguration.scanning())
+ .extension(new CassandraExtension())
+ .server(configuration -> CassandraJamesServerMain.createServer(configuration)
+ .overrideWith(new TestJMAPServerModule()))
+ .overrideServerModule(new TestingSessionModule())
+ .overrideServerModule(binder -> binder.bind(BatchSizes.class).toInstance(BatchSizes.uniqueBatchSize(5)))
.lifeCycle(JamesServerExtension.Lifecycle.PER_TEST)
.build();
+
+ @Test
+ void imapFetchBackPressureShouldNotLoadMoreDataThanNecessary(GuiceJamesServer server) throws Exception {
+ // 800K message
+ String msgIn = "MIME-Version: 1.0\r\n\r\nCONTENT\r\n\r\n" + "0123456789\r\n0123456789\r\n0123456789\r\n".repeat(25 * 1024);
+
+ server.getProbe(DataProbeImpl.class).fluent()
+ .addDomain(DOMAIN)
+ .addUser(JAMES_USER, PASSWORD)
+ .addUser(SENDER, PASSWORD);
+
+ MailboxProbeImpl mailboxProbe = server.getProbe(MailboxProbeImpl.class);
+ mailboxProbe.createMailbox("#private", JAMES_USER, DefaultMailboxes.INBOX);
+
+ int messageCount = 20;
+
+ Port smtpPort = Port.of(smtpPort(server));
+ try (SMTPMessageSender sender = new SMTPMessageSender(Domain.LOCALHOST.asString())) {
+ Mono.fromRunnable(
+ Throwing.runnable(() -> {
+ sender.connect(JAMES_SERVER_HOST, smtpPort).authenticate(SENDER, PASSWORD);
+ sender.sendMessageWithHeaders("bob@apache.org", JAMES_USER, msgIn);
+ }))
+ .repeat(messageCount - 1)
+ .subscribeOn(Schedulers.fromExecutor(EXECUTOR))
+ .blockLast();
+ }
+
+ CALMLY_AWAIT_FIVE_MINUTE.until(() -> server.getProbe(SpoolerProbe.class).processingFinished());
+
+ SocketChannel clientConnection = SocketChannel.open();
+ clientConnection.connect(new InetSocketAddress(LOCALHOST_IP, imapPort(server)));
+ readBytes(clientConnection);
+ clientConnection.write(ByteBuffer.wrap(String.format("a0 LOGIN %s %s\r\n", JAMES_USER, PASSWORD).getBytes(StandardCharsets.UTF_8)));
+ readBytes(clientConnection);
+
+ clientConnection.write(ByteBuffer.wrap(("A1 SELECT INBOX\r\n").getBytes(StandardCharsets.UTF_8)));
+ // Select completes first
+ readStringUntil(clientConnection, s -> s.contains("A1 OK [READ-WRITE] SELECT completed."));
+
+ TestingSession testingSession = server.getProbe(TestingSessionProbe.class).getTestingSession();
+
+ StatementRecorder statementRecorder = testingSession.recordStatements();
+ clientConnection.write(ByteBuffer.wrap(("A2 UID FETCH 1:500 (BODY.PEEK[])\r\n").getBytes(StandardCharsets.UTF_8)));
+
+ // Await backpressure to trigger
+ Thread.sleep(2000);
+ assertThat(statementRecorder
+ .listExecutedStatements(StatementRecorder.Selector.preparedStatement("SELECT * FROM blobs WHERE id=:id")))
+ .hasSizeLessThanOrEqualTo(30); // times 2 for header and blob, 10 skipped 5 prefetch.
+ }
+
+ private byte[] readBytes(SocketChannel channel) throws IOException {
+ ByteBuffer line = ByteBuffer.allocate(1024);
+ channel.read(line);
+ line.rewind();
+ byte[] bline = new byte[line.remaining()];
+ line.get(bline);
+ return bline;
+ }
+
+ private List<String> readStringUntil(SocketChannel channel, Predicate<String> condition) throws IOException {
+ ImmutableList.Builder<String> result = ImmutableList.builder();
+ while (true) {
+ String line = new String(readBytes(channel), StandardCharsets.US_ASCII);
+ System.out.println(line);
+ result.add(line);
+ if (condition.test(line)) {
+ return result.build();
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 01/04: JAMES-3977 Add bytes preservation in manager and mapper contracts
Posted by rc...@apache.org.
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
commit d2502daef62c7414ede66508d2cf8131e25487a1
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Wed Mar 27 10:25:16 2024 +0100
JAMES-3977 Add bytes preservation in manager and mapper contracts
This was not preserved for the Cassandra mailbox, leading to
innacurrate backpressure and down the line to OutOfMemoryExceptions
---
.../apache/james/mailbox/MailboxManagerTest.java | 7 +++
.../cassandra/mail/CassandraMessageMetadata.java | 10 ++++
.../inmemory/mail/InMemoryMapperProvider.java | 3 +-
.../store/mail/model/DelegatingMailboxMessage.java | 5 ++
.../store/mail/model/impl/SimpleMessage.java | 20 +++++--
.../mailbox/store/mail/model/MapperProvider.java | 4 +-
.../store/mail/model/MessageMapperTest.java | 62 +++++++++++++++++++---
7 files changed, 98 insertions(+), 13 deletions(-)
diff --git a/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java b/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java
index f70d7f5fc4..e7d875ae8c 100644
--- a/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java
+++ b/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java
@@ -2923,6 +2923,13 @@ public abstract class MailboxManagerTest<T extends MailboxManager> {
MessageResult messageResult = inboxManager.getMessages(MessageRange.one(composeId1.getUid()), FetchGroup.MINIMAL, session).next();
assertThat(messageResult.getSaveDate()).isPresent();
}
+
+ @Test
+ void fullFetchGroupShouldPreserveBytesSequence() throws Exception {
+ ComposedMessageId composeId1 = inboxManager.appendMessage(AppendCommand.builder().build(message), session).getId();
+ MessageResult messageResult = inboxManager.getMessages(MessageRange.one(composeId1.getUid()), FetchGroup.FULL_CONTENT, session).next();
+ assertThat(messageResult.getFullContent().asBytesSequence()).isPresent();
+ }
}
@Nested
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMetadata.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMetadata.java
index 56e1203e41..c741f7e645 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMetadata.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMetadata.java
@@ -268,6 +268,16 @@ public class CassandraMessageMetadata {
return delegate.getFullContent();
}
+ @Override
+ public Optional<byte[][]> getFullBytes() {
+ return delegate.getFullBytes();
+ }
+
+ @Override
+ public Optional<byte[][]> getHeadersBytes() {
+ return delegate.getHeadersBytes();
+ }
+
@Override
public ThreadId getThreadId() {
return delegate.getThreadId();
diff --git a/mailbox/memory/src/test/java/org/apache/james/mailbox/inmemory/mail/InMemoryMapperProvider.java b/mailbox/memory/src/test/java/org/apache/james/mailbox/inmemory/mail/InMemoryMapperProvider.java
index 4667385719..286057ee16 100644
--- a/mailbox/memory/src/test/java/org/apache/james/mailbox/inmemory/mail/InMemoryMapperProvider.java
+++ b/mailbox/memory/src/test/java/org/apache/james/mailbox/inmemory/mail/InMemoryMapperProvider.java
@@ -113,7 +113,8 @@ public class InMemoryMapperProvider implements MapperProvider {
Capabilities.ANNOTATION,
Capabilities.MOVE,
Capabilities.ACL_STORAGE,
- Capabilities.UNIQUE_MESSAGE_ID);
+ Capabilities.UNIQUE_MESSAGE_ID,
+ Capabilities.FULL_BYTES);
}
@Override
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/DelegatingMailboxMessage.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/DelegatingMailboxMessage.java
index 29726f4aa0..4a743ecf41 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/DelegatingMailboxMessage.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/DelegatingMailboxMessage.java
@@ -132,4 +132,9 @@ public abstract class DelegatingMailboxMessage implements MailboxMessage {
public Optional<byte[][]> getFullBytes() {
return message.getFullBytes();
}
+
+ @Override
+ public Optional<byte[][]> getHeadersBytes() {
+ return message.getHeadersBytes();
+ }
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/impl/SimpleMessage.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/impl/SimpleMessage.java
index ffe9b02ce2..1fb0a44722 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/impl/SimpleMessage.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/impl/SimpleMessage.java
@@ -106,11 +106,23 @@ public class SimpleMessage implements Message {
@Override
public InputStream getHeaderContent() throws IOException {
- long headerEnd = bodyStartOctet;
- if (headerEnd < 0) {
- headerEnd = 0;
+ return new BoundedInputStream(content.getInputStream(), headerSize());
+ }
+
+ private long headerSize() {
+ return Math.max(0, bodyStartOctet);
+ }
+
+ @Override
+ public Optional<byte[][]> getHeadersBytes() {
+ try {
+ if (headerSize() == content.size()) {
+ return content.asBytesSequence();
+ }
+ } catch (MailboxException e) {
+ return Optional.empty();
}
- return new BoundedInputStream(content.getInputStream(), headerEnd);
+ return Optional.empty();
}
@Override
diff --git a/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MapperProvider.java b/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MapperProvider.java
index 2db1faa2ae..b6bd054d2e 100644
--- a/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MapperProvider.java
+++ b/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MapperProvider.java
@@ -42,7 +42,9 @@ public interface MapperProvider {
UNIQUE_MESSAGE_ID,
THREAD_SAFE_FLAGS_UPDATE,
INCREMENTAL_APPLICABLE_FLAGS,
- ACL_STORAGE
+ ACL_STORAGE,
+ FULL_BYTES,
+ HEADER_BYTES
}
List<Capabilities> getSupportedCapabilities();
diff --git a/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MessageMapperTest.java b/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MessageMapperTest.java
index b4514d3728..3d16316189 100644
--- a/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MessageMapperTest.java
+++ b/mailbox/store/src/test/java/org/apache/james/mailbox/store/mail/model/MessageMapperTest.java
@@ -73,7 +73,7 @@ public abstract class MessageMapperTest {
private static final char DELIMITER = '.';
private static final int LIMIT = 10;
- private static final int BODY_START = 16;
+ private static final int BODY_START = 19;
private static final UidValidity UID_VALIDITY = UidValidity.of(42);
private static final String CUSTOMS_USER_FLAGS_VALUE = "CustomsFlags";
@@ -115,12 +115,12 @@ public abstract class MessageMapperTest {
benwaInboxMailbox = createMailbox(MailboxPath.forUser(BENWA, "INBOX"));
benwaWorkMailbox = createMailbox(MailboxPath.forUser(BENWA, "INBOX" + DELIMITER + "work"));
- message1 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test1 \n\nBody1\n.\n", BODY_START, new PropertyBuilder());
- message2 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test2 \n\nBody2\n.\n", BODY_START, new PropertyBuilder());
- message3 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test3 \n\nBody3\n.\n", BODY_START, new PropertyBuilder());
- message4 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test4 \n\nBody4\n.\n", BODY_START, new PropertyBuilder());
- message5 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test5 \n\nBody5\n.\n", BODY_START, new PropertyBuilder());
- message6 = createMessage(benwaWorkMailbox, mapperProvider.generateMessageId(), "Subject: Test6 \n\nBody6\n.\n", BODY_START, new PropertyBuilder());
+ message1 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test1 \r\n\r\nBody1\n.\n", BODY_START, new PropertyBuilder());
+ message2 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test2 \r\n\r\nBody2\n.\n", BODY_START, new PropertyBuilder());
+ message3 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test3 \r\n\r\nBody3\n.\n", BODY_START, new PropertyBuilder());
+ message4 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test4 \r\n\r\nBody4\n.\n", BODY_START, new PropertyBuilder());
+ message5 = createMessage(benwaInboxMailbox, mapperProvider.generateMessageId(), "Subject: Test5 \r\n\r\nBody5\n.\n", BODY_START, new PropertyBuilder());
+ message6 = createMessage(benwaWorkMailbox, mapperProvider.generateMessageId(), "Subject: Test6 \r\n\r\nBody6\n.\n", BODY_START, new PropertyBuilder());
}
@Test
@@ -212,6 +212,54 @@ public abstract class MessageMapperTest {
.isEqualToWithoutAttachment(message1, fetchType);
}
+ @Test
+ void getFullBytesShouldBePresentWhenFullFetchType() throws MailboxException, IOException {
+ Assume.assumeTrue(mapperProvider.getSupportedCapabilities().contains(Capabilities.FULL_BYTES));
+ saveMessages();
+ int limit = 10;
+ MailboxMessage next = messageMapper.findInMailbox(benwaInboxMailbox, MessageRange.one(message1.getUid()), FetchType.FULL, limit).next();
+ assertThat(next.getFullBytes())
+ .isNotEmpty();
+ }
+
+ @Test
+ void getFullBytesShouldBePresentWhenHeadersFetchType() throws Exception {
+ Assume.assumeTrue(mapperProvider.getSupportedCapabilities().contains(Capabilities.FULL_BYTES));
+ saveMessages();
+ int limit = 10;
+ assertThat(messageMapper.findInMailbox(benwaInboxMailbox, MessageRange.one(message1.getUid()), FetchType.HEADERS, limit).next().getFullBytes())
+ .get()
+ .satisfies(b -> assertThat(b.length).isEqualTo(1));
+ }
+
+ @Test
+ void getHeadersBytesShouldBePresentWhenHeadersFetchType() throws Exception {
+ Assume.assumeTrue(mapperProvider.getSupportedCapabilities().contains(Capabilities.HEADER_BYTES));
+ saveMessages();
+ int limit = 10;
+ assertThat(messageMapper.findInMailbox(benwaInboxMailbox, MessageRange.one(message1.getUid()), FetchType.HEADERS, limit).next().getHeadersBytes())
+ .isNotEmpty();
+ }
+
+ @Test
+ void getFullBytesShouldBePresentWhenAttachmentMetadataFetchType() throws Exception {
+ Assume.assumeTrue(mapperProvider.getSupportedCapabilities().contains(Capabilities.FULL_BYTES));
+ saveMessages();
+ int limit = 10;
+ assertThat(messageMapper.findInMailbox(benwaInboxMailbox, MessageRange.one(message1.getUid()), FetchType.ATTACHMENTS_METADATA, limit).next().getFullBytes())
+ .get()
+ .satisfies(b -> assertThat(b.length).isEqualTo(1));
+ }
+
+ @Test
+ void getHeadersBytesShouldBePresentWhenAttachmentMetadataFetchType() throws Exception {
+ Assume.assumeTrue(mapperProvider.getSupportedCapabilities().contains(Capabilities.HEADER_BYTES));
+ saveMessages();
+ int limit = 10;
+ assertThat(messageMapper.findInMailbox(benwaInboxMailbox, MessageRange.one(message1.getUid()), FetchType.ATTACHMENTS_METADATA, limit).next().getHeadersBytes())
+ .isNotEmpty();
+ }
+
@Test
void messagesCanBeRetrievedInMailboxWithRangeTypeRange() throws MailboxException, IOException {
saveMessages();
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 02/04: JAMES-3977 Improve IMAP FETCH concurrency control
Posted by rc...@apache.org.
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
commit 487120bf77685d2a41e22a72a45485f94763ec7e
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Wed Mar 27 12:25:54 2024 +0100
JAMES-3977 Improve IMAP FETCH concurrency control
---
.../james/imap/processor/fetch/FetchProcessor.java | 19 ++++++++++---------
1 file changed, 10 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 d90a147e01..6619064152 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
@@ -79,7 +79,6 @@ import reactor.core.publisher.Sinks;
public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
static class FetchSubscriber implements Subscriber<FetchResponse> {
private final AtomicReference<Subscription> subscription = new AtomicReference<>();
- private final AtomicBoolean requested = new AtomicBoolean(false);
private final Sinks.One<Void> sink = Sinks.one();
private final ImapSession imapSession;
private final Responder responder;
@@ -97,21 +96,23 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
@Override
public void onNext(FetchResponse fetchResponse) {
- requested.getAndSet(false);
+ AtomicBoolean mustRequestOne = new AtomicBoolean(true);
responder.respond(fetchResponse);
- if (imapSession.backpressureNeeded(this::requestOne)) {
+ Runnable requestOne = () -> {
+ if (mustRequestOne.getAndSet(false)) {
+ requestOne();
+ }
+ };
+ if (imapSession.backpressureNeeded(requestOne)) {
LOGGER.debug("Applying backpressure as we encounter a slow reader");
} else {
- requestOne();
+ requestOne.run();
}
}
private void requestOne() {
- boolean alreadyRequested = requested.getAndSet(true);
- if (!alreadyRequested) {
- Optional.ofNullable(subscription.get())
- .ifPresent(s -> s.request(1));
- }
+ Optional.ofNullable(subscription.get())
+ .ifPresent(s -> s.request(1));
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 04/04: JAMES-3977 IMAP log should embed username where possible
Posted by rc...@apache.org.
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
commit 8155100e36e81668d803693daff8693177376299
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Thu Mar 28 09:41:57 2024 +0100
JAMES-3977 IMAP log should embed username where possible
Easier filtering during log analysis
---
.../james/imap/processor/fetch/FetchProcessor.java | 8 ++++++--
.../imapserver/netty/ImapChannelUpstreamHandler.java | 17 ++++++++++++++---
2 files changed, 20 insertions(+), 5 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 6619064152..90291cae84 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
@@ -99,14 +99,18 @@ public class FetchProcessor extends AbstractMailboxProcessor<FetchRequest> {
AtomicBoolean mustRequestOne = new AtomicBoolean(true);
responder.respond(fetchResponse);
Runnable requestOne = () -> {
+ LOGGER.info("Resuming IMAP FETCH for user {}", imapSession.getUserName().asString());
if (mustRequestOne.getAndSet(false)) {
requestOne();
}
};
if (imapSession.backpressureNeeded(requestOne)) {
- LOGGER.debug("Applying backpressure as we encounter a slow reader");
+ LOGGER.info("Applying backpressure as user {} is a slow reader",
+ imapSession.getUserName().asString());
} else {
- requestOne.run();
+ if (mustRequestOne.getAndSet(false)) {
+ requestOne();
+ }
}
}
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 094d0284e3..d6110a4f1f 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
@@ -34,6 +34,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import javax.net.ssl.SSLHandshakeException;
+import org.apache.james.core.Username;
import org.apache.james.imap.api.ConnectionCheck;
import org.apache.james.imap.api.ImapConstants;
import org.apache.james.imap.api.ImapMessage;
@@ -269,7 +270,7 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
ImapSession imapSession = ctx.channel().attr(IMAP_SESSION_ATTRIBUTE_KEY).getAndSet(null);
try (Closeable closeable = mdc(imapSession).build()) {
InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
- LOGGER.info("Connection closed for {}", address.getAddress().getHostAddress());
+ LOGGER.info("Connection closed for {} and user {}", address.getAddress().getHostAddress(), retrieveUsername(imapSession));
Optional.ofNullable(imapSession).ifPresent(ImapSession::cancelOngoingProcessing);
Optional.ofNullable(imapSession)
@@ -285,11 +286,20 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
}
}
+ private static String retrieveUsername(ImapSession imapSession) {
+ return Optional.ofNullable(imapSession)
+ .flatMap(session -> Optional.ofNullable(session.getUserName()))
+ .map(Username::asString)
+ .orElse("");
+ }
+
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ ImapSession imapSession = ctx.channel().attr(IMAP_SESSION_ATTRIBUTE_KEY).getAndSet(null);
+ String username = retrieveUsername(imapSession);
try (Closeable closeable = mdc(ctx).build()) {
if (cause instanceof SocketException) {
- LOGGER.info("Socket exception encountered: {}", cause.getMessage());
+ LOGGER.info("Socket exception encountered for user {}: {}", username, cause.getMessage());
} else if (isSslHandshkeException(cause)) {
LOGGER.info("SSH handshake rejected {}", cause.getMessage());
} else if (isNotSslRecordException(cause)) {
@@ -370,7 +380,8 @@ public class ImapChannelUpstreamHandler extends ChannelInboundHandlerAdapter imp
.subscribe(any -> {
}, e -> {
- LOGGER.error("Exception while handling errors for channel {}", ctx.channel(), e);
+ LOGGER.error("Exception while handling errors for channel {} and user {}", ctx.channel(),
+ Optional.ofNullable(imapSession).map(u -> u.getUserName().asString()).orElse(""), e);
Channel channel = ctx.channel();
if (channel.isActive()) {
channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org