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 2022/06/03 01:20:52 UTC
[james-project] 01/06: [PERF] Improve SMTP performance
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 675eeb7d251a593ec9f33d6cdafcbaa4fac80780
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon May 30 10:29:24 2022 +0700
[PERF] Improve SMTP performance
This change set proposes a 38% performance improvement for the
following workload (reduced CPU usage):
```
smtp-source -A -C100 -r 1 -l 500000 -m 5000 -s 20 -d -c -f any@other.com -M other.com -t firstname14.surname14@upn.integration-open-paas.org localhost:25
```
Rationals:
- Submitted emails are read line by line and a couple of session
scoped values where retrieved from session attachment for each line:
mail size, header reached, etc. This causes not cheap HashMap reads
for each email line. This can easily be replaced by
getters/setters on the SMTP session to save the map read/writes.
- The data was accessed with the following transformation chain:
ByteBuf -> NIO byte buffer -> byte[]. But expensive reference
counting is triggered by Netty under the hood. Doing directly
ByteBuf -> byte[] prevents this.
- SMTP commands metric was incremented on each body line (!) which
also incurs a performance hit.
- Immutable handlers where retrieved on each request while we could
save them to fields to avoid re-computation.
- Avoid re-parsing charset.
- Use Deque::peekFirst instead of Iterables::getFirst when looking
up for line overrides (same than IMAP).
- Compute debug logs only if needed.
- Fasten slightly parsing SMTP parameters
---
.../api/handler/AbstractProtocolHandlerChain.java | 16 ++++++++--
.../protocols/api/handler/CommandDispatcher.java | 14 ++------
.../api/handler/CommandHandlerResultLogger.java | 2 +-
.../james/protocols/api/handler/LineHandler.java | 4 +--
.../api/handler/ProtocolHandlerChain.java | 5 ++-
.../netty/BasicChannelInboundHandler.java | 22 ++++++-------
.../netty/LineHandlerUpstreamHandler.java | 8 ++---
.../apache/james/protocols/smtp/SMTPSession.java | 12 +++++++
.../james/protocols/smtp/SMTPSessionImpl.java | 37 ++++++++++++++++++++++
.../james/protocols/smtp/core/DataCmdHandler.java | 8 ++---
.../james/protocols/smtp/core/DataLineFilter.java | 4 +--
.../smtp/core/DataLineMessageHookHandler.java | 27 ++++------------
.../james/protocols/smtp/core/RcptCmdHandler.java | 8 +++--
.../smtp/core/ReceivedDataLineFilter.java | 11 +++----
.../smtp/core/SeparatingDataLineFilter.java | 37 ++++++++--------------
.../protocols/smtp/core/esmtp/AuthCmdHandler.java | 19 ++---------
.../smtp/core/esmtp/MailSizeEsmtpExtension.java | 20 ++++--------
.../protocols/smtp/utils/BaseFakeSMTPSession.java | 33 +++++++++++++++++++
.../server/core/MimeMessageInputStreamSource.java | 9 +++---
.../protocols/lib/ProtocolHandlerChainImpl.java | 11 +++++++
.../james/lmtpserver/DataLineLMTPHandler.java | 3 +-
.../james/lmtpserver/MailetContainerHandler.java | 5 +--
.../apache/james/lmtpserver/netty/LMTPServer.java | 12 +++++--
.../DataLineJamesMessageHookHandler.java | 19 ++++-------
.../james/smtpserver/ExtendedSMTPSession.java | 18 +++++++++++
.../james/smtpserver/JamesDataCmdHandler.java | 4 +--
.../org/apache/james/smtpserver/SMTPConstants.java | 2 --
.../smtpserver/jmx/SMTPCommandHandlerStats.java | 25 +++++----------
.../netty/SMTPChannelInboundHandler.java | 29 ++++++++++-------
.../apache/james/smtpserver/netty/SMTPServer.java | 2 +-
30 files changed, 244 insertions(+), 182 deletions(-)
diff --git a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/AbstractProtocolHandlerChain.java b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/AbstractProtocolHandlerChain.java
index 3abc7540f0..16d4823369 100644
--- a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/AbstractProtocolHandlerChain.java
+++ b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/AbstractProtocolHandlerChain.java
@@ -20,6 +20,7 @@ package org.apache.james.protocols.api.handler;
import java.util.LinkedList;
import java.util.List;
+import java.util.Optional;
import java.util.stream.Collectors;
@@ -40,13 +41,22 @@ public abstract class AbstractProtocolHandlerChain implements ProtocolHandlerCha
@Override
@SuppressWarnings("unchecked")
public <T> LinkedList<T> getHandlers(Class<T> type) {
- List<ProtocolHandler> handlers = getHandlers();
- return handlers.stream()
+ return getHandlers()
+ .stream()
.filter(type::isInstance)
.map(handler -> (T) handler)
.collect(Collectors.toCollection(LinkedList::new));
}
-
+
+ @Override
+ public <T> Optional<T> getFirstHandler(Class<T> type) {
+ return getHandlers()
+ .stream()
+ .filter(type::isInstance)
+ .map(handler -> (T) handler)
+ .findFirst();
+ }
+
/**
* ExtensibleHandler wiring. This should get called after the class was constructed
* and every stuff was set
diff --git a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/CommandDispatcher.java b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/CommandDispatcher.java
index 0c9e4b29ce..98b8ea5204 100644
--- a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/CommandDispatcher.java
+++ b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/CommandDispatcher.java
@@ -19,7 +19,6 @@
package org.apache.james.protocols.api.handler;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -57,7 +56,7 @@ public class CommandDispatcher<SessionT extends ProtocolSession> implements Exte
}
public CommandDispatcher() {
- this(Collections.<String>emptyList());
+ this(Collections.emptyList());
}
/**
@@ -127,7 +126,7 @@ public class CommandDispatcher<SessionT extends ProtocolSession> implements Exte
}
@Override
- public Response onLine(SessionT session, ByteBuffer line) {
+ public Response onLine(SessionT session, byte[] line) {
Request request;
try {
@@ -193,16 +192,9 @@ public class CommandDispatcher<SessionT extends ProtocolSession> implements Exte
/**
* Parse the line into a {@link Request}
*/
- protected Request parseRequest(SessionT session, ByteBuffer buffer) throws Exception {
+ protected Request parseRequest(SessionT session, byte[] line) throws Exception {
String curCommandName;
String curCommandArgument = null;
- byte[] line;
- if (buffer.hasArray()) {
- line = buffer.array();
- } else {
- line = new byte[buffer.remaining()];
- buffer.get(line);
- }
// This should be changed once we move to java6
String cmdString = new String(line, session.getCharset().name()).trim();
int spaceIndex = cmdString.indexOf(" ");
diff --git a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/CommandHandlerResultLogger.java b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/CommandHandlerResultLogger.java
index dd33bc95be..f6d88a1d90 100644
--- a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/CommandHandlerResultLogger.java
+++ b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/CommandHandlerResultLogger.java
@@ -32,7 +32,7 @@ public class CommandHandlerResultLogger implements ProtocolHandlerResultHandler<
@Override
public Response onResponse(ProtocolSession session, Response response, long executionTime, ProtocolHandler handler) {
- if (handler instanceof CommandHandler) {
+ if (handler instanceof CommandHandler && LOGGER.isDebugEnabled()) {
LOGGER.debug("{}: {}", handler.getClass().getName(), response);
}
return response;
diff --git a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/LineHandler.java b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/LineHandler.java
index f140f2bd0f..4834da50f0 100644
--- a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/LineHandler.java
+++ b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/LineHandler.java
@@ -19,8 +19,6 @@
package org.apache.james.protocols.api.handler;
-import java.nio.ByteBuffer;
-
import org.apache.james.protocols.api.ProtocolSession;
import org.apache.james.protocols.api.Response;
@@ -38,6 +36,6 @@ public interface LineHandler<SessionT extends ProtocolSession> extends ProtocolH
* @param buffer not null
* @return response or null
*/
- Response onLine(SessionT session, ByteBuffer buffer);
+ Response onLine(SessionT session, byte[] buffer);
}
diff --git a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/ProtocolHandlerChain.java b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/ProtocolHandlerChain.java
index 1edbeeed95..0433655d59 100644
--- a/protocols/api/src/main/java/org/apache/james/protocols/api/handler/ProtocolHandlerChain.java
+++ b/protocols/api/src/main/java/org/apache/james/protocols/api/handler/ProtocolHandlerChain.java
@@ -22,6 +22,7 @@
package org.apache.james.protocols.api.handler;
import java.util.LinkedList;
+import java.util.Optional;
/**
* Chain which can be used to get all Handlers for a given Class.
@@ -35,7 +36,9 @@ public interface ProtocolHandlerChain {
* @return a List of handlers
*/
<T> LinkedList<T> getHandlers(Class<T> type);
-
+
+ <T> Optional<T> getFirstHandler(Class<T> type);
+
/**
* Destroy the {@link ProtocolHandlerChain}. After this call it will not be usable anymore
*/
diff --git a/protocols/netty/src/main/java/org/apache/james/protocols/netty/BasicChannelInboundHandler.java b/protocols/netty/src/main/java/org/apache/james/protocols/netty/BasicChannelInboundHandler.java
index a84e7b08a4..fd52184fdc 100644
--- a/protocols/netty/src/main/java/org/apache/james/protocols/netty/BasicChannelInboundHandler.java
+++ b/protocols/netty/src/main/java/org/apache/james/protocols/netty/BasicChannelInboundHandler.java
@@ -43,8 +43,6 @@ import org.apache.james.util.MDCBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Iterables;
-
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
@@ -61,11 +59,13 @@ public class BasicChannelInboundHandler extends ChannelInboundHandlerAdapter imp
public static final AttributeKey<CommandDetectionSession> SESSION_ATTRIBUTE_KEY =
AttributeKey.valueOf("session");
- private final ProtocolMDCContextFactory mdcContextFactory;
protected final Protocol protocol;
protected final ProtocolHandlerChain chain;
protected final Encryption secure;
+ private final ProtocolMDCContextFactory mdcContextFactory;
private final Deque<ChannelInboundHandlerAdapter> behaviourOverrides = new ConcurrentLinkedDeque<>();
+ private final Optional<LineHandler> lineHandler;
+ protected final LinkedList<ProtocolHandlerResultHandler> resultHandlers;
public BasicChannelInboundHandler(ProtocolMDCContextFactory mdcContextFactory, Protocol protocol) {
this(mdcContextFactory, protocol, null);
@@ -76,6 +76,8 @@ public class BasicChannelInboundHandler extends ChannelInboundHandlerAdapter imp
this.protocol = protocol;
this.chain = protocol.getProtocolChain();
this.secure = secure;
+ this.lineHandler = chain.getFirstHandler(LineHandler.class);
+ this.resultHandlers = chain.getHandlers(ProtocolHandlerResultHandler.class);
}
@@ -147,7 +149,7 @@ public class BasicChannelInboundHandler extends ChannelInboundHandlerAdapter imp
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
- ChannelInboundHandlerAdapter override = Iterables.getFirst(behaviourOverrides, null);
+ ChannelInboundHandlerAdapter override = behaviourOverrides.peekFirst();
if (override != null) {
override.channelRead(ctx, msg);
return;
@@ -155,16 +157,14 @@ public class BasicChannelInboundHandler extends ChannelInboundHandlerAdapter imp
try (Closeable closeable = mdc(ctx).build()) {
ProtocolSession pSession = (ProtocolSession) ctx.channel().attr(SESSION_ATTRIBUTE_KEY).get();
- LinkedList<LineHandler> lineHandlers = chain.getHandlers(LineHandler.class);
- LinkedList<ProtocolHandlerResultHandler> resultHandlers = chain.getHandlers(ProtocolHandlerResultHandler.class);
-
-
- if (!lineHandlers.isEmpty()) {
+ if (lineHandler.isPresent()) {
ByteBuf buf = (ByteBuf) msg;
- LineHandler lHandler = lineHandlers.getLast();
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.getBytes(0, bytes);
+ LineHandler lHandler = lineHandler.get();
long start = System.currentTimeMillis();
- Response response = lHandler.onLine(pSession, buf.nioBuffer());
+ Response response = lHandler.onLine(pSession, bytes);
long executionTime = System.currentTimeMillis() - start;
for (ProtocolHandlerResultHandler resultHandler : resultHandlers) {
diff --git a/protocols/netty/src/main/java/org/apache/james/protocols/netty/LineHandlerUpstreamHandler.java b/protocols/netty/src/main/java/org/apache/james/protocols/netty/LineHandlerUpstreamHandler.java
index 36ab1cd1b1..fcc63c887b 100644
--- a/protocols/netty/src/main/java/org/apache/james/protocols/netty/LineHandlerUpstreamHandler.java
+++ b/protocols/netty/src/main/java/org/apache/james/protocols/netty/LineHandlerUpstreamHandler.java
@@ -47,14 +47,14 @@ public class LineHandlerUpstreamHandler<S extends ProtocolSession> extends Chann
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
-
- Response response = handler.onLine(session, buf.nioBuffer());
+ byte[] bytes = new byte[buf.readableBytes()];
+ buf.getBytes(0, bytes);
+ Response response = handler.onLine(session, bytes);
if (response != null) {
// TODO: This kind of sucks but I was not able to come up with something more elegant here
((ProtocolSessionImpl)session).getProtocolTransport().writeResponse(response, session);
}
-
- ((ByteBuf) msg).release();
+ buf.release();
}
}
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/SMTPSession.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/SMTPSession.java
index 3bfda3203b..595f06c352 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/SMTPSession.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/SMTPSession.java
@@ -81,5 +81,17 @@ public interface SMTPSession extends ProtocolSession {
boolean supportsOAuth();
+ long currentMessageSize();
+
+ void setCurrentMessageSize(long increment);
+
+ boolean headerComplete();
+
+ void setHeaderComplete(boolean value);
+
+ boolean messageFailed();
+
+ void setMessageFailed(boolean value);
+
}
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/SMTPSessionImpl.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/SMTPSessionImpl.java
index 814c4926a5..0c6c6eedc7 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/SMTPSessionImpl.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/SMTPSessionImpl.java
@@ -34,7 +34,10 @@ public class SMTPSessionImpl extends ProtocolSessionImpl implements SMTPSession
private static final Response FATAL_ERROR = new SMTPResponse(SMTPRetCode.LOCAL_ERROR, "Unable to process request").immutable();
private static final Response UNKNOWN_COMMAND_ERROR = new SMTPResponse(SMTPRetCode.SYNTAX_ERROR_COMMAND_UNRECOGNIZED, "Unable to process request: the command is unknown").immutable();
+ private long currentMessageSize = 0L;
private boolean relayingAllowed;
+ private boolean headerComplete = false;
+ private boolean messageFailed = false;
public SMTPSessionImpl(ProtocolTransport transport, SMTPConfiguration config) {
super(transport, config);
@@ -55,6 +58,10 @@ public class SMTPSessionImpl extends ProtocolSessionImpl implements SMTPSession
// start again with the old helo mode
currentHeloMode.ifPresent(heloMode -> setAttachment(CURRENT_HELO_MODE, heloMode, State.Connection));
+
+ currentMessageSize = 0L;
+ headerComplete = false;
+ messageFailed = false;
}
@Override
@@ -98,4 +105,34 @@ public class SMTPSessionImpl extends ProtocolSessionImpl implements SMTPSession
public SMTPConfiguration getConfiguration() {
return (SMTPConfiguration) config;
}
+
+ @Override
+ public long currentMessageSize() {
+ return currentMessageSize;
+ }
+
+ @Override
+ public void setCurrentMessageSize(long newSize) {
+ currentMessageSize = newSize;
+ }
+
+ @Override
+ public boolean headerComplete() {
+ return headerComplete;
+ }
+
+ @Override
+ public void setHeaderComplete(boolean value) {
+ headerComplete = value;
+ }
+
+ @Override
+ public boolean messageFailed() {
+ return messageFailed;
+ }
+
+ @Override
+ public void setMessageFailed(boolean value) {
+ messageFailed = value;
+ }
}
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataCmdHandler.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataCmdHandler.java
index 4373ff8bf3..11997bf67f 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataCmdHandler.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataCmdHandler.java
@@ -20,7 +20,6 @@ package org.apache.james.protocols.smtp.core;
import java.io.Closeable;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
@@ -64,9 +63,9 @@ public class DataCmdHandler implements CommandHandler<SMTPSession>, ExtensibleHa
public static final class DataConsumerLineHandler implements LineHandler<SMTPSession> {
@Override
- public SMTPResponse onLine(SMTPSession session, ByteBuffer line) {
+ public SMTPResponse onLine(SMTPSession session, byte[] line) {
// Discard everything until the end of DATA session
- if (line.remaining() == 3 && line.get() == 46) {
+ if (line.length == 3 && line[0] == 46) {
session.popLineHandler();
}
return null;
@@ -84,8 +83,7 @@ public class DataCmdHandler implements CommandHandler<SMTPSession>, ExtensibleHa
}
@Override
- public Response onLine(SMTPSession session, ByteBuffer line) {
- line.rewind();
+ public Response onLine(SMTPSession session, byte[] line) {
return filter.onLine(session, line, next);
}
}
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataLineFilter.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataLineFilter.java
index acec85e950..c9f61e225c 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataLineFilter.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataLineFilter.java
@@ -21,8 +21,6 @@
package org.apache.james.protocols.smtp.core;
-import java.nio.ByteBuffer;
-
import org.apache.james.protocols.api.Response;
import org.apache.james.protocols.api.handler.LineHandler;
import org.apache.james.protocols.api.handler.ProtocolHandler;
@@ -37,5 +35,5 @@ public interface DataLineFilter extends ProtocolHandler {
/**
* Handle line processing
*/
- Response onLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next);
+ Response onLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next);
}
\ No newline at end of file
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataLineMessageHookHandler.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataLineMessageHookHandler.java
index d7352da741..b9a3a9e233 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataLineMessageHookHandler.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/DataLineMessageHookHandler.java
@@ -23,7 +23,6 @@ package org.apache.james.protocols.smtp.core;
import java.io.IOException;
import java.io.OutputStream;
-import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.List;
@@ -58,16 +57,15 @@ public class DataLineMessageHookHandler implements DataLineFilter, ExtensibleHan
private List<?> rHooks;
@Override
- public Response onLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
+ public Response onLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
MailEnvelope env = session.getAttachment(DataCmdHandler.MAILENV, ProtocolSession.State.Transaction)
.orElseThrow(() -> new RuntimeException("'" + DataCmdHandler.MAILENV.asString() + "' has not been filled."));
OutputStream out = getMessageOutputStream(env);
try {
// 46 is "."
- // Stream terminated
- int c = line.get();
- if (line.remaining() == 2 && c == 46) {
+ // Stream terminated
+ if (line.length == 3 && line[0] == 46) {
out.flush();
out.close();
@@ -77,15 +75,14 @@ public class DataLineMessageHookHandler implements DataLineFilter, ExtensibleHan
return response;
// DotStuffing.
- } else if (c == 46 && line.get() == 46) {
- byte[] bline = readBytes(line);
- out.write(bline,1,bline.length - 1);
+ } else if (line[0] == 46 && line[1] == 46) {
+ out.write(line,1,line.length - 1);
// Standard write
} else {
// TODO: maybe we should handle the Header/Body recognition here
// and if needed let a filter to cache the headers to apply some
// transformation before writing them to output.
- out.write(readBytes(line));
+ out.write(line);
}
out.flush();
} catch (IOException e) {
@@ -105,18 +102,6 @@ public class DataLineMessageHookHandler implements DataLineFilter, ExtensibleHan
}
}
- private byte[] readBytes(ByteBuffer line) {
- line.rewind();
- byte[] bline;
- if (line.hasArray()) {
- bline = line.array();
- } else {
- bline = new byte[line.remaining()];
- line.get(bline);
- }
- return bline;
- }
-
protected Response processExtensions(SMTPSession session, MailEnvelope mail) {
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/RcptCmdHandler.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/RcptCmdHandler.java
index c91a40dff6..77c9d9cccd 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/RcptCmdHandler.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/RcptCmdHandler.java
@@ -223,9 +223,11 @@ public class RcptCmdHandler extends AbstractHookableCmdHandler<RcptHook> impleme
}
private Map<String, String> parseParameters(String rcptOptions) {
- return Splitter.on(' ').splitToStream(rcptOptions)
- .map(this::parseParameter)
- .collect(ImmutableMap.toImmutableMap(Pair::getKey, Pair::getValue));
+ ImmutableMap.Builder<String, String> result = ImmutableMap.builder();
+ for (String option: Splitter.on(' ').split(rcptOptions)) {
+ result.put(parseParameter(option));
+ }
+ return result.build();
}
private Pair<String, String> parseParameter(String rcptOption) {
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/ReceivedDataLineFilter.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/ReceivedDataLineFilter.java
index 08b5bc361b..725094dbc6 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/ReceivedDataLineFilter.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/ReceivedDataLineFilter.java
@@ -19,7 +19,6 @@
package org.apache.james.protocols.smtp.core;
import java.io.UnsupportedEncodingException;
-import java.nio.ByteBuffer;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
@@ -134,7 +133,7 @@ public class ReceivedDataLineFilter extends SeparatingDataLineFilter {
}
@Override
- protected Response onSeparatorLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
+ protected Response onSeparatorLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
if (getLocation() == Location.SUFFIX && !session.getAttachment(headersSuffixAdded, State.Transaction).isPresent()) {
session.setAttachment(headersSuffixAdded, Boolean.TRUE, State.Transaction);
return addHeaders(session, line, next);
@@ -143,7 +142,7 @@ public class ReceivedDataLineFilter extends SeparatingDataLineFilter {
}
@Override
- protected Response onHeadersLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
+ protected Response onHeadersLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
if (getLocation() == Location.PREFIX && !session.getAttachment(headersPrefixAdded, State.Transaction).isPresent()) {
session.setAttachment(headersPrefixAdded, Boolean.TRUE, State.Transaction);
return addHeaders(session, line, next);
@@ -156,7 +155,7 @@ public class ReceivedDataLineFilter extends SeparatingDataLineFilter {
*
* @return response
*/
- private Response addHeaders(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
+ private Response addHeaders(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
Response response;
for (Header header: headers(session)) {
response = header.transferTo(session, next);
@@ -195,7 +194,7 @@ public class ReceivedDataLineFilter extends SeparatingDataLineFilter {
/**
* Transfer the content of the {@link Header} to the given {@link LineHandler}.
*
- * This is done for each line of the {@link Header} until the end is reached or the {@link LineHandler#onLine(org.apache.james.protocols.api.ProtocolSession, ByteBuffer)}
+ * This is done for each line of the {@link Header} until the end is reached or the {@link LineHandler#onLine(org.apache.james.protocols.api.ProtocolSession, byte[])}
* return <code>non-null</code>
*
* @return response
@@ -212,7 +211,7 @@ public class ReceivedDataLineFilter extends SeparatingDataLineFilter {
} else {
line = MULTI_LINE_PREFIX + values.get(i);
}
- response = handler.onLine(session, ByteBuffer.wrap((line + session.getLineDelimiter()).getBytes(charset)));
+ response = handler.onLine(session, (line + session.getLineDelimiter()).getBytes(charset));
if (response != null) {
break;
}
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/SeparatingDataLineFilter.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/SeparatingDataLineFilter.java
index 2b18f3a0e3..4bb422bee1 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/SeparatingDataLineFilter.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/SeparatingDataLineFilter.java
@@ -18,10 +18,6 @@
****************************************************************/
package org.apache.james.protocols.smtp.core;
-import java.nio.ByteBuffer;
-
-import org.apache.james.protocols.api.ProtocolSession;
-import org.apache.james.protocols.api.ProtocolSession.State;
import org.apache.james.protocols.api.Response;
import org.apache.james.protocols.api.handler.LineHandler;
import org.apache.james.protocols.smtp.SMTPSession;
@@ -39,27 +35,22 @@ import org.apache.james.protocols.smtp.SMTPSession;
*
* Subclasses should override at least one of these methods:
* </br>
- * {@link #onHeadersLine(SMTPSession, ByteBuffer, LineHandler)}</br>
- * {@link #onSeparatorLine(SMTPSession, ByteBuffer, LineHandler)}</br>
- * {@link #onBodyLine(SMTPSession, ByteBuffer, LineHandler)}</br>
+ * {@link #onHeadersLine(SMTPSession, byte[], LineHandler)}</br>
+ * {@link #onSeparatorLine(SMTPSession, byte[], LineHandler)}</br>
+ * {@link #onBodyLine(SMTPSession, byte[], LineHandler)}</br>
*
*
*/
public abstract class SeparatingDataLineFilter implements DataLineFilter {
-
- private static final ProtocolSession.AttachmentKey<Boolean> HEADERS_COMPLETE = ProtocolSession.AttachmentKey.of("HEADERS_COMPLETE", Boolean.class);
-
@Override
- public final Response onLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
- if (!session.getAttachment(HEADERS_COMPLETE, State.Transaction).isPresent()) {
- if (line.remaining() == 2) {
- if (line.get() == '\r' && line.get() == '\n') {
- line.rewind();
+ public final Response onLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
+ if (!session.headerComplete()) {
+ if (line.length == 2) {
+ if (line[0] == '\r' && line[1] == '\n') {
Response response = onSeparatorLine(session, line, next);
- session.setAttachment(HEADERS_COMPLETE, Boolean.TRUE, State.Transaction);
+ session.setHeaderComplete(true);
return response;
}
- line.rewind();
}
return onHeadersLine(session, line, next);
}
@@ -70,42 +61,42 @@ public abstract class SeparatingDataLineFilter implements DataLineFilter {
/**
* Gets called when the separating line is received. This is the CLRF sequence.
*
- * This implementation just calls {@link LineHandler#onLine(org.apache.james.protocols.api.ProtocolSession, ByteBuffer)} but subclasses should override it if needed.
+ * This implementation just calls {@link LineHandler#onLine(org.apache.james.protocols.api.ProtocolSession, byte[])} but subclasses should override it if needed.
*
* @param session
* @param line
* @param next
* @return response
*/
- protected Response onSeparatorLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
+ protected Response onSeparatorLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
return next.onLine(session, line);
}
/**
* Gets called for each received line until the CRLF sequence was received.
*
- * This implementation just calls {@link LineHandler#onLine(org.apache.james.protocols.api.ProtocolSession, ByteBuffer)} but subclasses should override it if needed.
+ * This implementation just calls {@link LineHandler#onLine(org.apache.james.protocols.api.ProtocolSession, byte[])} but subclasses should override it if needed.
*
* @param session
* @param line
* @param next
* @return response
*/
- protected Response onHeadersLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
+ protected Response onHeadersLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
return next.onLine(session, line);
}
/**
* Gets called for each received line after the CRLF sequence was received.
*
- * This implementation just calls {@link LineHandler#onLine(org.apache.james.protocols.api.ProtocolSession, ByteBuffer)} but subclasses should override it if needed.
+ * This implementation just calls {@link LineHandler#onLine(org.apache.james.protocols.api.ProtocolSession, byte[])} but subclasses should override it if needed.
*
* @param session
* @param line
* @param next
* @return response
*/
- protected Response onBodyLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
+ protected Response onBodyLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
return next.onLine(session, line);
}
}
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/esmtp/AuthCmdHandler.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/esmtp/AuthCmdHandler.java
index c18d391b9d..c476c95139 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/esmtp/AuthCmdHandler.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/esmtp/AuthCmdHandler.java
@@ -21,8 +21,6 @@
package org.apache.james.protocols.smtp.core.esmtp;
-import java.io.UnsupportedEncodingException;
-import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Base64;
@@ -88,21 +86,8 @@ public class AuthCmdHandler
private abstract static class AbstractSMTPLineHandler implements LineHandler<SMTPSession> {
@Override
- public Response onLine(SMTPSession session, ByteBuffer line) {
- String charset = session.getCharset().name();
- try {
- byte[] l;
- if (line.hasArray()) {
- l = line.array();
- } else {
- l = new byte[line.remaining()];
- line.get(l);
- }
- return handleCommand(session, new String(l, charset));
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException("No " + charset + " support!");
- }
-
+ public Response onLine(SMTPSession session, byte[] line) {
+ return handleCommand(session, new String(line, session.getCharset()));
}
private Response handleCommand(SMTPSession session, String line) {
diff --git a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/esmtp/MailSizeEsmtpExtension.java b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/esmtp/MailSizeEsmtpExtension.java
index 9773208ba2..9de63f01ce 100644
--- a/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/esmtp/MailSizeEsmtpExtension.java
+++ b/protocols/smtp/src/main/java/org/apache/james/protocols/smtp/core/esmtp/MailSizeEsmtpExtension.java
@@ -19,7 +19,6 @@
package org.apache.james.protocols.smtp.core.esmtp;
-import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -52,7 +51,6 @@ public class MailSizeEsmtpExtension implements MailParametersHook, EhloExtension
private static final ProtocolSession.AttachmentKey<Integer> MESG_SIZE = ProtocolSession.AttachmentKey.of("MESG_SIZE", Integer.class); // The size of the
private static final ProtocolSession.AttachmentKey<Boolean> MESG_FAILED = ProtocolSession.AttachmentKey.of("MESG_FAILED", Boolean.class); // Message failed flag
- public static final ProtocolSession.AttachmentKey<Long> CURRENT_SIZE = ProtocolSession.AttachmentKey.of("CURRENT_SIZE", Long.class);
private static final String[] MAIL_PARAMS = { "SIZE" };
private static final HookResult SYNTAX_ERROR = HookResult.builder()
@@ -136,13 +134,12 @@ public class MailSizeEsmtpExtension implements MailParametersHook, EhloExtension
@Override
- public Response onLine(SMTPSession session, ByteBuffer line, LineHandler<SMTPSession> next) {
+ public Response onLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
Optional<Boolean> failed = session.getAttachment(MESG_FAILED, State.Transaction);
// If we already defined we failed and sent a reply we should simply
// wait for a CRLF.CRLF to be sent by the client.
if (failed.isPresent() && failed.get()) {
if (isDataTerminated(line)) {
- line.rewind();
next.onLine(session, line);
return new SMTPResponse(SMTPRetCode.QUOTA_EXCEEDED, "Quota exceeded");
} else {
@@ -150,15 +147,13 @@ public class MailSizeEsmtpExtension implements MailParametersHook, EhloExtension
}
} else {
if (isDataTerminated(line)) {
- line.rewind();
return next.onLine(session, line);
} else {
- line.rewind();
- Long newSize = session.getAttachment(CURRENT_SIZE, State.Transaction)
- .map(currentSize -> Long.valueOf(currentSize.intValue() + line.remaining()))
- .orElseGet(() -> Long.valueOf(line.remaining()));
+ Long newSize = Optional.ofNullable(session.currentMessageSize())
+ .map(currentSize -> Long.valueOf(currentSize.intValue() + line.length))
+ .orElseGet(() -> Long.valueOf(line.length));
- session.setAttachment(CURRENT_SIZE, newSize, State.Transaction);
+ session.setCurrentMessageSize(newSize);
if (session.getConfiguration().getMaxMessageSize() > 0 && newSize.intValue() > session.getConfiguration().getMaxMessageSize()) {
// Add an item to the state to suppress
@@ -169,15 +164,14 @@ public class MailSizeEsmtpExtension implements MailParametersHook, EhloExtension
return null;
} else {
- line.rewind();
return next.onLine(session, line);
}
}
}
}
- private boolean isDataTerminated(ByteBuffer line) {
- return line.remaining() == SINGLE_CHARACTER_LINE && line.get() == DOT_BYTE;
+ private boolean isDataTerminated(byte[] line) {
+ return line.length == SINGLE_CHARACTER_LINE && line[0] == DOT_BYTE;
}
@Override
diff --git a/protocols/smtp/src/test/java/org/apache/james/protocols/smtp/utils/BaseFakeSMTPSession.java b/protocols/smtp/src/test/java/org/apache/james/protocols/smtp/utils/BaseFakeSMTPSession.java
index b474f322ce..5089886f06 100644
--- a/protocols/smtp/src/test/java/org/apache/james/protocols/smtp/utils/BaseFakeSMTPSession.java
+++ b/protocols/smtp/src/test/java/org/apache/james/protocols/smtp/utils/BaseFakeSMTPSession.java
@@ -37,6 +37,10 @@ import org.apache.james.protocols.smtp.SMTPSession;
*/
public class BaseFakeSMTPSession implements SMTPSession {
+ private long currentMessageSize = 0L;
+ private boolean headerComplete = false;
+ private boolean messageFailed = false;
+
@Override
public boolean needsCommandInjectionDetection() {
throw new UnsupportedOperationException("Unimplemented Stub Method");
@@ -182,4 +186,33 @@ public class BaseFakeSMTPSession implements SMTPSession {
throw new UnsupportedOperationException("Unimplemented Stub Method");
}
+ @Override
+ public long currentMessageSize() {
+ return currentMessageSize;
+ }
+
+ @Override
+ public void setCurrentMessageSize(long increment) {
+ currentMessageSize = increment;
+ }
+
+ @Override
+ public boolean headerComplete() {
+ return headerComplete;
+ }
+
+ @Override
+ public void setHeaderComplete(boolean value) {
+ headerComplete = value;
+ }
+
+ @Override
+ public boolean messageFailed() {
+ return messageFailed;
+ }
+
+ @Override
+ public void setMessageFailed(boolean value) {
+ messageFailed = value;
+ }
}
diff --git a/server/container/core/src/main/java/org/apache/james/server/core/MimeMessageInputStreamSource.java b/server/container/core/src/main/java/org/apache/james/server/core/MimeMessageInputStreamSource.java
index 4419909030..f4875a75ea 100644
--- a/server/container/core/src/main/java/org/apache/james/server/core/MimeMessageInputStreamSource.java
+++ b/server/container/core/src/main/java/org/apache/james/server/core/MimeMessageInputStreamSource.java
@@ -187,14 +187,13 @@ public class MimeMessageInputStreamSource extends Disposable.LeakAware<MimeMessa
*/
@Override
public InputStream getInputStream() throws IOException {
- InputStream in;
if (getResource().getOut().isInMemory()) {
- in = new SharedByteArrayInputStream(getResource().getOut().getData());
+ return new SharedByteArrayInputStream(getResource().getOut().getData());
} else {
- in = new SharedFileInputStream(getResource().getOut().getFile());
+ InputStream in = new SharedFileInputStream(getResource().getOut().getFile());
+ getResource().streams.add(in);
+ return in;
}
- getResource().streams.add(in);
- return in;
}
/**
diff --git a/server/protocols/protocols-library/src/main/java/org/apache/james/protocols/lib/ProtocolHandlerChainImpl.java b/server/protocols/protocols-library/src/main/java/org/apache/james/protocols/lib/ProtocolHandlerChainImpl.java
index 944dc6c742..0979c3dd9c 100644
--- a/server/protocols/protocols-library/src/main/java/org/apache/james/protocols/lib/ProtocolHandlerChainImpl.java
+++ b/server/protocols/protocols-library/src/main/java/org/apache/james/protocols/lib/ProtocolHandlerChainImpl.java
@@ -20,6 +20,7 @@ package org.apache.james.protocols.lib;
import java.util.LinkedList;
import java.util.List;
+import java.util.Optional;
import java.util.stream.Collectors;
import org.apache.commons.configuration2.BaseHierarchicalConfiguration;
@@ -142,6 +143,16 @@ public class ProtocolHandlerChainImpl implements ProtocolHandlerChain {
.collect(Collectors.toCollection(LinkedList::new));
}
+ @Override
+ public <T> Optional<T> getFirstHandler(Class<T> type) {
+ for (Object handler: handlers) {
+ if (type.isInstance(handler)) {
+ return Optional.of((T) handler);
+ }
+ }
+ return Optional.empty();
+ }
+
/**
* Destroy all loaded {@link ProtocolHandler}
*/
diff --git a/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/DataLineLMTPHandler.java b/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/DataLineLMTPHandler.java
index f7da5d38ef..cddf7f5090 100644
--- a/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/DataLineLMTPHandler.java
+++ b/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/DataLineLMTPHandler.java
@@ -34,6 +34,7 @@ import org.apache.james.protocols.smtp.SMTPRetCode;
import org.apache.james.protocols.smtp.SMTPSession;
import org.apache.james.protocols.smtp.core.AbstractHookableCmdHandler;
import org.apache.james.protocols.smtp.dsn.DSNStatus;
+import org.apache.james.server.core.MimeMessageInputStreamSource;
import org.apache.james.smtpserver.DataLineJamesMessageHookHandler;
import org.apache.mailet.Mail;
@@ -46,7 +47,7 @@ public class DataLineLMTPHandler extends DataLineJamesMessageHookHandler {
@Override
- protected Response processExtensions(SMTPSession session, Mail mail) {
+ protected Response processExtensions(SMTPSession session, Mail mail, MimeMessageInputStreamSource mmiss) {
LMTPMultiResponse mResponse = null;
executeJamesMessageHooks(session, mail);
diff --git a/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/MailetContainerHandler.java b/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/MailetContainerHandler.java
index e0aa205193..62da6794af 100644
--- a/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/MailetContainerHandler.java
+++ b/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/MailetContainerHandler.java
@@ -37,6 +37,7 @@ import org.apache.james.protocols.smtp.core.AbstractHookableCmdHandler;
import org.apache.james.protocols.smtp.dsn.DSNStatus;
import org.apache.james.protocols.smtp.hook.HookResult;
import org.apache.james.protocols.smtp.hook.HookReturnCode;
+import org.apache.james.server.core.MimeMessageInputStreamSource;
import org.apache.james.smtpserver.DataLineJamesMessageHookHandler;
import org.apache.mailet.Mail;
@@ -90,11 +91,11 @@ public class MailetContainerHandler extends DataLineJamesMessageHookHandler {
}
@Override
- protected Response processExtensions(SMTPSession session, Mail mail) {
+ protected Response processExtensions(SMTPSession session, Mail mail, MimeMessageInputStreamSource mmiss) {
Collection<MailAddress> recipients = ImmutableList.copyOf(mail.getRecipients());
executeJamesMessageHooks(session, mail);
- if (recipients.size() == 0) {
+ if (recipients.isEmpty()) {
// Return 503 see https://datatracker.ietf.org/doc/html/rfc2033#section-4.2
AbstractHookableCmdHandler.calcDefaultSMTPResponse(HookResult.builder()
.hookReturnCode(HookReturnCode.ok())
diff --git a/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/netty/LMTPServer.java b/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/netty/LMTPServer.java
index c51fe75684..fcf15a105e 100644
--- a/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/netty/LMTPServer.java
+++ b/server/protocols/protocols-lmtp/src/main/java/org/apache/james/lmtpserver/netty/LMTPServer.java
@@ -26,6 +26,8 @@ import org.apache.commons.configuration2.tree.ImmutableNode;
import org.apache.james.lmtpserver.CoreCmdHandlerLoader;
import org.apache.james.lmtpserver.jmx.JMXHandlersLoader;
import org.apache.james.protocols.api.OidcSASLConfiguration;
+import org.apache.james.protocols.api.ProtocolSession;
+import org.apache.james.protocols.api.ProtocolTransport;
import org.apache.james.protocols.lib.handler.HandlersPackage;
import org.apache.james.protocols.lib.netty.AbstractProtocolAsyncServer;
import org.apache.james.protocols.lmtp.LMTPConfiguration;
@@ -33,6 +35,7 @@ import org.apache.james.protocols.netty.AbstractChannelPipelineFactory;
import org.apache.james.protocols.netty.ChannelHandlerFactory;
import org.apache.james.protocols.netty.LineDelimiterBasedChannelHandlerFactory;
import org.apache.james.protocols.smtp.SMTPProtocol;
+import org.apache.james.smtpserver.ExtendedSMTPSession;
import org.apache.james.smtpserver.netty.SMTPChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -141,8 +144,13 @@ public class LMTPServer extends AbstractProtocolAsyncServer implements LMTPServe
@Override
protected ChannelInboundHandlerAdapter createCoreHandler() {
- SMTPProtocol protocol = new SMTPProtocol(getProtocolHandlerChain(), lmtpConfig);
- return new SMTPChannelInboundHandler(protocol, lmtpMetrics, getExecutorGroup());
+ SMTPProtocol transport = new SMTPProtocol(getProtocolHandlerChain(), lmtpConfig) {
+ @Override
+ public ProtocolSession newSession(ProtocolTransport transport) {
+ return new ExtendedSMTPSession(lmtpConfig, transport);
+ }
+ };
+ return new SMTPChannelInboundHandler(transport, lmtpMetrics);
}
@Override
diff --git a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/DataLineJamesMessageHookHandler.java b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/DataLineJamesMessageHookHandler.java
index f71f11ece3..3a09c74848 100644
--- a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/DataLineJamesMessageHookHandler.java
+++ b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/DataLineJamesMessageHookHandler.java
@@ -23,7 +23,6 @@ import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.List;
@@ -70,13 +69,9 @@ public class DataLineJamesMessageHookHandler implements DataLineFilter, Extensib
private List<MessageHook> mHandlers;
@Override
- public Response onLine(SMTPSession session, ByteBuffer lineByteBuffer, LineHandler<SMTPSession> next) {
-
- byte[] line = new byte[lineByteBuffer.remaining()];
- lineByteBuffer.get(line, 0, line.length);
-
- MimeMessageInputStreamSource mmiss = session.getAttachment(SMTPConstants.DATA_MIMEMESSAGE_STREAMSOURCE, State.Transaction)
- .orElseThrow(() -> new RuntimeException("'" + SMTPConstants.DATA_MIMEMESSAGE_STREAMSOURCE.asString() + "' has not been filled."));
+ public Response onLine(SMTPSession session, byte[] line, LineHandler<SMTPSession> next) {
+ ExtendedSMTPSession extendedSMTPSession = (ExtendedSMTPSession) session;
+ MimeMessageInputStreamSource mmiss = extendedSMTPSession.getMimeMessageWriter();
try {
OutputStream out = mmiss.getWritableOutputStream();
@@ -84,7 +79,7 @@ public class DataLineJamesMessageHookHandler implements DataLineFilter, Extensib
// 46 is "."
// Stream terminated
if (line.length == 3 && line[0] == 46) {
- String mailName = MailImpl.getId();
+ String mailName = mmiss.getSourceId();
try (Closeable closeable = SMTPMDCContextFactory.forSession(session).addToContext("mail", mailName).build()) {
out.flush();
out.close();
@@ -104,7 +99,7 @@ public class DataLineJamesMessageHookHandler implements DataLineFilter, Extensib
try {
mail.setMessageContent(mmiss);
- Response response = processExtensions(session, mail);
+ Response response = processExtensions(session, mail, mmiss);
session.popLineHandler();
return response;
@@ -138,11 +133,9 @@ public class DataLineJamesMessageHookHandler implements DataLineFilter, Extensib
return null;
}
- protected Response processExtensions(SMTPSession session, Mail mail) {
+ protected Response processExtensions(SMTPSession session, Mail mail, MimeMessageInputStreamSource mmiss) {
if (mail != null && messageHandlers != null) {
try {
- MimeMessageInputStreamSource mmiss = session.getAttachment(SMTPConstants.DATA_MIMEMESSAGE_STREAMSOURCE, State.Transaction)
- .orElseThrow(() -> new RuntimeException("'" + SMTPConstants.DATA_MIMEMESSAGE_STREAMSOURCE.asString() + "' has not been filled."));
OutputStream out;
out = mmiss.getWritableOutputStream();
for (MessageHook rawHandler : mHandlers) {
diff --git a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/ExtendedSMTPSession.java b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/ExtendedSMTPSession.java
index 03cf9556a5..651b4434ab 100644
--- a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/ExtendedSMTPSession.java
+++ b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/ExtendedSMTPSession.java
@@ -18,9 +18,11 @@
****************************************************************/
package org.apache.james.smtpserver;
+import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.protocols.api.ProtocolTransport;
import org.apache.james.protocols.smtp.SMTPConfiguration;
import org.apache.james.protocols.smtp.SMTPSession;
+import org.apache.james.server.core.MimeMessageInputStreamSource;
import org.apache.james.smtpserver.netty.SMTPServer.SMTPHandlerConfigurationDataImpl;
/**
@@ -29,6 +31,7 @@ import org.apache.james.smtpserver.netty.SMTPServer.SMTPHandlerConfigurationData
public class ExtendedSMTPSession extends org.apache.james.protocols.smtp.SMTPSessionImpl {
private final SMTPConfiguration smtpConfiguration;
+ private MimeMessageInputStreamSource mimeMessageInputStreamSource;
public ExtendedSMTPSession(SMTPConfiguration smtpConfiguration, ProtocolTransport transport) {
super(transport, smtpConfiguration);
@@ -38,4 +41,19 @@ public class ExtendedSMTPSession extends org.apache.james.protocols.smtp.SMTPSes
public boolean verifyIdentity() {
return !(smtpConfiguration instanceof SMTPHandlerConfigurationDataImpl) || ((SMTPHandlerConfigurationDataImpl) smtpConfiguration).verifyIdentity();
}
+
+ public MimeMessageInputStreamSource getMimeMessageWriter() {
+ return mimeMessageInputStreamSource;
+ }
+
+ public void setMimeMessageInputStreamSource(MimeMessageInputStreamSource mimeMessageInputStreamSource) {
+ this.mimeMessageInputStreamSource = mimeMessageInputStreamSource;
+ }
+
+ @Override
+ public void resetState() {
+ super.resetState();
+ LifecycleUtil.dispose(mimeMessageInputStreamSource);
+ mimeMessageInputStreamSource = null;
+ }
}
diff --git a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/JamesDataCmdHandler.java b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/JamesDataCmdHandler.java
index 0ffa698a55..7dd478a26b 100644
--- a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/JamesDataCmdHandler.java
+++ b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/JamesDataCmdHandler.java
@@ -21,7 +21,6 @@ package org.apache.james.smtpserver;
import javax.inject.Inject;
import org.apache.james.metrics.api.MetricFactory;
-import org.apache.james.protocols.api.ProtocolSession.State;
import org.apache.james.protocols.smtp.SMTPResponse;
import org.apache.james.protocols.smtp.SMTPRetCode;
import org.apache.james.protocols.smtp.SMTPSession;
@@ -53,7 +52,8 @@ public class JamesDataCmdHandler extends DataCmdHandler {
protected SMTPResponse doDATA(SMTPSession session, String argument) {
try {
MimeMessageInputStreamSource mmiss = MimeMessageInputStreamSource.create(MailImpl.getId());
- session.setAttachment(SMTPConstants.DATA_MIMEMESSAGE_STREAMSOURCE, mmiss, State.Transaction);
+ ExtendedSMTPSession extendedSMTPSession = (ExtendedSMTPSession) session;
+ extendedSMTPSession.setMimeMessageInputStreamSource(mmiss);
} catch (Exception e) {
LOGGER.warn("Error creating mimemessagesource for incoming data", e);
return new SMTPResponse(SMTPRetCode.LOCAL_ERROR, "Unexpected error preparing to receive DATA.");
diff --git a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/SMTPConstants.java b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/SMTPConstants.java
index 1655b96db5..642ec45ae0 100644
--- a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/SMTPConstants.java
+++ b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/SMTPConstants.java
@@ -21,7 +21,6 @@ package org.apache.james.smtpserver;
import org.apache.james.protocols.api.ProtocolSession;
import org.apache.james.protocols.smtp.SMTPSession;
-import org.apache.james.server.core.MimeMessageInputStreamSource;
import org.apache.mailet.Mail;
import io.netty.util.AttributeKey;
@@ -31,7 +30,6 @@ import io.netty.util.AttributeKey;
*/
public interface SMTPConstants {
- ProtocolSession.AttachmentKey<MimeMessageInputStreamSource> DATA_MIMEMESSAGE_STREAMSOURCE = ProtocolSession.AttachmentKey.of("org.apache.james.core.DataCmdHandler.DATA_MIMEMESSAGE_STREAMSOURCE", MimeMessageInputStreamSource.class);
ProtocolSession.AttachmentKey<Mail> MAIL = ProtocolSession.AttachmentKey.of("MAIL", Mail.class);
AttributeKey<SMTPSession> SMTP_SESSION_ATTRIBUTE_KEY = AttributeKey.valueOf("SmtpSession");
diff --git a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/jmx/SMTPCommandHandlerStats.java b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/jmx/SMTPCommandHandlerStats.java
index 0cf4a9d80b..a011fe1d94 100644
--- a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/jmx/SMTPCommandHandlerStats.java
+++ b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/jmx/SMTPCommandHandlerStats.java
@@ -29,16 +29,12 @@ import org.apache.james.lifecycle.api.Disposable;
import org.apache.james.protocols.api.Response;
import org.apache.james.protocols.api.handler.CommandHandler;
import org.apache.james.protocols.lib.jmx.AbstractCommandHandlerStats;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Expose statistics for {@link CommandHandler} via JMX
*/
public class SMTPCommandHandlerStats extends AbstractCommandHandlerStats implements SMTPCommandHandlerStatsMBean, Disposable {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(SMTPCommandHandlerStats.class);
private final AtomicLong temp = new AtomicLong(0);
private final AtomicLong perm = new AtomicLong(0);
private final AtomicLong ok = new AtomicLong(0);
@@ -64,19 +60,14 @@ public class SMTPCommandHandlerStats extends AbstractCommandHandlerStats impleme
@Override
protected void incrementStats(Response response) {
- try {
- String code = response.getRetCode();
- char c = code.charAt(0);
- if (c == '5') {
- perm.incrementAndGet();
- } else if (c == '4') {
- temp.incrementAndGet();
- } else if (c == '2' || c == '3') {
- ok.incrementAndGet();
- }
-
- } catch (NumberFormatException e) {
- LOGGER.error("Error while reading SMTP return code {}", response.getRetCode(), e);
+ String code = response.getRetCode();
+ char c = code.charAt(0);
+ if (c == '5') {
+ perm.incrementAndGet();
+ } else if (c == '4') {
+ temp.incrementAndGet();
+ } else if (c == '2' || c == '3') {
+ ok.incrementAndGet();
}
}
diff --git a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/netty/SMTPChannelInboundHandler.java b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/netty/SMTPChannelInboundHandler.java
index 44a695473d..1d63b73601 100644
--- a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/netty/SMTPChannelInboundHandler.java
+++ b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/netty/SMTPChannelInboundHandler.java
@@ -18,17 +18,19 @@
****************************************************************/
package org.apache.james.smtpserver.netty;
+import java.util.Optional;
+
import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.protocols.api.Protocol;
import org.apache.james.protocols.api.ProtocolSession.State;
+import org.apache.james.protocols.api.handler.ProtocolHandlerResultHandler;
import org.apache.james.protocols.netty.BasicChannelInboundHandler;
import org.apache.james.protocols.netty.Encryption;
-import org.apache.james.protocols.smtp.SMTPSession;
import org.apache.james.protocols.smtp.core.SMTPMDCContextFactory;
+import org.apache.james.smtpserver.ExtendedSMTPSession;
import org.apache.james.smtpserver.SMTPConstants;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.util.concurrent.EventExecutorGroup;
/**
* {@link BasicChannelInboundHandler} which is used by the SMTPServer
@@ -37,14 +39,23 @@ public class SMTPChannelInboundHandler extends BasicChannelInboundHandler {
private final SmtpMetrics smtpMetrics;
- public SMTPChannelInboundHandler(Protocol protocol, Encryption encryption, SmtpMetrics smtpMetrics, EventExecutorGroup eventExecutorGroup) {
+ public SMTPChannelInboundHandler(Protocol protocol, Encryption encryption, SmtpMetrics smtpMetrics) {
super(new SMTPMDCContextFactory(), protocol, encryption);
this.smtpMetrics = smtpMetrics;
+ this.resultHandlers.add(recordCommandCount(smtpMetrics));
}
- public SMTPChannelInboundHandler(Protocol protocol, SmtpMetrics smtpMetrics, EventExecutorGroup eventExecutorGroup) {
+ public SMTPChannelInboundHandler(Protocol protocol, SmtpMetrics smtpMetrics) {
super(new SMTPMDCContextFactory(), protocol);
this.smtpMetrics = smtpMetrics;
+ this.resultHandlers.add(recordCommandCount(smtpMetrics));
+ }
+
+ private ProtocolHandlerResultHandler recordCommandCount(SmtpMetrics smtpMetrics) {
+ return (session, response, executionTime, handler) -> {
+ smtpMetrics.getCommandsMetric().increment();
+ return response;
+ };
}
@Override
@@ -53,12 +64,6 @@ public class SMTPChannelInboundHandler extends BasicChannelInboundHandler {
smtpMetrics.getConnectionMetric().increment();
}
- @Override
- public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
- super.channelRead(ctx, msg);
- smtpMetrics.getCommandsMetric().increment();
- }
-
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
@@ -71,11 +76,11 @@ public class SMTPChannelInboundHandler extends BasicChannelInboundHandler {
@Override
protected void cleanup(ChannelHandlerContext ctx) {
// Make sure we dispose everything on exit on session close
- SMTPSession smtpSession = ctx.channel().attr(SMTPConstants.SMTP_SESSION_ATTRIBUTE_KEY).get();
+ ExtendedSMTPSession smtpSession = (ExtendedSMTPSession) ctx.channel().attr(SMTPConstants.SMTP_SESSION_ATTRIBUTE_KEY).get();
if (smtpSession != null) {
smtpSession.getAttachment(SMTPConstants.MAIL, State.Transaction).ifPresent(LifecycleUtil::dispose);
- smtpSession.getAttachment(SMTPConstants.DATA_MIMEMESSAGE_STREAMSOURCE, State.Transaction).ifPresent(LifecycleUtil::dispose);
+ Optional.ofNullable(smtpSession.getMimeMessageWriter()).ifPresent(LifecycleUtil::dispose);
}
super.cleanup(ctx);
diff --git a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/netty/SMTPServer.java b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/netty/SMTPServer.java
index 4f079785b0..1468edc620 100644
--- a/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/netty/SMTPServer.java
+++ b/server/protocols/protocols-smtp/src/main/java/org/apache/james/smtpserver/netty/SMTPServer.java
@@ -385,7 +385,7 @@ public class SMTPServer extends AbstractProtocolAsyncServer implements SMTPServe
@Override
protected ChannelInboundHandlerAdapter createCoreHandler() {
- return new SMTPChannelInboundHandler(transport, getEncryption(), smtpMetrics, getExecutorGroup());
+ return new SMTPChannelInboundHandler(transport, getEncryption(), smtpMetrics);
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org