Reset the contentExpected flag when RSET is written

Motivation:

If the remote server returns a 4xx/5xx error in response to
a DATA command (or earlier command if using pipelining),
SmtpRequestEncoder can become stuck in an invalid state,
not allowing any requests to be sent.

This makes the channel unusable and the connection has to be closed,
or the encoder handler has to be replaced.

Modifications:

If a RSET command is written to the channel, the contentExpected
flag is set to false, and the RSET is written to the channel.

Result:

Sending a RSET command after a server 4xx/5xx error will make it
possible to use the current connection for new mail transactions.

codec-smtp/src/main/java/io/netty/handler/codec/smtp/SmtpRequestEncoder.java

@@ -47,13 +47,17 @@ public final class SmtpRequestEncoder extends MessageToMessageEncoder<Object> {
     @Override
     protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {
         if (msg instanceof SmtpRequest) {
+            final SmtpRequest req = (SmtpRequest) msg;
             if (contentExpected) {
-                throw new IllegalStateException("SmtpContent expected");
+                if (req.command().equals(SmtpCommand.RSET)) {
+                    contentExpected = false;
+                } else {
+                    throw new IllegalStateException("SmtpContent expected");
+                }
             }
             boolean release = true;
             final ByteBuf buffer = ctx.alloc().buffer();
             try {
-                final SmtpRequest req = (SmtpRequest) msg;
                 req.command().encode(buffer);
                 writeParameters(req.parameters(), buffer);
                 buffer.writeBytes(CRLF);

codec-smtp/src/test/java/io/netty/handler/codec/smtp/SmtpRequestEncoderTest.java

@@ -18,6 +18,7 @@ package io.netty.handler.codec.smtp;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.EncoderException;
 import io.netty.util.CharsetUtil;
 import org.junit.Test;
 
@@ -85,6 +86,29 @@ public class SmtpRequestEncoderTest {
                 new DefaultLastSmtpContent(Unpooled.copiedBuffer("Test\r\n", CharsetUtil.US_ASCII))));
         assertTrue(channel.finish());
 
+        assertEquals("DATA\r\nSubject: Test\r\n\r\nTest\r\n.\r\n", getWrittenString(channel));
+    }
+
+    @Test(expected = EncoderException.class)
+    public void testThrowsIfContentExpected() {
+        EmbeddedChannel channel = new EmbeddedChannel(new SmtpRequestEncoder());
+        assertTrue(channel.writeOutbound(SmtpRequests.data()));
+        channel.writeOutbound(SmtpRequests.noop());
+    }
+
+    @Test
+    public void testRsetClearsContentExpectedFlag() {
+        EmbeddedChannel channel = new EmbeddedChannel(new SmtpRequestEncoder());
+
+        assertTrue(channel.writeOutbound(SmtpRequests.data()));
+        assertTrue(channel.writeOutbound(SmtpRequests.rset()));
+        assertTrue(channel.writeOutbound(SmtpRequests.noop()));
+        assertTrue(channel.finish());
+
+        assertEquals("DATA\r\nRSET\r\nNOOP\r\n", getWrittenString(channel));
+    }
+
+    private static String getWrittenString(EmbeddedChannel channel) {
         ByteBuf written = Unpooled.buffer();
 
         for (;;) {
@@ -95,8 +119,11 @@ public class SmtpRequestEncoderTest {
             written.writeBytes(buffer);
             buffer.release();
         }
-        assertEquals("DATA\r\nSubject: Test\r\n\r\nTest\r\n.\r\n", written.toString(CharsetUtil.US_ASCII));
+
+        String writtenString = written.toString(CharsetUtil.US_ASCII);
         written.release();
+
+        return writtenString;
     }
 
     private static void testEncode(SmtpRequest request, String expected) {