SMTP pipelining support

k9mail-library/src/main/java/com/fsck/k9/mail/transport/smtp/SmtpTransport.java

@@ -16,9 +16,11 @@ import java.security.GeneralSecurityException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Queue;
 
 import android.support.annotation.VisibleForTesting;
 import android.text.TextUtils;
@@ -78,6 +80,8 @@ public class SmtpTransport extends Transport {
     private boolean isEnhancedStatusCodesProvided;
     private int largestAcceptableMessage;
     private boolean retryXoauthWithNewToken;
+    private boolean isPipeliningSupported;
+    private Queue<String> pipelinedCommand;
 
 
     public SmtpTransport(StoreConfig storeConfig, TrustedSocketFactory trustedSocketFactory,
@@ -165,6 +169,7 @@ public class SmtpTransport extends Transport {
 
             is8bitEncodingAllowed = extensions.containsKey("8BITMIME");
             isEnhancedStatusCodesProvided = extensions.containsKey("ENHANCEDSTATUSCODES");
+            isPipeliningSupported = extensions.containsKey("PIPELINING");
 
             if (connectionSecurity == ConnectionSecurity.STARTTLS_REQUIRED) {
                 if (extensions.containsKey("STARTTLS")) {
@@ -434,17 +439,37 @@ public class SmtpTransport extends Transport {
         Address[] from = message.getFrom();
         try {
             String fromAddress = from[0].getAddress();
-            if (is8bitEncodingAllowed) {
+            if (isPipeliningSupported) {
-                executeCommand("MAIL FROM:<%s> BODY=8BITMIME", fromAddress);
+                pipelinedCommand = new LinkedList<>();
+                if (is8bitEncodingAllowed) {
+                    executeCommandPipelined("MAIL FROM:<%s> BODY=8BITMIME", fromAddress);
+                } else {
+                    executeCommandPipelined("MAIL FROM:<%s>", fromAddress);
+                }
+
+                for (String address : addresses) {
+                    executeCommandPipelined("RCPT TO:<%s>", address);
+                }
+
+                executeCommandPipelined("DATA");
+                CommandResponse commandResponse = readPipelinedResponse();
+                if (commandResponse.replyCode != 354) {
+                    String replyText = TextUtils.join(" ", commandResponse.results);
+                    throw new NegativeSmtpReplyException(commandResponse.replyCode, replyText);
+                }
             } else {
-                executeCommand("MAIL FROM:<%s>", fromAddress);
+                if (is8bitEncodingAllowed) {
-            }
+                    executeCommand("MAIL FROM:<%s> BODY=8BITMIME", fromAddress);
+                } else {
+                    executeCommand("MAIL FROM:<%s>", fromAddress);
+                }
 
-            for (String address : addresses) {
+                for (String address : addresses) {
-                executeCommand("RCPT TO:<%s>", address);
+                    executeCommand("RCPT TO:<%s>", address);
-            }
+                }
 
-            executeCommand("DATA");
+                executeCommand("DATA");
+            }
 
             EOLConvertingOutputStream msgOut = new EOLConvertingOutputStream(
                     new LineWrapOutputStream(new SmtpDataStuffing(outputStream), 1000));
@@ -622,6 +647,64 @@ public class SmtpTransport extends Transport {
         return line;
     }
 
+    private void executeCommandPipelined(String format, Object... args) throws IOException {
+        if (format != null) {
+            String command = String.format(Locale.ROOT, format, args);
+            pipelinedCommand.add(command);
+            writeLine(command, false);
+        }
+    }
+
+    private CommandResponse readPipelinedResponse() throws IOException, MessagingException {
+        String responseLine = null;
+        List<String> results = null;
+        int noOfPipelinedResponse = pipelinedCommand.size();
+        while (noOfPipelinedResponse > 0) {
+            results = new ArrayList<>();
+            responseLine = readCommandResponseLine(results);
+            try {
+                responseLineToCommandResponse(responseLine, results);
+
+            } catch (NegativeSmtpReplyException exception) {
+                //continue reading response till DATA response .
+                Timber.d("SMTP <<< " + exception.getReplyCode() + exception.getReplyText());
+
+            } catch (MessagingException exception) {
+                //continue reading response till DATA response .
+
+            }
+            noOfPipelinedResponse-- ;
+        }
+        return responseLineToCommandResponse(responseLine, results);
+    }
+
+    private CommandResponse responseLineToCommandResponse(String line, List<String> results) throws MessagingException {
+        int length = line.length();
+        if (length < 1) {
+            throw new MessagingException("SMTP response is 0 length");
+        }
+
+        int replyCode = -1;
+        if (length >= 3) {
+            try {
+                replyCode = Integer.parseInt(line.substring(0, 3));
+            } catch (NumberFormatException e) { /* ignore */ }
+        }
+
+        char replyCodeCategory = line.charAt(0);
+        boolean isReplyCodeErrorCategory = (replyCodeCategory == '4') || (replyCodeCategory == '5');
+        if (isReplyCodeErrorCategory) {
+            if (isEnhancedStatusCodesProvided) {
+                throw buildEnhancedNegativeSmtpReplyException(replyCode, results);
+            } else {
+                String replyText = TextUtils.join(" ", results);
+                throw new NegativeSmtpReplyException(replyCode, replyText);
+            }
+        }
+
+        return new CommandResponse(replyCode, results);
+    }
+
 
     private void saslAuthLogin() throws MessagingException, IOException {
         try {

k9mail-library/src/test/java/com/fsck/k9/mail/transport/smtp/SmtpTransportTest.java

@@ -689,6 +689,78 @@ public class SmtpTransportTest {
         server.verifyInteractionCompleted();
     }
 
+    @Test
+    public void sendMessage_withPipelining() throws Exception {
+        Message message = getDefaultMessage();
+        MockSmtpServer server = createServerAndSetupForPlainAuthentication("PIPELINING");
+        server.expect("MAIL FROM:<user@localhost>");
+        server.expect("RCPT TO:<user2@localhost>");
+        server.expect("DATA");
+        server.output("250 OK");
+        server.output("250 OK");
+        server.output("354 End data with <CR><LF>.<CR><LF>");
+        server.expect("[message data]");
+        server.expect(".");
+        server.output("250 OK: queued as 12345");
+        server.expect("QUIT");
+        server.output("221 BYE");
+        server.closeConnection();
+        SmtpTransport transport = startServerAndCreateSmtpTransport(server);
+
+        transport.sendMessage(message);
+
+        server.verifyConnectionClosed();
+        server.verifyInteractionCompleted();
+    }
+
+    @Test
+    public void sendMessage_withoutPipelining() throws Exception {
+        Message message = getDefaultMessage();
+        MockSmtpServer server = createServerAndSetupForPlainAuthentication();
+        server.expect("MAIL FROM:<user@localhost>");
+        server.output("250 OK");
+        server.expect("RCPT TO:<user2@localhost>");
+        server.output("250 OK");
+        server.expect("DATA");
+        server.output("354 End data with <CR><LF>.<CR><LF>");
+        server.expect("[message data]");
+        server.expect(".");
+        server.output("250 OK: queued as 12345");
+        server.expect("QUIT");
+        server.output("221 BYE");
+        server.closeConnection();
+        SmtpTransport transport = startServerAndCreateSmtpTransport(server);
+
+        transport.sendMessage(message);
+
+        server.verifyConnectionClosed();
+        server.verifyInteractionCompleted();
+    }
+
+    @Test
+    public void sendMessagePipelining_withNegativeReply() throws Exception {
+        Message message = getDefaultMessage();
+        MockSmtpServer server = createServerAndSetupForPlainAuthentication("PIPELINING");
+        server.expect("MAIL FROM:<user@localhost>");
+        server.expect("RCPT TO:<user2@localhost>");
+        server.expect("DATA");
+        server.output("250 OK");
+        server.output("421 4.7.0 Temporary system problem");
+        server.output("354 End data with <CR><LF>.<CR><LF>");
+        server.expect("[message data]");
+        server.expect(".");
+        server.output("250 OK: queued as 12345");
+        server.expect("QUIT");
+        server.output("221 BYE");
+        server.closeConnection();
+        SmtpTransport transport = startServerAndCreateSmtpTransport(server);
+
+        transport.sendMessage(message);
+
+        server.verifyConnectionClosed();
+        server.verifyInteractionCompleted();
+    }
+
     private SmtpTransport startServerAndCreateSmtpTransport(MockSmtpServer server) throws IOException,
             MessagingException {
         return startServerAndCreateSmtpTransport(server, AuthType.PLAIN, ConnectionSecurity.NONE);