Only reclaim once at the end of the loop (#1958)

Same issue as fixed #1733 for ByteToMessageHandler.

### Motivation:

`NIOSingleStepByteToMessageProcessor` called out to the decoder's shouldReclaimBytes method after every parsing attempt, even if the decoder returned a value (which means continue in the `decodeLoop`).

That's quite pointless because we won't add any bytes into the buffer before we're trying the decoder again. Further it is a huge performance penalty for a use-cases in which we only consume small frames from the buffer. (Like reading data rows in a database client)

### Modifications:

- Only ask the decoder if we should reclaim bytes if the decoder actually is not able to process more frames from the input buffer.

Sources/NIOCore/SingleStepByteToMessageDecoder.swift

@@ -252,13 +252,6 @@ public final class NIOSingleStepByteToMessageProcessor<Decoder: NIOSingleStepByt
         assert(self._buffer != nil)
 
         func decodeOnce(buffer: inout ByteBuffer) throws -> Decoder.InboundOut? {
-            defer {
-                if buffer.readableBytes > 0 {
-                    if self.decoder.shouldReclaimBytes(buffer: buffer) {
-                        buffer.discardReadBytes()
-                    }
-                }
-            }
             if decodeMode == .normal {
                 return try self.decoder.decode(buffer: &buffer)
             } else {
@@ -275,7 +268,7 @@ public final class NIOSingleStepByteToMessageProcessor<Decoder: NIOSingleStepByt
         }
 
         // force unwrapping is safe because nil case is handled already and asserted above
-        if self._buffer!.readableBytes == 0 {
+        if self._buffer!.readerIndex > 0, self.decoder.shouldReclaimBytes(buffer: self._buffer!) {
             self._buffer!.discardReadBytes()
         }
     }

Tests/NIOPosixTests/SingleStepByteToMessageDecoderTest+XCTest.swift

@@ -36,6 +36,7 @@ extension NIOSingleStepByteToMessageDecoderTest {
                 ("testPayloadTooLarge", testPayloadTooLarge),
                 ("testPayloadTooLargeButHandlerOk", testPayloadTooLargeButHandlerOk),
                 ("testReentrancy", testReentrancy),
+                ("testWeDoNotCallShouldReclaimMemoryAsLongAsFramesAreProduced", testWeDoNotCallShouldReclaimMemoryAsLongAsFramesAreProduced),
            ]
    }
 }

Tests/NIOPosixTests/SingleStepByteToMessageDecoderTest.swift

@@ -13,7 +13,6 @@
 //===----------------------------------------------------------------------===//
 
 import XCTest
-import NIOPosix
 @testable import NIOCore
 import NIOEmbedded
 
@@ -449,4 +448,51 @@ public final class NIOSingleStepByteToMessageDecoderTest: XCTestCase {
         XCTAssertNoThrow(XCTAssertEqual("a", try channel.readInbound()))
         XCTAssertNoThrow(XCTAssertTrue(try channel.finish().isClean))
     }
+    
+    func testWeDoNotCallShouldReclaimMemoryAsLongAsFramesAreProduced() {
+        struct TestByteToMessageDecoder: NIOSingleStepByteToMessageDecoder {
+            typealias InboundOut = TestMessage
+            
+            enum TestMessage: Equatable {
+                case foo
+            }
+            
+            var lastByteBuffer: ByteBuffer?
+            var decodeHits = 0
+            var reclaimHits = 0
+            
+            mutating func decode(buffer: inout ByteBuffer) throws -> TestMessage? {
+                XCTAssertEqual(self.decodeHits * 3, buffer.readerIndex)
+                self.decodeHits += 1
+                guard buffer.readableBytes >= 3 else {
+                    return nil
+                }
+                buffer.moveReaderIndex(forwardBy: 3)
+                return .foo
+            }
+            
+            mutating func decodeLast(buffer: inout ByteBuffer, seenEOF: Bool) throws -> TestMessage? {
+                try self.decode(buffer: &buffer)
+            }
+            
+            mutating func shouldReclaimBytes(buffer: ByteBuffer) -> Bool {
+                self.reclaimHits += 1
+                return true
+            }
+        }
+        
+        let decoder = TestByteToMessageDecoder()
+        let processor = NIOSingleStepByteToMessageProcessor(decoder, maximumBufferSize: nil)
+        
+        let buffer = ByteBuffer(repeating: 0, count: 3001)
+        var callbackCount = 0
+        XCTAssertNoThrow(try processor.process(buffer: buffer) { _ in
+            callbackCount += 1
+        })
+        
+        XCTAssertEqual(callbackCount, 1000)
+        XCTAssertEqual(processor.decoder.decodeHits, 1001)
+        XCTAssertEqual(processor.decoder.reclaimHits, 1)
+        XCTAssertEqual(processor._buffer!.readableBytes, 1)
+    }
 }