Adds a line-based frame decoder that can split received buffers on line endings. (#11)

* 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,92 @@
+//===----------------------------------------------------------------------===//
+//
+// 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?
+    // keep track of the last scan offset from the buffer's reader index (if we didn't find the delimiter)
+    private var lastScanOffset = 0
+    
+    public init() { }
+    
+    public func decode(ctx: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
+        if let frame = try self.findNextFrame(buffer: &buffer) {
+            ctx.fireChannelRead(wrapInboundOut(frame))
+            return .continue
+        } else {
+            return .needMoreData
+        }
+    }
+    
+    private func findNextFrame(buffer: inout ByteBuffer) throws -> ByteBuffer? {
+        let view = buffer.readableBytesView.dropFirst(self.lastScanOffset)
+        // look for the delimiter
+        if let delimiterIndex = view.firstIndex(of: 0x0A) { // '\n'
+            let length = delimiterIndex - buffer.readerIndex
+            let dropCarriageReturn = delimiterIndex > view.startIndex && view[delimiterIndex - 1] == 0x0D // '\r'
+            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.lastScanOffset = 0
+            return buff
+        }
+        // next scan we start where we stopped
+        self.lastScanOffset = buffer.readableBytes
+        return nil
+    }
+    
+    public func handlerRemoved(ctx: ChannelHandlerContext) {
+        self.handleLeftOverBytes(ctx: ctx)
+    }
+    
+    public func channelInactive(ctx: ChannelHandlerContext) {
+        self.handleLeftOverBytes(ctx: ctx)
+    }
+    
+    private func handleLeftOverBytes(ctx: ChannelHandlerContext) {
+        if let buffer = self.cumulationBuffer, buffer.readableBytes > 0 {
+            self.cumulationBuffer?.clear()
+            ctx.fireErrorCaught(NIOExtrasErrors.LeftOverBytesError(leftOverBytes: buffer))
+        }
+    }
+}
+
+#if !swift(>=4.2)
+private extension ByteBufferView {
+    func firstIndex(of element: UInt8) -> Int? {
+        return self.index(of: element)
+    }
+}
+#endif

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,39 @@
+//===----------------------------------------------------------------------===//
+//
+// 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),
+                ("testReaderIndexNotZero", testReaderIndexNotZero),
+                ("testChannelInactiveWithLeftOverBytes", testChannelInactiveWithLeftOverBytes),
+           ]
+   }
+}
+

Tests/NIOExtrasTests/LineBasedFramceDecoderTest.swift

@@ -0,0 +1,143 @@
+//===----------------------------------------------------------------------===//
+//
+// 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 {
+    
+    private var channel: EmbeddedChannel!
+    private var handler: LineBasedFrameDecoder!
+    
+    override func setUp() {
+        self.channel = EmbeddedChannel()
+        self.handler = LineBasedFrameDecoder()
+        try? self.channel.pipeline.add(handler: self.handler).wait()
+    }
+    
+    func testDecodeOneCharacterAtATime() throws {
+        let message = "abcdefghij\r"
+        // we write one character at a time
+        try message.forEach {
+            var buffer = self.channel.allocator.buffer(capacity: 1)
+            buffer.write(string: "\($0)")
+            XCTAssertFalse(try self.channel.writeInbound(buffer))
+        }
+        // let's add `\n`
+        var buffer = self.channel.allocator.buffer(capacity: 1)
+        buffer.write(string: "\n")
+        XCTAssertTrue(try self.channel.writeInbound(buffer))
+        
+        var outputBuffer: ByteBuffer? = self.channel.readInbound()
+        XCTAssertEqual("abcdefghij", outputBuffer?.readString(length: 10))
+        XCTAssertFalse(try self.channel.finish())
+    }
+    
+    func testRemoveHandlerWhenBufferIsNotEmpty() throws {
+        var buffer = self.channel.allocator.buffer(capacity: 8)
+        buffer.write(string: "foo\r\nbar")
+        XCTAssertTrue(try self.channel.writeInbound(buffer))
+        var outputBuffer: ByteBuffer? = self.channel.readInbound()
+        XCTAssertEqual(3, outputBuffer?.readableBytes)
+        XCTAssertEqual("foo", outputBuffer?.readString(length: 3))
+        
+        _ = try self.channel.pipeline.remove(handler: handler).wait()
+        XCTAssertThrowsError(try self.channel.throwIfErrorCaught()) { error in
+            guard let error = error as? NIOExtrasErrors.LeftOverBytesError else {
+                XCTFail()
+                return
+            }
+            
+            var expectedBuffer = self.channel.allocator.buffer(capacity: 7)
+            expectedBuffer.write(string: "bar")
+            XCTAssertEqual(error.leftOverBytes, expectedBuffer)
+        }
+        XCTAssertFalse(try self.channel.finish())
+    }
+    
+    func testRemoveHandlerWhenBufferIsEmpty() throws {
+        var buffer = self.channel.allocator.buffer(capacity: 4)
+        buffer.write(string: "foo\n")
+        XCTAssertTrue(try self.channel.writeInbound(buffer))
+        
+        var outputBuffer: ByteBuffer? = self.channel.readInbound()
+        XCTAssertEqual("foo", outputBuffer?.readString(length: 3))
+        
+        _ = try self.channel.pipeline.remove(handler: handler).wait()
+        XCTAssertNoThrow(try self.channel.throwIfErrorCaught())
+        XCTAssertFalse(try self.channel.finish())
+    }
+    
+    func testEmptyLine() throws {
+        var buffer = self.channel.allocator.buffer(capacity: 1)
+        buffer.write(string: "\n")
+        XCTAssertTrue(try self.channel.writeInbound(buffer))
+        
+        var outputBuffer: ByteBuffer? = self.channel.readInbound()
+        XCTAssertEqual("", outputBuffer?.readString(length: 0))
+        XCTAssertFalse(try self.channel.finish())
+    }
+    
+    func testEmptyBuffer() throws {
+        var buffer = self.channel.allocator.buffer(capacity: 1)
+        buffer.write(string: "")
+        XCTAssertFalse(try self.channel.writeInbound(buffer))
+        XCTAssertFalse(try self.channel.finish())
+    }
+    
+    func testReaderIndexNotZero() throws {
+        var buffer = self.channel.allocator.buffer(capacity: 8)
+        // read "abc" so the reader index is not 0
+        buffer.write(string: "abcfoo\r\nbar")
+        XCTAssertEqual("abc", buffer.readString(length: 3))
+        XCTAssertEqual(3, buffer.readerIndex)
+        
+        XCTAssertTrue(try self.channel.writeInbound(buffer))
+        var outputBuffer: ByteBuffer? = self.channel.readInbound()
+        XCTAssertEqual(3, outputBuffer?.readableBytes)
+        XCTAssertEqual("foo", outputBuffer?.readString(length: 3))
+        // discard the read bytes - this will reset the reader index to 0
+        var buf = self.handler.cumulationBuffer
+        buf?.discardReadBytes()
+        self.handler.cumulationBuffer = buf
+        XCTAssertEqual(0, self.handler.cumulationBuffer?.readerIndex ?? -1)
+        
+        buffer.write(string: "\r\n")
+        XCTAssertTrue(try self.channel.writeInbound(buffer))
+        outputBuffer = self.channel.readInbound()
+        XCTAssertEqual("bar", outputBuffer?.readString(length: 3))
+    }
+    
+    func testChannelInactiveWithLeftOverBytes() throws {
+        // add some data to the buffer
+        var buffer = self.channel.allocator.buffer(capacity: 2)
+        // read "abc" so the reader index is not 0
+        buffer.write(string: "hi")
+        XCTAssertFalse(try self.channel.writeInbound(buffer))
+        
+        try self.channel.close().wait()
+        XCTAssertThrowsError(try self.channel.throwIfErrorCaught()) { error in
+            guard let error = error as? NIOExtrasErrors.LeftOverBytesError else {
+                XCTFail()
+                return
+            }
+            var expectedBuffer = self.channel.allocator.buffer(capacity: 7)
+            expectedBuffer.write(string: "hi")
+            XCTAssertEqual(error.leftOverBytes, expectedBuffer)
+            // make sure we have cleared the buffer
+            XCTAssertEqual(handler.cumulationBuffer?.readableBytes, 0)
+        }
+    }
+}