This is an automated email from the ASF dual-hosted git repository.
Arsnael 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 75a3c1e7a4 [ENHANCEMENT] Prevent FETCH fragmentation (#3094)
75a3c1e7a4 is described below
commit 75a3c1e7a4ae0656ffc8558c5853aba00a4f9009
Author: Benoit TELLIER <[email protected]>
AuthorDate: Wed Jul 22 08:43:07 2026 +0200
[ENHANCEMENT] Prevent FETCH fragmentation (#3094)
---
.../james/imap/api/process/ImapProcessor.java | 5 +
.../james/imap/encode/ImapResponseComposer.java | 3 +
.../imap/encode/base/ImapResponseComposerImpl.java | 37 +++++-
.../apache/james/imap/main/ResponseEncoder.java | 9 ++
.../james/imap/message/SequencedLiteral.java | 74 ++++++++++++
.../james/imap/processor/fetch/FetchProcessor.java | 3 +
.../imap/encode/ImapResponseComposerImplTest.java | 100 ++++++++++++++++-
.../james/imap/message/SequencedLiteralTest.java | 90 +++++++++++++++
.../fetch/FetchResponseLiteralSizeTest.java | 125 +++++++++++++++++++++
.../netty/ChannelImapResponseWriter.java | 47 +++++---
10 files changed, 474 insertions(+), 19 deletions(-)
diff --git
a/protocols/imap/src/main/java/org/apache/james/imap/api/process/ImapProcessor.java
b/protocols/imap/src/main/java/org/apache/james/imap/api/process/ImapProcessor.java
index 5917f6c81a..83cebcdb23 100644
---
a/protocols/imap/src/main/java/org/apache/james/imap/api/process/ImapProcessor.java
+++
b/protocols/imap/src/main/java/org/apache/james/imap/api/process/ImapProcessor.java
@@ -67,5 +67,10 @@ public interface ImapProcessor {
void respond(ImapResponseMessage message);
void flush();
+
+ /** Flushes iff the response just written deferred a literal, so its
short-lived source is read while live. */
+ default void flushIfNeeded() {
+
+ }
}
}
diff --git
a/protocols/imap/src/main/java/org/apache/james/imap/encode/ImapResponseComposer.java
b/protocols/imap/src/main/java/org/apache/james/imap/encode/ImapResponseComposer.java
index ebf546217b..c76e0f4624 100644
---
a/protocols/imap/src/main/java/org/apache/james/imap/encode/ImapResponseComposer.java
+++
b/protocols/imap/src/main/java/org/apache/james/imap/encode/ImapResponseComposer.java
@@ -34,6 +34,9 @@ public interface ImapResponseComposer {
void flush() throws IOException;
+ /** Flushes iff a literal was deferred (its source may be short-lived),
leaving literal-free responses to batch. */
+ void flushIfNeeded() throws IOException;
+
/**
* Writes an untagged NO response. Indicates that a warning. The command
may
* still complete sucessfully.
diff --git
a/protocols/imap/src/main/java/org/apache/james/imap/encode/base/ImapResponseComposerImpl.java
b/protocols/imap/src/main/java/org/apache/james/imap/encode/base/ImapResponseComposerImpl.java
index b489204c0c..dff62000a6 100644
---
a/protocols/imap/src/main/java/org/apache/james/imap/encode/base/ImapResponseComposerImpl.java
+++
b/protocols/imap/src/main/java/org/apache/james/imap/encode/base/ImapResponseComposerImpl.java
@@ -22,6 +22,8 @@ package org.apache.james.imap.encode.base;
import static java.nio.charset.StandardCharsets.US_ASCII;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Optional;
import jakarta.mail.Flags;
@@ -34,7 +36,9 @@ import org.apache.james.imap.api.message.IdRange;
import org.apache.james.imap.api.message.UidRange;
import org.apache.james.imap.encode.ImapResponseComposer;
import org.apache.james.imap.encode.ImapResponseWriter;
+import org.apache.james.imap.message.BytesBackedLiteral;
import org.apache.james.imap.message.Literal;
+import org.apache.james.imap.message.SequencedLiteral;
import org.apache.james.imap.utils.FastByteArrayOutputStream;
/**
@@ -64,6 +68,9 @@ public class ImapResponseComposerImpl implements
ImapConstants, ImapResponseComp
private boolean skipNextSpace;
+ // Text chunks and literals gathered to be emitted as a single
SequencedLiteral (one flush, no copy). Null until a literal is buffered.
+ private List<Literal> pendingLiteralParts;
+
public ImapResponseComposerImpl(ImapResponseWriter writer, int bufferSize)
{
skipNextSpace = false;
this.writer = writer;
@@ -146,12 +153,26 @@ public class ImapResponseComposerImpl implements
ImapConstants, ImapResponseComp
}
public void flush() throws IOException {
- if (buffer.size() > 0) {
+ if (pendingLiteralParts != null) {
+ snapshotBufferAsPart();
+ List<Literal> parts = pendingLiteralParts;
+ // Reset before writing: the writer flush callback may re-enter
flush(), which must then be a no-op.
+ pendingLiteralParts = null;
+ writer.write(new SequencedLiteral(parts));
+ } else if (buffer.size() > 0) {
writer.write(buffer.toByteArray());
buffer.reset();
}
}
+ @Override
+ public void flushIfNeeded() throws IOException {
+ // A deferred literal may wrap short-lived content (e.g. a DB blob
stream): flush it while its source is live.
+ if (pendingLiteralParts != null) {
+ flush();
+ }
+ }
+
@Override
public ImapResponseComposer tag(Tag tag) throws IOException {
writeASCII(tag.asString());
@@ -317,11 +338,23 @@ public class ImapResponseComposerImpl implements
ImapConstants, ImapResponseComp
buffer.write(BYTE_CLOSE_BRACE);
end();
if (size > 0) {
- writer.write(literal);
+ // Defer: snapshot accumulated text then the literal, coalesced
into one SequencedLiteral at flush().
+ if (pendingLiteralParts == null) {
+ pendingLiteralParts = new ArrayList<>();
+ }
+ snapshotBufferAsPart();
+ pendingLiteralParts.add(literal);
}
return this;
}
+ private void snapshotBufferAsPart() {
+ if (buffer.size() > 0) {
+
pendingLiteralParts.add(BytesBackedLiteral.of(buffer.toByteArray()));
+ buffer.reset();
+ }
+ }
+
@Override
public ImapResponseComposer closeSquareBracket() throws IOException {
closeBracket(BYTE_CLOSE_SQUARE_BRACKET);
diff --git
a/protocols/imap/src/main/java/org/apache/james/imap/main/ResponseEncoder.java
b/protocols/imap/src/main/java/org/apache/james/imap/main/ResponseEncoder.java
index 6524bd7de4..825abcd7e3 100644
---
a/protocols/imap/src/main/java/org/apache/james/imap/main/ResponseEncoder.java
+++
b/protocols/imap/src/main/java/org/apache/james/imap/main/ResponseEncoder.java
@@ -64,4 +64,13 @@ public class ResponseEncoder implements Responder {
throw new RuntimeException(e);
}
}
+
+ @Override
+ public void flushIfNeeded() {
+ try {
+ composer.flushIfNeeded();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
}
diff --git
a/protocols/imap/src/main/java/org/apache/james/imap/message/SequencedLiteral.java
b/protocols/imap/src/main/java/org/apache/james/imap/message/SequencedLiteral.java
new file mode 100644
index 0000000000..4623d7f25d
--- /dev/null
+++
b/protocols/imap/src/main/java/org/apache/james/imap/message/SequencedLiteral.java
@@ -0,0 +1,74 @@
+/****************************************************************
+ * 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.imap.message;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Vector;
+
+/** A {@link Literal} concatenating an ordered sequence of {@link Literal}s,
so a whole response fragment (text + literal(s) + text) is emitted in one pass
and a single flush. */
+public class SequencedLiteral implements Literal {
+ private final List<Literal> parts;
+
+ public SequencedLiteral(List<Literal> parts) {
+ this.parts = parts;
+ }
+
+ public List<Literal> parts() {
+ return parts;
+ }
+
+ @Override
+ public long size() throws IOException {
+ long size = 0;
+ for (Literal part : parts) {
+ size += part.size();
+ }
+ return size;
+ }
+
+ @Override
+ public InputStream getInputStream() throws IOException {
+ Vector<InputStream> streams = new Vector<>(parts.size());
+ for (Literal part : parts) {
+ streams.add(part.getInputStream());
+ }
+ return new SequenceInputStream(streams.elements());
+ }
+
+ @Override
+ public Optional<byte[][]> asBytesSequence() {
+ List<byte[]> chunks = new ArrayList<>();
+ for (Literal part : parts) {
+ Optional<byte[][]> partChunks = part.asBytesSequence();
+ if (partChunks.isEmpty()) {
+ // A single non-in-memory part forces the whole sequence to be
streamed.
+ return Optional.empty();
+ }
+ Collections.addAll(chunks, partChunks.get());
+ }
+ return Optional.of(chunks.toArray(new byte[0][]));
+ }
+}
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 3be25820c7..c7e399ff74 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
@@ -191,6 +191,9 @@ public class FetchProcessor extends
AbstractMailboxProcessor<FetchRequest> {
public void onNext(FetchResponse fetchResponse) {
AtomicBoolean mustRequestOne = new AtomicBoolean(true);
responder.respond(fetchResponse);
+ // Flush iff a literal was deferred, while the message content is
still live; keeps the backpressure
+ // check below on an up-to-date writability, without penalizing
literal-free responses.
+ responder.flushIfNeeded();
Runnable requestOne = () -> {
if (mustRequestOne.getAndSet(false)) {
LOGGER.info("Resuming IMAP FETCH for user {}",
imapSession.getUserName().asString());
diff --git
a/protocols/imap/src/test/java/org/apache/james/imap/encode/ImapResponseComposerImplTest.java
b/protocols/imap/src/test/java/org/apache/james/imap/encode/ImapResponseComposerImplTest.java
index 9be8d9aec6..977dde53bc 100644
---
a/protocols/imap/src/test/java/org/apache/james/imap/encode/ImapResponseComposerImplTest.java
+++
b/protocols/imap/src/test/java/org/apache/james/imap/encode/ImapResponseComposerImplTest.java
@@ -19,20 +19,42 @@
package org.apache.james.imap.encode;
+import static java.nio.charset.StandardCharsets.US_ASCII;
import static org.assertj.core.api.Assertions.assertThat;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.james.imap.encode.base.ByteImapResponseWriter;
import org.apache.james.imap.encode.base.ImapResponseComposerImpl;
+import org.apache.james.imap.message.BytesBackedLiteral;
+import org.apache.james.imap.message.Literal;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
class ImapResponseComposerImplTest {
+ private AtomicInteger byteWrites;
+ private AtomicInteger literalWrites;
private ByteImapResponseWriter writer;
private ImapResponseComposer composer;
@BeforeEach
void setUp() {
- writer = new ByteImapResponseWriter();
+ byteWrites = new AtomicInteger();
+ literalWrites = new AtomicInteger();
+ writer = new ByteImapResponseWriter() {
+ @Override
+ public void write(byte[] buffer) throws IOException {
+ byteWrites.incrementAndGet();
+ super.write(buffer);
+ }
+
+ @Override
+ public void write(Literal literal) throws IOException {
+ literalWrites.incrementAndGet();
+ super.write(literal);
+ }
+ };
composer = new ImapResponseComposerImpl(writer);
}
@@ -45,4 +67,80 @@ class ImapResponseComposerImplTest {
assertThat(writer.getString()).isEqualTo(" \"?\"\r\n");
}
+
+ @Test
+ void plainResponseShouldBeWrittenAsBytesWithoutLiteralWrite() throws
Exception {
+ composer.untaggedResponse("OK completed");
+ composer.flush();
+
+ assertThat(writer.getString()).isEqualTo("* OK completed\r\n");
+ assertThat(byteWrites).hasValue(1);
+ assertThat(literalWrites).hasValue(0);
+ }
+
+ @Test
+ void fetchBodyResponseShouldBeEmittedAsASingleSequencedLiteral() throws
Exception {
+ composer.untagged().message("1").message("FETCH");
+ composer.openParen();
+ composer.message("BODY[]");
+ composer.literal(BytesBackedLiteral.of("Body of
Initial".getBytes(US_ASCII)));
+ composer.closeParen();
+ composer.end();
+ composer.flush();
+
+ assertThat(writer.getString()).isEqualTo("* 1 FETCH (BODY[]
{15}\r\nBody of Initial)\r\n");
+ // Whole response (prefix + literal + trailing ")" + CRLF) leaves
through a single writer call, no
+ // separate byte[] write that would split the literal from the ")".
+ assertThat(literalWrites).hasValue(1);
+ assertThat(byteWrites).hasValue(0);
+ }
+
+ @Test
+ void multipleLiteralsShouldBeEmittedAsASingleSequencedLiteral() throws
Exception {
+ composer.openParen();
+ composer.message("BODY[HEADER]");
+ composer.literal(BytesBackedLiteral.of("H".getBytes(US_ASCII)));
+ composer.message("BODY[TEXT]");
+ composer.literal(BytesBackedLiteral.of("T".getBytes(US_ASCII)));
+ composer.closeParen();
+ composer.end();
+ composer.flush();
+
+ assertThat(writer.getString()).isEqualTo(" (BODY[HEADER] {1}\r\nH
BODY[TEXT] {1}\r\nT)\r\n");
+ assertThat(literalWrites).hasValue(1);
+ assertThat(byteWrites).hasValue(0);
+ }
+
+ @Test
+ void flushIfNeededShouldFlushWhenTheResponseContainsALiteral() throws
Exception {
+ composer.message("BODY[]");
+ composer.literal(BytesBackedLiteral.of("body".getBytes(US_ASCII)));
+ composer.closeParen();
+ composer.end();
+ composer.flushIfNeeded();
+
+ assertThat(writer.getString()).isEqualTo(" BODY[] {4}\r\nbody)\r\n");
+ assertThat(literalWrites).hasValue(1);
+ }
+
+ @Test
+ void flushIfNeededShouldNotFlushALiteralFreeResponse() throws Exception {
+ composer.untaggedResponse("OK");
+ composer.flushIfNeeded();
+
+ assertThat(byteWrites).hasValue(0);
+ assertThat(literalWrites).hasValue(0);
+ }
+
+ @Test
+ void emptyLiteralShouldNotForceASequencedLiteral() throws Exception {
+ composer.message("BODY[]");
+ composer.literal(BytesBackedLiteral.of(new byte[0]));
+ composer.end();
+ composer.flush();
+
+ assertThat(writer.getString()).isEqualTo(" BODY[] {0}\r\n\r\n");
+ assertThat(byteWrites).hasValue(1);
+ assertThat(literalWrites).hasValue(0);
+ }
}
diff --git
a/protocols/imap/src/test/java/org/apache/james/imap/message/SequencedLiteralTest.java
b/protocols/imap/src/test/java/org/apache/james/imap/message/SequencedLiteralTest.java
new file mode 100644
index 0000000000..509bea47b2
--- /dev/null
+++
b/protocols/imap/src/test/java/org/apache/james/imap/message/SequencedLiteralTest.java
@@ -0,0 +1,90 @@
+/****************************************************************
+ * 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.imap.message;
+
+import static java.nio.charset.StandardCharsets.US_ASCII;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.commons.io.IOUtils;
+import org.junit.jupiter.api.Test;
+
+class SequencedLiteralTest {
+ private static Literal streamedLiteral(byte[] content) {
+ return new Literal() {
+ @Override
+ public long size() {
+ return content.length;
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ return new ByteArrayInputStream(content);
+ }
+ };
+ }
+
+ @Test
+ void sizeShouldBeTheSumOfTheParts() throws Exception {
+ SequencedLiteral literal = new SequencedLiteral(List.of(
+ BytesBackedLiteral.of("ab".getBytes(US_ASCII)),
+ BytesBackedLiteral.of("cde".getBytes(US_ASCII))));
+
+ assertThat(literal.size()).isEqualTo(5);
+ }
+
+ @Test
+ void getInputStreamShouldConcatenateTheParts() throws Exception {
+ SequencedLiteral literal = new SequencedLiteral(List.of(
+ BytesBackedLiteral.of(" (BODY[] {15}\r\n".getBytes(US_ASCII)),
+ BytesBackedLiteral.of("Body of Initial".getBytes(US_ASCII)),
+ BytesBackedLiteral.of(")\r\n".getBytes(US_ASCII))));
+
+ assertThat(new String(IOUtils.toByteArray(literal.getInputStream()),
US_ASCII))
+ .isEqualTo(" (BODY[] {15}\r\nBody of Initial)\r\n");
+ }
+
+ @Test
+ void asBytesSequenceShouldAggregateChunksInOrderWhenAllPartsAreInMemory()
throws Exception {
+ SequencedLiteral literal = new SequencedLiteral(List.of(
+ BytesBackedLiteral.of("ab".getBytes(US_ASCII)),
+ BytesBackedLiteral.of("cde".getBytes(US_ASCII))));
+
+ Optional<byte[][]> chunks = literal.asBytesSequence();
+
+ assertThat(chunks).isPresent();
+ assertThat(chunks.get().length).isEqualTo(2);
+ assertThat(new String(chunks.get()[0], US_ASCII)).isEqualTo("ab");
+ assertThat(new String(chunks.get()[1], US_ASCII)).isEqualTo("cde");
+ }
+
+ @Test
+ void asBytesSequenceShouldBeEmptyWhenAPartIsNotInMemory() {
+ SequencedLiteral literal = new SequencedLiteral(List.of(
+ BytesBackedLiteral.of("ab".getBytes(US_ASCII)),
+ streamedLiteral("cde".getBytes(US_ASCII))));
+
+ assertThat(literal.asBytesSequence()).isEmpty();
+ }
+}
diff --git
a/protocols/imap/src/test/java/org/apache/james/imap/processor/fetch/FetchResponseLiteralSizeTest.java
b/protocols/imap/src/test/java/org/apache/james/imap/processor/fetch/FetchResponseLiteralSizeTest.java
new file mode 100644
index 0000000000..3fbdd78f33
--- /dev/null
+++
b/protocols/imap/src/test/java/org/apache/james/imap/processor/fetch/FetchResponseLiteralSizeTest.java
@@ -0,0 +1,125 @@
+/****************************************************************
+ * 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.imap.processor.fetch;
+
+import static java.nio.charset.StandardCharsets.US_ASCII;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.james.imap.encode.FetchResponseEncoder;
+import org.apache.james.imap.encode.ImapResponseComposer;
+import org.apache.james.imap.encode.base.ByteImapResponseWriter;
+import org.apache.james.imap.encode.base.ImapResponseComposerImpl;
+import org.apache.james.imap.message.response.FetchResponse;
+import org.apache.james.imap.message.response.FetchResponse.BodyElement;
+import org.apache.james.mailbox.MessageSequenceNumber;
+import org.apache.james.mailbox.model.ByteContent;
+import org.apache.james.mailbox.model.Header;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Byte-exact regression guard: the {@code {N}} literal advertised for every
{@code BODY[...]} section MUST equal
+ * the exact number of octets streamed between {@code {N}\r\n} and the closing
{@code )}. An undercount here
+ * mis-frames the response for a byte-exact client (the leftover octets
surface as a spurious tagged line).
+ *
+ * <p>Unlike the MPT {@code .test} scripts (which are line-oriented and never
frame the literal by its declared
+ * size), this test parses the raw wire bytes and asserts declared-vs-streamed
equality.
+ */
+class FetchResponseLiteralSizeTest {
+ private static final MessageSequenceNumber MSN =
MessageSequenceNumber.of(1);
+ // Matches the FIRST literal marker "{N}\r\n" in the response.
+ private static final Pattern LITERAL = Pattern.compile("\\{(\\d+)}\r\n");
+
+ private byte[] encode(BodyElement element) throws Exception {
+ ByteImapResponseWriter writer = new ByteImapResponseWriter();
+ ImapResponseComposer composer = new ImapResponseComposerImpl(writer);
+ FetchResponse response = new FetchResponse(MSN, null, null, null,
null, null, null,
+ null, null, null, List.of(element), null, null);
+ new FetchResponseEncoder(false).encode(response, composer);
+ composer.flush();
+ return writer.getBytes();
+ }
+
+ /**
+ * Parses "... {N}\r\n<content>)\r\n" and returns [declaredN,
measuredContentLength].
+ * The response always ends with the section-closing ")" then CRLF, so the
streamed content is everything
+ * between the "{N}\r\n" marker and the trailing ")\r\n".
+ */
+ private long[] declaredVsStreamed(byte[] response) {
+ String asString = new String(response, US_ASCII);
+ Matcher matcher = LITERAL.matcher(asString);
+ assertThat(matcher.find()).as("response must contain a literal marker:
%s", asString).isTrue();
+ long declared = Long.parseLong(matcher.group(1));
+ int contentStart = matcher.end();
+ // Response tail is ")\r\n" (3 octets): ")" closing the FETCH paren,
then CRLF.
+ int contentEnd = response.length - 3;
+ assertThat(new String(response, contentEnd, 3,
US_ASCII)).isEqualTo(")\r\n");
+ return new long[] {declared, contentEnd - contentStart};
+ }
+
+ private void assertLiteralIsExact(BodyElement element) throws Exception {
+ long[] declaredVsStreamed = declaredVsStreamed(encode(element));
+ assertThat(declaredVsStreamed[0])
+ .as("declared {%d} must equal streamed %d octets",
declaredVsStreamed[0], declaredVsStreamed[1])
+ .isEqualTo(declaredVsStreamed[1]);
+ }
+
+ @Test
+ void headerFieldsMessageIdShouldAdvertiseExactLiteralSize() throws
Exception {
+ // The exact shape reported by the failing client: BODY[HEADER.FIELDS
(MESSAGE-ID)] with a single header.
+ // Streamed content is "Message-ID: <initial@source>\r\n" (30) +
section-terminating "\r\n" (2) = 32 octets.
+ HeaderBodyElement element = new HeaderBodyElement("BODY[HEADER.FIELDS
(MESSAGE-ID)]",
+ List.of(new Header("Message-ID", "<initial@source>")));
+
+ long[] declaredVsStreamed = declaredVsStreamed(encode(element));
+
+ assertThat(declaredVsStreamed[0]).isEqualTo(32);
+ assertThat(declaredVsStreamed[1]).isEqualTo(32);
+ }
+
+ @Test
+ void headerFieldsWithSeveralHeadersShouldAdvertiseExactLiteralSize()
throws Exception {
+ assertLiteralIsExact(new HeaderBodyElement("BODY[HEADER.FIELDS (FROM
TO SUBJECT)]",
+ List.of(
+ new Header("From", "Timothy Tayler <[email protected]>"),
+ new Header("To", "Samual Smith <[email protected]>"),
+ new Header("Subject", "A Simple Email"))));
+ }
+
+ @Test
+ void fullBodyShouldAdvertiseExactLiteralSize() throws Exception {
+ byte[] message = ("Message-ID: <initial@source>\r\n"
+ + "Subject: A Simple Email\r\n"
+ + "\r\n"
+ + "This is a very simple email.\r\n").getBytes(US_ASCII);
+ assertLiteralIsExact(new ContentBodyElement("BODY[]", new
ByteContent(message)));
+ }
+
+ @Test
+ void headerBlockShouldAdvertiseExactLiteralSize() throws Exception {
+ byte[] headerBlock = ("Message-ID: <initial@source>\r\n"
+ + "Subject: A Simple Email\r\n"
+ + "\r\n").getBytes(US_ASCII);
+ assertLiteralIsExact(new ContentBodyElement("BODY[HEADER]", new
ByteContent(headerBlock)));
+ }
+}
diff --git
a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ChannelImapResponseWriter.java
b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ChannelImapResponseWriter.java
index 3ab11de936..362c4ee729 100644
---
a/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ChannelImapResponseWriter.java
+++
b/server/protocols/protocols-imap4/src/main/java/org/apache/james/imapserver/netty/ChannelImapResponseWriter.java
@@ -25,10 +25,12 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.channels.FileChannel;
+import java.util.Optional;
import org.apache.james.imap.api.process.ImapSession;
import org.apache.james.imap.encode.ImapResponseWriter;
import org.apache.james.imap.message.Literal;
+import org.apache.james.imap.message.SequencedLiteral;
import org.apache.james.util.MDCStructuredLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -97,25 +99,38 @@ public class ChannelImapResponseWriter implements
ImapResponseWriter {
public void write(Literal literal) throws IOException {
flushCallback.run();
if (channel.isActive()) {
- if (literal.asBytesSequence().isPresent()) {
-
channel.writeAndFlush(Unpooled.wrappedBuffer(literal.asBytesSequence().get()));
- return;
- }
- InputStream in = literal.getInputStream();
- if (in instanceof FileInputStream) {
- FileChannel fc = ((FileInputStream) in).getChannel();
- // Zero-copy is only possible if no SSL/TLS and no COMPRESS
is in place
- //
- // See JAMES-1305 and JAMES-1306
- ChannelPipeline cp = channel.pipeline();
- if (zeroCopy && cp.get(SslHandler.class) == null &&
cp.get(ZlibEncoder.class) == null) {
- channel.writeAndFlush(new DefaultFileRegion(fc,
fc.position(), literal.size()));
- } else {
- channel.writeAndFlush(new ChunkedNioFile(fc, 8192));
+ // A SequencedLiteral's parts are written in one pass and flushed
once, each keeping its best representation.
+ if (literal instanceof SequencedLiteral sequencedLiteral) {
+ for (Literal part : sequencedLiteral.parts()) {
+ writePart(part);
}
} else {
- channel.writeAndFlush(new ChunkedStream(in));
+ writePart(literal);
+ }
+ channel.flush();
+ }
+ }
+
+ private void writePart(Literal literal) throws IOException {
+ Optional<byte[][]> bytesSequence = literal.asBytesSequence();
+ if (bytesSequence.isPresent()) {
+ channel.write(Unpooled.wrappedBuffer(bytesSequence.get()));
+ return;
+ }
+ InputStream in = literal.getInputStream();
+ if (in instanceof FileInputStream) {
+ FileChannel fc = ((FileInputStream) in).getChannel();
+ // Zero-copy is only possible if no SSL/TLS and no COMPRESS is in
place
+ //
+ // See JAMES-1305 and JAMES-1306
+ ChannelPipeline cp = channel.pipeline();
+ if (zeroCopy && cp.get(SslHandler.class) == null &&
cp.get(ZlibEncoder.class) == null) {
+ channel.write(new DefaultFileRegion(fc, fc.position(),
literal.size()));
+ } else {
+ channel.write(new ChunkedNioFile(fc, 8192));
}
+ } else {
+ channel.write(new ChunkedStream(in));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
