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/26 03:11:25 UTC
(james-project) 02/02: JAMES-3977 Allow ChannelImapResponseWriter to work with sequence of bytes
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 672898bc579e876cb0705c8d713ee465822d91fb
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Fri Mar 22 11:43:04 2024 +0100
JAMES-3977 Allow ChannelImapResponseWriter to work with sequence of bytes
This allows accurate representation by the netty stack of the memory used.
---
.../apache/james/mailbox/model/ByteContent.java | 8 ++++++++
.../org/apache/james/mailbox/model/Content.java | 7 ++++++-
.../mailbox/model/HeaderAndBodyByteContent.java | 9 +++++++++
.../jpa/mail/model/openjpa/JPAMailboxMessage.java | 23 ++++++++++++++++++++++
.../james/mailbox/store/MessageResultImpl.java | 5 +++++
.../store/mail/model/DelegatingMailboxMessage.java | 6 ++++++
.../james/mailbox/store/mail/model/Message.java | 12 +++++++++++
.../store/mail/model/impl/SimpleMessage.java | 6 ++++++
.../store/streaming/InputStreamContent.java | 12 ++++++++++-
.../james/imap/message/BytesBackedLiteral.java | 8 ++++++++
.../org/apache/james/imap/message/Literal.java | 5 +++++
.../imap/processor/fetch/ContentBodyElement.java | 7 ++++++-
.../netty/ChannelImapResponseWriter.java | 4 ++++
13 files changed, 109 insertions(+), 3 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/model/ByteContent.java b/mailbox/api/src/main/java/org/apache/james/mailbox/model/ByteContent.java
index a14a7daa8a..f10f348c46 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/model/ByteContent.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/model/ByteContent.java
@@ -25,6 +25,7 @@ package org.apache.james.mailbox.model;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import java.util.Optional;
import org.reactivestreams.Publisher;
@@ -56,4 +57,11 @@ public final class ByteContent implements Content {
return Flux.just(contents)
.map(ByteBuffer::wrap);
}
+
+ @Override
+ public Optional<byte[][]> asBytesSequence() {
+ byte[][] answer = new byte[1][];
+ answer[0] = contents;
+ return Optional.of(answer);
+ }
}
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/model/Content.java b/mailbox/api/src/main/java/org/apache/james/mailbox/model/Content.java
index abee852e0a..ef7a989fef 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/model/Content.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/model/Content.java
@@ -22,6 +22,7 @@ package org.apache.james.mailbox.model;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import java.util.Optional;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.util.ReactorUtils;
@@ -41,7 +42,11 @@ public interface Content {
* Return the content as {@link InputStream}
*/
InputStream getInputStream() throws IOException;
-
+
+ default Optional<byte[][]> asBytesSequence() {
+ return Optional.empty();
+ }
+
/**
* Size (in octets) of the content.
*
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/model/HeaderAndBodyByteContent.java b/mailbox/api/src/main/java/org/apache/james/mailbox/model/HeaderAndBodyByteContent.java
index bbd1740d6e..eb6c607c51 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/model/HeaderAndBodyByteContent.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/model/HeaderAndBodyByteContent.java
@@ -23,6 +23,7 @@ import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.io.SequenceInputStream;
import java.nio.ByteBuffer;
+import java.util.Optional;
import org.reactivestreams.Publisher;
@@ -53,6 +54,14 @@ public final class HeaderAndBodyByteContent implements Content {
new ByteArrayInputStream(body));
}
+ @Override
+ public Optional<byte[][]> asBytesSequence() {
+ byte[][] answer = new byte[2][];
+ answer[0] = headers;
+ answer[1] = body;
+ return Optional.of(answer);
+ }
+
@Override
public Publisher<ByteBuffer> reactiveBytes() {
return Flux.concat(Flux.just(headers).map(ByteBuffer::wrap), Flux.just(body).map(ByteBuffer::wrap));
diff --git a/mailbox/jpa/src/main/java/org/apache/james/mailbox/jpa/mail/model/openjpa/JPAMailboxMessage.java b/mailbox/jpa/src/main/java/org/apache/james/mailbox/jpa/mail/model/openjpa/JPAMailboxMessage.java
index 90dd699584..776cbef69e 100644
--- a/mailbox/jpa/src/main/java/org/apache/james/mailbox/jpa/mail/model/openjpa/JPAMailboxMessage.java
+++ b/mailbox/jpa/src/main/java/org/apache/james/mailbox/jpa/mail/model/openjpa/JPAMailboxMessage.java
@@ -22,6 +22,7 @@ import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Date;
+import java.util.Optional;
import jakarta.mail.Flags;
import jakarta.persistence.Basic;
@@ -123,4 +124,26 @@ public class JPAMailboxMessage extends AbstractJPAMailboxMessage {
public MailboxMessage copy(Mailbox mailbox) throws MailboxException {
return new JPAMailboxMessage(JPAMailbox.from(mailbox), getUid(), getModSeq(), this);
}
+
+ @Override
+ public Optional<byte[][]> getBodyBytes() {
+ byte[][] answer = new byte[1][];
+ answer[0] = body;
+ return Optional.of(answer);
+ }
+
+ @Override
+ public Optional<byte[][]> getFullBytes() {
+ byte[][] answer = new byte[2][];
+ answer[0] = header;
+ answer[1] = body;
+ return Optional.of(answer);
+ }
+
+ @Override
+ public Optional<byte[][]> getHeadersBytes() {
+ byte[][] answer = new byte[1][];
+ answer[0] = header;
+ return Optional.of(answer);
+ }
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/MessageResultImpl.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/MessageResultImpl.java
index ce2a46cbb6..60a164e140 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/MessageResultImpl.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/MessageResultImpl.java
@@ -364,6 +364,11 @@ public class MessageResultImpl implements MessageResult {
return msg.getHeaderOctets();
}
+ @Override
+ public Optional<byte[][]> asBytesSequence() {
+ return msg.getHeadersBytes();
+ }
+
@Override
public Iterator<Header> headers() throws MailboxException {
if (headers == null) {
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 c37516bbba..29726f4aa0 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
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.util.Date;
import java.util.List;
+import java.util.Optional;
import jakarta.mail.Flags;
@@ -126,4 +127,9 @@ public abstract class DelegatingMailboxMessage implements MailboxMessage {
public List<MessageAttachmentMetadata> getAttachments() {
return message.getAttachments();
}
+
+ @Override
+ public Optional<byte[][]> getFullBytes() {
+ return message.getFullBytes();
+ }
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/Message.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/Message.java
index a5f37bb198..bb45474030 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/Message.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/mail/model/Message.java
@@ -25,6 +25,7 @@ import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.List;
+import java.util.Optional;
import org.apache.james.mailbox.model.MessageAttachmentMetadata;
import org.apache.james.mailbox.model.MessageId;
@@ -50,6 +51,10 @@ public interface Message {
*/
InputStream getBodyContent() throws IOException;
+ default Optional<byte[][]> getBodyBytes() {
+ return Optional.empty();
+ }
+
default Publisher<ByteBuffer> getBodyContentReactive() {
try {
return ReactorUtils.toChunks(getBodyContent(), BUFFER_SIZE)
@@ -103,6 +108,10 @@ public interface Message {
*/
InputStream getHeaderContent() throws IOException;
+ default Optional<byte[][]> getHeadersBytes() {
+ return Optional.empty();
+ }
+
default Publisher<ByteBuffer> getHeaderContentReactive() {
try {
@@ -121,6 +130,9 @@ public interface Message {
*/
InputStream getFullContent() throws IOException;
+ default Optional<byte[][]> getFullBytes() {
+ return Optional.empty();
+ }
default Publisher<ByteBuffer> getFullContentReactive() {
try {
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 add87a967d..ffe9b02ce2 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
@@ -23,6 +23,7 @@ import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.List;
+import java.util.Optional;
import org.apache.commons.io.input.BoundedInputStream;
import org.apache.james.mailbox.exception.MailboxException;
@@ -117,6 +118,11 @@ public class SimpleMessage implements Message {
return content.getInputStream();
}
+ @Override
+ public Optional<byte[][]> getFullBytes() {
+ return content.asBytesSequence();
+ }
+
@Override
public Properties getProperties() {
return properties;
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/streaming/InputStreamContent.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/streaming/InputStreamContent.java
index 5a2662095a..92be0a277d 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/streaming/InputStreamContent.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/streaming/InputStreamContent.java
@@ -21,6 +21,7 @@ package org.apache.james.mailbox.store.streaming;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import java.util.Optional;
import org.apache.james.mailbox.model.Content;
import org.apache.james.mailbox.store.mail.model.Message;
@@ -64,7 +65,16 @@ public final class InputStreamContent implements Content {
default:
return m.getBodyContent();
}
-
+ }
+
+ @Override
+ public Optional<byte[][]> asBytesSequence() {
+ switch (type) {
+ case FULL:
+ return m.getFullBytes();
+ default:
+ return m.getBodyBytes();
+ }
}
@Override
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/message/BytesBackedLiteral.java b/protocols/imap/src/main/java/org/apache/james/imap/message/BytesBackedLiteral.java
index d5ff6b8b46..abca5df42a 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/message/BytesBackedLiteral.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/message/BytesBackedLiteral.java
@@ -22,6 +22,7 @@ package org.apache.james.imap.message;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.util.Optional;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.output.UnsynchronizedByteArrayOutputStream;
@@ -60,4 +61,11 @@ public class BytesBackedLiteral implements Literal {
public InputStream getInputStream() {
return new ByteArrayInputStream(content);
}
+
+ @Override
+ public Optional<byte[][]> asBytesSequence() {
+ byte[][] answer = new byte[1][];
+ answer[0] = content;
+ return Optional.of(answer);
+ }
}
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/message/Literal.java b/protocols/imap/src/main/java/org/apache/james/imap/message/Literal.java
index 94851f07f7..9ef6fba741 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/message/Literal.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/message/Literal.java
@@ -21,6 +21,7 @@ package org.apache.james.imap.message;
import java.io.IOException;
import java.io.InputStream;
+import java.util.Optional;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.model.Content;
@@ -42,6 +43,10 @@ public interface Literal {
*/
InputStream getInputStream() throws IOException;
+ default Optional<byte[][]> asBytesSequence() {
+ return Optional.empty();
+ }
+
default Content asMailboxContent() {
Literal literal = this;
return new Content() {
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/ContentBodyElement.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/ContentBodyElement.java
index 5cbf722047..1aef078f8f 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/ContentBodyElement.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/fetch/ContentBodyElement.java
@@ -24,6 +24,7 @@ package org.apache.james.imap.processor.fetch;
import java.io.IOException;
import java.io.InputStream;
+import java.util.Optional;
import org.apache.james.imap.message.response.FetchResponse.BodyElement;
import org.apache.james.mailbox.exception.MailboxException;
@@ -51,7 +52,11 @@ class ContentBodyElement implements BodyElement {
throw new IOException("Unable to get size for body element", e);
}
}
-
+
+ @Override
+ public Optional<byte[][]> asBytesSequence() {
+ return content.asBytesSequence();
+ }
@Override
public InputStream getInputStream() throws IOException {
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 0a4f53ce9c..55734944b8 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
@@ -77,6 +77,10 @@ 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();
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org