fail unhandled outbound user events (#867)

Motivation:

Unhandled outbound user events should be failed but previously they were
succeeded.

Modifications:

Succeeding them is odd because nothing happened to them, so they should
be failed.

Result:

fixes #487

Sources/NIO/BaseSocketChannel.swift

@@ -814,7 +814,7 @@ class BaseSocketChannel<T: BaseSocket>: SelectableChannel, ChannelCore {
     }
 
     public final func triggerUserOutboundEvent0(_ event: Any, promise: EventLoopPromise<Void>?) {
-        promise?.succeed(())
+        promise?.fail(ChannelError.operationUnsupported)
     }
 
     // Methods invoked from the EventLoop itself

Sources/NIO/Embedded.swift

@@ -239,7 +239,7 @@ class EmbeddedChannelCore: ChannelCore {
     }
 
     public final func triggerUserOutboundEvent0(_ event: Any, promise: EventLoopPromise<Void>?) {
-        promise?.succeed(())
+        promise?.fail(ChannelError.operationUnsupported)
     }
 
     func channelRead0(_ data: NIOAny) {

Tests/NIOTests/ChannelTests+XCTest.swift

@@ -77,6 +77,7 @@ extension ChannelTests {
                 ("testTryingToBindOnPortThatIsAlreadyBoundFailsButDoesNotCrash", testTryingToBindOnPortThatIsAlreadyBoundFailsButDoesNotCrash),
                 ("testCloseInReadTriggeredByDrainingTheReceiveBufferBecauseOfWriteError", testCloseInReadTriggeredByDrainingTheReceiveBufferBecauseOfWriteError),
                 ("testApplyingTwoDistinctSocketOptionsOfSameTypeWorks", testApplyingTwoDistinctSocketOptionsOfSameTypeWorks),
+                ("testUnprocessedOutboundUserEventFailsOnServerSocketChannel", testUnprocessedOutboundUserEventFailsOnServerSocketChannel),
            ]
    }
 }

Tests/NIOTests/ChannelTests.swift

@@ -2014,7 +2014,13 @@ public class ChannelTests: XCTestCase {
             }
         }
         withChannel { channel in
-            XCTAssertNoThrow(try channel.triggerUserOutboundEvent("foo").wait())
+            XCTAssertThrowsError(try channel.triggerUserOutboundEvent("foo").wait()) { error in
+                if let error = error as? ChannelError {
+                    XCTAssertEqual(ChannelError.operationUnsupported, error)
+                } else {
+                    XCTFail("wrong error: \(error)")
+                }
+            }
         }
         withChannel { channel in
             XCTAssertFalse(channel.isActive)
@@ -2671,6 +2677,23 @@ public class ChannelTests: XCTestCase {
 
         XCTAssertTrue(try getBoolSocketOption(channel: accepted3, level: IPPROTO_TCP, name: TCP_NODELAY))
     }
+
+    func testUnprocessedOutboundUserEventFailsOnServerSocketChannel() throws {
+        let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+        defer {
+            XCTAssertNoThrow(try group.syncShutdownGracefully())
+        }
+        let channel = try ServerSocketChannel(eventLoop: group.next() as! SelectableEventLoop,
+                                              group: group,
+                                              protocolFamily: AF_INET)
+        XCTAssertThrowsError(try channel.triggerUserOutboundEvent("event").wait()) { (error: Error) in
+            if let error = error as? ChannelError {
+                XCTAssertEqual(ChannelError.operationUnsupported, error)
+            } else {
+                XCTFail("unexpected error: \(error)")
+            }
+        }
+    }
 }
 
 fileprivate final class FailRegistrationAndDelayCloseHandler: ChannelOutboundHandler {

Tests/NIOTests/DatagramChannelTests+XCTest.swift

@@ -42,6 +42,7 @@ extension DatagramChannelTests {
                 ("testSetGetOptionClosedDatagramChannel", testSetGetOptionClosedDatagramChannel),
                 ("testWritesAreAccountedCorrectly", testWritesAreAccountedCorrectly),
                 ("testSettingTwoDistinctChannelOptionsWorksForDatagramChannel", testSettingTwoDistinctChannelOptionsWorksForDatagramChannel),
+                ("testUnprocessedOutboundUserEventFailsOnDatagramChannel", testUnprocessedOutboundUserEventFailsOnDatagramChannel),
            ]
    }
 }

Tests/NIOTests/DatagramChannelTests.swift

@@ -457,4 +457,20 @@ final class DatagramChannelTests: XCTestCase {
         XCTAssertTrue(try getBoolSocketOption(channel: channel, level: SOL_SOCKET, name: SO_TIMESTAMP))
         XCTAssertFalse(try getBoolSocketOption(channel: channel, level: SOL_SOCKET, name: SO_KEEPALIVE))
     }
+
+    func testUnprocessedOutboundUserEventFailsOnDatagramChannel() throws {
+        let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+        defer {
+            XCTAssertNoThrow(try group.syncShutdownGracefully())
+        }
+        let channel = try DatagramChannel(eventLoop: group.next() as! SelectableEventLoop,
+                                          protocolFamily: AF_INET)
+        XCTAssertThrowsError(try channel.triggerUserOutboundEvent("event").wait()) { (error: Error) in
+            if let error = error as? ChannelError {
+                XCTAssertEqual(ChannelError.operationUnsupported, error)
+            } else {
+                XCTFail("unexpected error: \(error)")
+            }
+        }
+    }
 }

Tests/NIOTests/EmbeddedChannelTest+XCTest.swift

@@ -39,6 +39,7 @@ extension EmbeddedChannelTest {
                 ("testWriteWithoutFlushDoesNotWrite", testWriteWithoutFlushDoesNotWrite),
                 ("testSetLocalAddressAfterSuccessfulBind", testSetLocalAddressAfterSuccessfulBind),
                 ("testSetRemoteAddressAfterSuccessfulConnect", testSetRemoteAddressAfterSuccessfulConnect),
+                ("testUnprocessedOutboundUserEventFailsOnEmbeddedChannel", testUnprocessedOutboundUserEventFailsOnEmbeddedChannel),
            ]
    }
 }

Tests/NIOTests/EmbeddedChannelTest.swift

@@ -213,4 +213,15 @@ class EmbeddedChannelTest: XCTestCase {
         }
         try connectPromise.futureResult.wait()
     }
+
+    func testUnprocessedOutboundUserEventFailsOnEmbeddedChannel() {
+        let channel = EmbeddedChannel()
+        XCTAssertThrowsError(try channel.triggerUserOutboundEvent("event").wait()) { (error: Error) in
+            if let error = error as? ChannelError {
+                XCTAssertEqual(ChannelError.operationUnsupported, error)
+            } else {
+                XCTFail("unexpected error: \(error)")
+            }
+        }
+    }
 }

Tests/NIOTests/SocketChannelTest+XCTest.swift

@@ -46,6 +46,8 @@ extension SocketChannelTest {
                 ("testLocalAndRemoteAddressNotNilInChannelInactiveAndHandlerRemoved", testLocalAndRemoteAddressNotNilInChannelInactiveAndHandlerRemoved),
                 ("testSocketFlagNONBLOCKWorks", testSocketFlagNONBLOCKWorks),
                 ("testInstantTCPConnectionResetThrowsError", testInstantTCPConnectionResetThrowsError),
+                ("testUnprocessedOutboundUserEventFailsOnServerSocketChannel", testUnprocessedOutboundUserEventFailsOnServerSocketChannel),
+                ("testUnprocessedOutboundUserEventFailsOnSocketChannel", testUnprocessedOutboundUserEventFailsOnSocketChannel),
            ]
    }
 }

Tests/NIOTests/SocketChannelTest.swift

@@ -614,4 +614,37 @@ public class SocketChannelTest : XCTestCase {
             XCTAssertEqual(result.errnoCode, EINVAL)
         #endif
     }
+
+    func testUnprocessedOutboundUserEventFailsOnServerSocketChannel() throws {
+        let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+        defer {
+            XCTAssertNoThrow(try group.syncShutdownGracefully())
+        }
+        let channel = try ServerSocketChannel(eventLoop: group.next() as! SelectableEventLoop,
+                                              group: group,
+                                              protocolFamily: AF_INET)
+        XCTAssertThrowsError(try channel.triggerUserOutboundEvent("event").wait()) { (error: Error) in
+            if let error = error as? ChannelError {
+                XCTAssertEqual(ChannelError.operationUnsupported, error)
+            } else {
+                XCTFail("unexpected error: \(error)")
+            }
+        }
+    }
+
+    func testUnprocessedOutboundUserEventFailsOnSocketChannel() throws {
+        let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+        defer {
+            XCTAssertNoThrow(try group.syncShutdownGracefully())
+        }
+        let channel = try SocketChannel(eventLoop: group.next() as! SelectableEventLoop,
+                                        protocolFamily: AF_INET)
+        XCTAssertThrowsError(try channel.triggerUserOutboundEvent("event").wait()) { (error: Error) in
+            if let error = error as? ChannelError {
+                XCTAssertEqual(ChannelError.operationUnsupported, error)
+            } else {
+                XCTFail("unexpected error: \(error)")
+            }
+        }
+    }
 }