Adds a line-based frame decoder that can split received buffers on line endings.

Motivation:

As per https://github.com/apple/swift-nio/issues/473

Modifications:

Added a new decoder (LineBasedFrameDecoder) that splits incoming buffers on line end characters.

Result:

Received buffers will be split on line end character(s) ('\n' or '\r\n'), with these characters
stripped in the resulting buffers.

CONTRIBUTORS.txt

@@ -16,3 +16,4 @@ needs to be listed here.
 - Johannes Weiß <johannesweiss@apple.com>
 - Norman Maurer <norman_maurer@apple.com>
 - Tom Doron <tomerd@apple.com>
+- Ludovic Dewailly <ldewailly@apple.com>

README.md

@@ -30,3 +30,4 @@ functionality.
 
 - [`QuiescingHelper`](Sources/NIOExtras/QuiescingHelper.swift): Helps to quiesce
   a server by notifying user code when all previously open connections have closed.
+- [`LineBasedFrameDecoder`](Sources/NIOExtras/LineBasedFrameDecoder.swift) Splits incoming `ByteBuffer`s on line endings.

Sources/NIOExtras/LineBasedFrameDecoder.swift

@@ -0,0 +1,86 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2017-2018 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIO
+
+///
+/// A decoder that splits incoming `ByteBuffer`s around line end
+/// character(s) (`'\n'` or `'\r\n'`).
+///
+/// Let's, for example, consider the following received buffer:
+///
+///     +----+-------+------------+
+///     | AB | C\nDE | F\r\nGHI\n |
+///     +----+-------+------------+
+///
+/// A instance of `LineBasedFrameDecoder` will split this buffer
+/// as follows:
+///
+///     +-----+-----+-----+
+///     | ABC | DEF | GHI |
+///     +-----+-----+-----+
+///
+public class LineBasedFrameDecoder: ByteToMessageDecoder {
+    
+    public typealias InboundIn = ByteBuffer
+    public typealias InboundOut = ByteBuffer
+    public var cumulationBuffer: ByteBuffer?
+    // we keep track of the last scan end index if we didn't find the delimiter
+    private var lastIndex: ByteBufferView.Index? = nil
+    
+    public init() { }
+    
+    public func decode(ctx: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
+        if let frame = try findNextFrame(buffer: &buffer) {
+            ctx.fireChannelRead(wrapInboundOut(frame))
+            return .continue
+        } else {
+            return .needMoreData
+        }
+    }
+    
+    private func findNextFrame(buffer: inout ByteBuffer) throws -> ByteBuffer? {
+        var view = buffer.readableBytesView
+        // get the view's true start, end indexes
+        let _startIndex = view.startIndex
+        let _endIndex = view.endIndex
+        // start where we left off last scan or from the beginning
+        let firstIndex = min(self.lastIndex ?? _startIndex, _endIndex)
+        view = view.dropFirst(firstIndex)
+        while !view.isEmpty {
+            if view.starts(with: "\n".utf8) {
+                let length = view.startIndex - _startIndex
+                // check if the line ends with carriage return, if so drop it
+                let dropCarriageReturn = length > 0
+                        && buffer.getBytes(at: buffer.readerIndex + length - 1, length: 1) == [0x0D]
+                let buff = buffer.readSlice(length: dropCarriageReturn ? length - 1 : length)
+                // drop the delimiter (and trailing carriage return if appicable)
+                buffer.moveReaderIndex(forwardBy: dropCarriageReturn ? 2 : 1)
+                // reset the last scan start index since we found a line
+                self.lastIndex = nil
+                return buff
+            }
+            view = view.dropFirst(1)
+        }
+        // next scan we start where we stopped
+        self.lastIndex = max(0, _endIndex - 1)
+        return nil
+    }
+    
+    public func handlerRemoved(ctx: ChannelHandlerContext) {
+        if let buffer = cumulationBuffer, buffer.readableBytes > 0 {
+            ctx.fireErrorCaught(NIOExtrasErrors.LeftOverBytesError(leftOverBytes: buffer))
+        }
+    }
+}

Tests/LinuxMain.swift

@@ -27,6 +27,7 @@ import XCTest
 
    XCTMain([
          testCase(FixedLengthFrameDecoderTest.allTests),
+         testCase(LineBasedFrameDecoderTest.allTests),
          testCase(QuiescingHelperTest.allTests),
     ])
 #endif

Tests/NIOExtrasTests/LineBasedFramceDecoderTest+XCTest.swift

@@ -0,0 +1,37 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2017-2018 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+//
+// LineBasedFramceDecoderTest+XCTest.swift
+//
+import XCTest
+
+///
+/// NOTE: This file was generated by generate_linux_tests.rb
+///
+/// Do NOT edit this file directly as it will be regenerated automatically when needed.
+///
+
+extension LineBasedFrameDecoderTest {
+
+   static var allTests : [(String, (LineBasedFrameDecoderTest) -> () throws -> Void)] {
+      return [
+                ("testDecodeOneCharacterAtATime", testDecodeOneCharacterAtATime),
+                ("testRemoveHandlerWhenBufferIsNotEmpty", testRemoveHandlerWhenBufferIsNotEmpty),
+                ("testRemoveHandlerWhenBufferIsEmpty", testRemoveHandlerWhenBufferIsEmpty),
+                ("testEmptyLine", testEmptyLine),
+                ("testEmptyBuffer", testEmptyBuffer),
+           ]
+   }
+}
+

Tests/NIOExtrasTests/LineBasedFramceDecoderTest.swift

@@ -0,0 +1,106 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2017-2018 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import XCTest
+import NIO
+import NIOExtras
+
+class LineBasedFrameDecoderTest: XCTestCase {
+    
+    func testDecodeOneCharacterAtATime() throws {
+        let channel = EmbeddedChannel()
+        try channel.pipeline.add(handler: LineBasedFrameDecoder()).wait()
+        
+        let message = "abcdefghij\r"
+        // we write one character at a time
+        try message.forEach {
+            var buffer = channel.allocator.buffer(capacity: 1)
+            buffer.write(string: "\($0)")
+            XCTAssertFalse(try channel.writeInbound(buffer))
+        }
+        // let's add `\n`
+        var buffer = channel.allocator.buffer(capacity: 1)
+        buffer.write(string: "\n")
+        XCTAssertTrue(try channel.writeInbound(buffer))
+        
+        var outputBuffer: ByteBuffer? = channel.readInbound()
+        XCTAssertEqual("abcdefghij", outputBuffer?.readString(length: 10))
+        XCTAssertFalse(try channel.finish())
+    }
+    
+    func testRemoveHandlerWhenBufferIsNotEmpty() throws {
+        let channel = EmbeddedChannel()
+        let handler = LineBasedFrameDecoder()
+        try channel.pipeline.add(handler: handler).wait()
+        
+        var buffer = channel.allocator.buffer(capacity: 8)
+        buffer.write(string: "foo\r\nbar")
+        XCTAssertTrue(try channel.writeInbound(buffer))
+        var outputBuffer: ByteBuffer? = channel.readInbound()
+        XCTAssertEqual(3, outputBuffer?.readableBytes)
+        XCTAssertEqual("foo", outputBuffer?.readString(length: 3))
+        
+        _ = try channel.pipeline.remove(handler: handler).wait()
+        XCTAssertThrowsError(try channel.throwIfErrorCaught()) { error in
+            guard let error = error as? NIOExtrasErrors.LeftOverBytesError else {
+                XCTFail()
+                return
+            }
+            
+            var expectedBuffer = channel.allocator.buffer(capacity: 7)
+            expectedBuffer.write(string: "bar")
+            XCTAssertEqual(error.leftOverBytes, expectedBuffer)
+        }
+        XCTAssertFalse(try channel.finish())
+    }
+    
+    func testRemoveHandlerWhenBufferIsEmpty() throws {
+        let channel = EmbeddedChannel()
+        
+        let handler = LineBasedFrameDecoder()
+        try channel.pipeline.add(handler: handler).wait()
+        
+        var buffer = channel.allocator.buffer(capacity: 4)
+        buffer.write(string: "foo\n")
+        XCTAssertTrue(try channel.writeInbound(buffer))
+        
+        var outputBuffer: ByteBuffer? = channel.readInbound()
+        XCTAssertEqual("foo", outputBuffer?.readString(length: 3))
+        
+        _ = try channel.pipeline.remove(handler: handler).wait()
+        XCTAssertNoThrow(try channel.throwIfErrorCaught())
+        XCTAssertFalse(try channel.finish())
+    }
+    
+    func testEmptyLine() throws {
+        let channel = EmbeddedChannel()
+        try channel.pipeline.add(handler: LineBasedFrameDecoder()).wait()
+        var buffer = channel.allocator.buffer(capacity: 1)
+        buffer.write(string: "\n")
+        XCTAssertTrue(try channel.writeInbound(buffer))
+        
+        var outputBuffer: ByteBuffer? = channel.readInbound()
+        XCTAssertEqual("", outputBuffer?.readString(length: 0))
+        XCTAssertFalse(try channel.finish())
+    }
+    
+    func testEmptyBuffer() throws {
+        let channel = EmbeddedChannel()
+        try channel.pipeline.add(handler: LineBasedFrameDecoder()).wait()
+        var buffer = channel.allocator.buffer(capacity: 1)
+        buffer.write(string: "")
+        XCTAssertFalse(try channel.writeInbound(buffer))
+        XCTAssertFalse(try channel.finish())
+    }
+}