apple · Lukasa · Aug 6, 2018 · Aug 6, 2018 · Lukasa · Aug 6, 2018
diff --git a/Sources/NIO/Embedded.swift b/Sources/NIO/Embedded.swift
@@ -157,8 +157,14 @@ class EmbeddedChannelCore: ChannelCore {
         closePromise.succeed(result: ())
     }
 
+    /// Contains the flushed items that went into the `Channel` (and on a regular channel would have hit the network).
     var outboundBuffer: [IOData] = []
+
+    /// Contains the unflushed items that went into the `Channel`
     var pendingOutboundBuffer: [(IOData, EventLoopPromise<Void>?)] = []
+
+    /// Contains the items that travelled the `ChannelPipeline` all the way and hit the tail channel handler. On a
+    /// regular `Channel` these items would be lost.
     var inboundBuffer: [NIOAny] = []
 
     func localAddress0() throws -> SocketAddress {
@@ -326,6 +332,13 @@ public class EmbeddedChannel: Channel {
         return readFromBuffer(buffer: &channelcore.inboundBuffer)
     }
 
+    /// Writes `data` into the `EmbeddedChannel`'s pipeline. This will result in a `channelRead` and a
+    /// `channelReadComplete` event for the first `ChannelHandler`.
+    ///
+    /// - parameters:
+    ///    - data: The data to fire through the pipeline.
+    /// - returns: If the `inboundBuffer` now contains items. The `inboundBuffer` will be empty until some item
+    ///            travels the `ChannelPipeline` all the way and hits the tail channel handler.
     @discardableResult public func writeInbound<T>(_ data: T) throws -> Bool {
         pipeline.fireChannelRead(NIOAny(data))
         pipeline.fireChannelReadComplete()

diff --git a/Tests/NIOHTTP1Tests/HTTPServerClientTest.swift b/Tests/NIOHTTP1Tests/HTTPServerClientTest.swift
@@ -19,18 +19,6 @@ import NIOFoundationCompat
 import Dispatch
 @testable import NIOHTTP1
 
-internal extension Channel {
-    func syncCloseAcceptingAlreadyClosed() throws {
-        do {
-            try self.close().wait()
-        } catch ChannelError.alreadyClosed {
-            /* we're happy with this one */
-        } catch let e {
-            throw e
-        }
-    }
-}
-
 extension Array where Array.Element == ByteBuffer {
     public func allAsBytes() -> [UInt8] {
         var out: [UInt8] = []
@@ -359,7 +347,7 @@ class HTTPServerClientTest : XCTestCase {
 
         let numBytes = 16 * 1024
         let httpHandler = SimpleHTTPServer(mode)
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
 
             // Set the handlers that are appled to the accepted Channels
@@ -368,20 +356,20 @@ class HTTPServerClientTest : XCTestCase {
                 channel.pipeline.configureHTTPServerPipeline(withPipeliningAssistance: false).then {
                     channel.pipeline.add(handler: httpHandler)
                 }
-            }.bind(host: "127.0.0.1", port: 0).wait()
+            }.bind(host: "127.0.0.1", port: 0).wait())
 
         defer {
             XCTAssertNoThrow(try serverChannel.syncCloseAcceptingAlreadyClosed())
         }
 
-        let clientChannel = try ClientBootstrap(group: group)
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
             .channelInitializer { channel in
                 channel.pipeline.addHTTPClientHandlers().then {
                     channel.pipeline.add(handler: accumulation)
                 }
             }
             .connect(to: serverChannel.localAddress!)
-            .wait()
+            .wait())
 
         defer {
             XCTAssertNoThrow(try clientChannel.syncCloseAcceptingAlreadyClosed())
@@ -417,7 +405,7 @@ class HTTPServerClientTest : XCTestCase {
 
         let numBytes = 16 * 1024
         let httpHandler = SimpleHTTPServer(mode)
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
 
             // Set the handlers that are appled to the accepted Channels
@@ -426,20 +414,20 @@ class HTTPServerClientTest : XCTestCase {
                 channel.pipeline.configureHTTPServerPipeline(withPipeliningAssistance: false).then {
                     channel.pipeline.add(handler: httpHandler)
                 }
-            }.bind(host: "127.0.0.1", port: 0).wait()
+            }.bind(host: "127.0.0.1", port: 0).wait())
 
         defer {
             XCTAssertNoThrow(try serverChannel.syncCloseAcceptingAlreadyClosed())
         }
 
-        let clientChannel = try ClientBootstrap(group: group)
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
             .channelInitializer { channel in
                 channel.pipeline.addHTTPClientHandlers().then {
                     channel.pipeline.add(handler: accumulation)
                 }
             }
             .connect(to: serverChannel.localAddress!)
-            .wait()
+            .wait())
 
         defer {
             XCTAssertNoThrow(try clientChannel.syncCloseAcceptingAlreadyClosed())
@@ -478,27 +466,26 @@ class HTTPServerClientTest : XCTestCase {
 
         let numBytes = 16 * 1024
         let httpHandler = SimpleHTTPServer(mode)
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
             .childChannelInitializer { channel in
                 channel.pipeline.configureHTTPServerPipeline(withPipeliningAssistance: false).then {
                     channel.pipeline.add(handler: httpHandler)
                 }
-            }.bind(host: "127.0.0.1", port: 0).wait()
+            }.bind(host: "127.0.0.1", port: 0).wait())
 
         defer {
             XCTAssertNoThrow(try serverChannel.syncCloseAcceptingAlreadyClosed())
         }
 
-        let clientChannel = try ClientBootstrap(group: group)
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
             .channelInitializer { channel in
                 channel.pipeline.addHTTPClientHandlers().then {
                     channel.pipeline.add(handler: accumulation)
                 }
             }
             .connect(to: serverChannel.localAddress!)
-            .wait()
-
+            .wait())
         defer {
             XCTAssertNoThrow(try clientChannel.syncCloseAcceptingAlreadyClosed())
         }
@@ -535,7 +522,7 @@ class HTTPServerClientTest : XCTestCase {
         }
         let numBytes = 16 * 1024
         let httpHandler = SimpleHTTPServer(mode)
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
 
             // Set the handlers that are appled to the accepted Channels
@@ -544,17 +531,15 @@ class HTTPServerClientTest : XCTestCase {
                 channel.pipeline.configureHTTPServerPipeline(withPipeliningAssistance: false).then {
                     channel.pipeline.add(handler: httpHandler)
                 }
-            }.bind(host: "127.0.0.1", port: 0).wait()
-
+            }.bind(host: "127.0.0.1", port: 0).wait())
         defer {
             XCTAssertNoThrow(try serverChannel.syncCloseAcceptingAlreadyClosed())
         }
 
-        let clientChannel = try ClientBootstrap(group: group)
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
             .channelInitializer({ $0.pipeline.add(handler: accumulation) })
             .connect(to: serverChannel.localAddress!)
-            .wait()
-
+            .wait())
         defer {
             XCTAssertNoThrow(try clientChannel.syncCloseAcceptingAlreadyClosed())
         }
@@ -580,25 +565,25 @@ class HTTPServerClientTest : XCTestCase {
 
         let numBytes = 16 * 1024
         let httpHandler = SimpleHTTPServer(.byteBuffer)
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
             .childChannelInitializer { channel in
                 channel.pipeline.configureHTTPServerPipeline(withPipeliningAssistance: false).then {
                     channel.pipeline.add(handler: httpHandler)
                 }
-            }.bind(host: "127.0.0.1", port: 0).wait()
+            }.bind(host: "127.0.0.1", port: 0).wait())
         defer {
             XCTAssertNoThrow(try serverChannel.syncCloseAcceptingAlreadyClosed())
         }
 
-        let clientChannel = try ClientBootstrap(group: group)
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
             .channelInitializer { channel in
                 channel.pipeline.addHTTPClientHandlers().then {
                     channel.pipeline.add(handler: accumulation)
                 }
             }
             .connect(to: serverChannel.localAddress!)
-            .wait()
+            .wait())
 
         defer {
             XCTAssertNoThrow(try clientChannel.syncCloseAcceptingAlreadyClosed())
@@ -625,26 +610,25 @@ class HTTPServerClientTest : XCTestCase {
 
         let numBytes = 16 * 1024
         let httpHandler = SimpleHTTPServer(.byteBuffer)
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
             .childChannelInitializer { channel in
                 channel.pipeline.configureHTTPServerPipeline(withPipeliningAssistance: false).then {
                     channel.pipeline.add(handler: httpHandler)
                 }
-            }.bind(host: "127.0.0.1", port: 0).wait()
+            }.bind(host: "127.0.0.1", port: 0).wait())
         defer {
             XCTAssertNoThrow(try serverChannel.syncCloseAcceptingAlreadyClosed())
         }
 
-        let clientChannel = try ClientBootstrap(group: group)
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
             .channelInitializer { channel in
                 channel.pipeline.addHTTPClientHandlers().then {
                     channel.pipeline.add(handler: accumulation)
                 }
             }
             .connect(to: serverChannel.localAddress!)
-            .wait()
-
+            .wait())
         defer {
             XCTAssertNoThrow(try clientChannel.syncCloseAcceptingAlreadyClosed())
         }
@@ -670,27 +654,26 @@ class HTTPServerClientTest : XCTestCase {
 
         let accumulation = HTTPClientResponsePartAssertHandler(HTTPVersion(major: 1, minor: 1), .ok, expectedHeaders, "Hello World!\r\n")
 
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
             .childChannelInitializer { channel in
                 channel.pipeline.addHTTPServerHandlers().then {
                     channel.pipeline.add(handler: SimpleHTTPServer(.byteBuffer))
                 }
-            }.bind(host: "127.0.0.1", port: 0).wait()
+            }.bind(host: "127.0.0.1", port: 0).wait())
 
         defer {
             XCTAssertNoThrow(try serverChannel.syncCloseAcceptingAlreadyClosed())
         }
 
-        let clientChannel = try ClientBootstrap(group: group)
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
             .channelInitializer { channel in
                 channel.pipeline.addHTTPClientHandlers().then {
                     channel.pipeline.add(handler: accumulation)
                 }
             }
             .connect(to: serverChannel.localAddress!)
-            .wait()
-
+            .wait())
         defer {
             XCTAssertNoThrow(try clientChannel.syncCloseAcceptingAlreadyClosed())
         }

diff --git a/Tests/NIOHTTP1Tests/TestUtils.swift b/Tests/NIOHTTP1Tests/TestUtils.swift
@@ -0,0 +1 @@
+../NIOTests/TestUtils.swift
diff --git a/Tests/NIOTests/AcceptBackoffHandlerTest.swift b/Tests/NIOTests/AcceptBackoffHandlerTest.swift
@@ -249,7 +249,9 @@ public class AcceptBackoffHandlerTest: XCTestCase {
     private func setupChannel(group: EventLoopGroup, readCountHandler: ReadCountHandler, backoffProvider: @escaping (IOError) -> TimeAmount? = AcceptBackoffHandler.defaultBackoffProvider, errors: [Int32]) throws -> ServerSocketChannel {
         let eventLoop = group.next() as! SelectableEventLoop
         let socket = try NonAcceptingServerSocket(errors: errors)
-        let serverChannel = try ServerSocketChannel(serverSocket: socket, eventLoop: eventLoop, group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerSocketChannel(serverSocket: socket,
+                                                                           eventLoop: eventLoop,
+                                                                           group: group))
 
         XCTAssertNoThrow(try serverChannel.setOption(option: ChannelOptions.autoRead, value: false).wait())
         XCTAssertNoThrow(try serverChannel.pipeline.add(handler: readCountHandler).then { _ in

diff --git a/Tests/NIOTests/ChannelNotificationTest.swift b/Tests/NIOTests/ChannelNotificationTest.swift
@@ -293,7 +293,7 @@ class ChannelNotificationTest: XCTestCase {
 
         let acceptedChannelPromise: EventLoopPromise<Channel> = group.next().newPromise()
 
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
             .serverChannelInitializer { channel in
                 channel.pipeline.add(handler: ServerSocketChannelLifecycleVerificationHandler())
@@ -302,15 +302,15 @@ class ChannelNotificationTest: XCTestCase {
             .childChannelInitializer { channel in
                 channel.pipeline.add(handler: AcceptedSocketChannelLifecycleVerificationHandler(acceptedChannelPromise))
             }
-            .bind(host: "127.0.0.1", port: 0).wait()
+            .bind(host: "127.0.0.1", port: 0).wait())
 
-        let clientChannel = try ClientBootstrap(group: group)
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
             .channelInitializer { channel in
                 channel.pipeline.add(handler: SocketChannelLifecycleVerificationHandler())
             }
-            .connect(to: serverChannel.localAddress!).wait()
-        try clientChannel.close().wait()
-        try clientChannel.closeFuture.wait()
+            .connect(to: serverChannel.localAddress!).wait())
+        XCTAssertNoThrow(try clientChannel.close().wait())
+        XCTAssertNoThrow(try clientChannel.closeFuture.wait())
 
         let channel = try acceptedChannelPromise.futureResult.wait()
         var buffer = channel.allocator.buffer(capacity: 8)
@@ -320,10 +320,10 @@ class ChannelNotificationTest: XCTestCase {
         while (try? channel.writeAndFlush(buffer).wait()) != nil {
             // Just write in a loop until it fails to ensure we detect the closed connection in a timely manner.
         }
-        try channel.closeFuture.wait()
+        XCTAssertNoThrow(try channel.closeFuture.wait())
 
-        try serverChannel.close().wait()
-        try serverChannel.closeFuture.wait()
+        XCTAssertNoThrow(try serverChannel.close().wait())
+        XCTAssertNoThrow(try serverChannel.closeFuture.wait())
     }
 
     func testActiveBeforeChannelRead() throws {
@@ -375,26 +375,26 @@ class ChannelNotificationTest: XCTestCase {
         }
 
         let promise: EventLoopPromise<Void> = group.next().newPromise()
-        let serverChannel = try ServerBootstrap(group: group)
+        let serverChannel = try assertNoThrowWithValue(ServerBootstrap(group: group)
             .serverChannelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
             .childChannelOption(ChannelOptions.autoRead, value: true)
             .childChannelInitializer { channel in
                 channel.pipeline.add(handler: OrderVerificationHandler(promise))
             }
-            .bind(host: "127.0.0.1", port: 0).wait()
+            .bind(host: "127.0.0.1", port: 0).wait())
 
-        let clientChannel = try ClientBootstrap(group: group)
-            .connect(to: serverChannel.localAddress!).wait()
+        let clientChannel = try assertNoThrowWithValue(ClientBootstrap(group: group)
+            .connect(to: serverChannel.localAddress!).wait())
 
         var buffer = clientChannel.allocator.buffer(capacity: 2)
         buffer.write(string: "X")
-        try clientChannel.writeAndFlush(buffer).then {
+        XCTAssertNoThrow(try clientChannel.writeAndFlush(buffer).then {
             clientChannel.close()
-        }.wait()
-        try promise.futureResult.wait()
+        }.wait())
+        XCTAssertNoThrow(try promise.futureResult.wait())
 
-        try clientChannel.closeFuture.wait()
-        try serverChannel.close().wait()
-        try serverChannel.closeFuture.wait()
+        XCTAssertNoThrow(try clientChannel.closeFuture.wait())
+        XCTAssertNoThrow(try serverChannel.close().wait())
+        XCTAssertNoThrow(try serverChannel.closeFuture.wait())
     }
 }
diff --git a/Tests/NIOTests/ChannelPipelineTest.swift b/Tests/NIOTests/ChannelPipelineTest.swift
@@ -64,7 +64,7 @@ class ChannelPipelineTest: XCTestCase {
     func testAddAfterClose() throws {
 
         let channel = EmbeddedChannel()
-        _ = channel.close()
+        XCTAssertNoThrow(try channel.close().wait())
 
         channel.pipeline.removeHandlers()
 
@@ -111,7 +111,7 @@ class ChannelPipelineTest: XCTestCase {
             return 1
         }).wait()
 
-        try channel.writeAndFlush(NIOAny("msg")).wait()
+        XCTAssertNoThrow(try channel.writeAndFlush(NIOAny("msg")).wait() as Void)
         if let data = channel.readOutbound() {
             XCTAssertEqual(IOData.byteBuffer(buf), data)
         } else {
@@ -194,15 +194,15 @@ class ChannelPipelineTest: XCTestCase {
 
     func testEmptyPipelineWorks() throws {
         let channel = EmbeddedChannel()
-        try channel.writeInbound(2)
+        XCTAssertTrue(try assertNoThrowWithValue(channel.writeInbound(2)))
         XCTAssertEqual(Optional<Int>.some(2), channel.readInbound())
         XCTAssertFalse(try channel.finish())
     }
 
     func testWriteAfterClose() throws {
 
         let channel = EmbeddedChannel()
-        _ = try channel.close().wait()
+        XCTAssertNoThrow(try channel.close().wait())
         let loop = channel.eventLoop as! EmbeddedEventLoop
         loop.run()
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		../NIOTests/TestUtils.swift
Copy link Contributor Lukasa Aug 6, 2018 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. Rather than create this symlink, should we just move this into a target and depend on it from the two test targets? Copy link Member Author weissi Aug 6, 2018 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. @Lukasa SwiftPM doesn't unfortunately support that AFAIK. We can't make it a test target as we can't depend on that and we can't make it a normal target as it depends on `XCTest`. Therefore this is the only way I'm aware of... Copy link Contributor Lukasa Aug 6, 2018 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. Ugh, ok then.