test that we tolerate out-of-band data

Sources/NIO/Socket.swift

@@ -134,7 +134,7 @@ typealias IOVector = iovec
     func sendto(pointer: UnsafeRawBufferPointer, destinationPtr: UnsafePointer<sockaddr>, destinationSize: socklen_t) throws -> IOResult<Int> {
         return try withUnsafeFileDescriptor { fd in
             try Posix.sendto(descriptor: fd, pointer: UnsafeMutableRawPointer(mutating: pointer.baseAddress!),
-                             size: pointer.count, destinationPtr: destinationPtr,
+                             size: pointer.count, flags: 0, destinationPtr: destinationPtr,
                              destinationSize: destinationSize)
         }
     }

Sources/NIO/System.swift

@@ -194,6 +194,7 @@ internal enum Posix {
     static let SHUT_RD: CInt = CInt(Darwin.SHUT_RD)
     static let SHUT_WR: CInt = CInt(Darwin.SHUT_WR)
     static let SHUT_RDWR: CInt = CInt(Darwin.SHUT_RDWR)
+    static let MSG_OOB: CInt = Darwin.MSG_OOB
 #elseif os(Linux) || os(FreeBSD) || os(Android)
 
 #if os(Android)
@@ -208,6 +209,7 @@ internal enum Posix {
     static let SHUT_RD: CInt = CInt(Glibc.SHUT_RD)
     static let SHUT_WR: CInt = CInt(Glibc.SHUT_WR)
     static let SHUT_RDWR: CInt = CInt(Glibc.SHUT_RDWR)
+    static let MSG_OOB: CInt = CInt(Glibc.MSG_OOB)
 #else
     static var SOCK_STREAM: CInt {
         fatalError("unsupported OS")
@@ -365,10 +367,13 @@ internal enum Posix {
     }
 
     @inline(never)
-    public static func sendto(descriptor: CInt, pointer: UnsafeRawPointer, size: size_t,
+    public static func sendto(descriptor: CInt,
+                              pointer: UnsafeRawPointer,
+                              size: size_t,
+                              flags: CInt,
                               destinationPtr: UnsafePointer<sockaddr>, destinationSize: socklen_t) throws -> IOResult<Int> {
         return try wrapSyscallMayBlock {
-            sysSendTo(descriptor, pointer, size, 0, destinationPtr, destinationSize)
+            sysSendTo(descriptor, pointer, size, flags,  destinationPtr, destinationSize)
         }
     }
 

Tests/NIOTests/ChannelTests+XCTest.swift

@@ -82,6 +82,7 @@ extension ChannelTests {
                 ("testTCP_NODELAYisOnByDefault", testTCP_NODELAYisOnByDefault),
                 ("testDescriptionCanBeCalledFromNonEventLoopThreads", testDescriptionCanBeCalledFromNonEventLoopThreads),
                 ("testFixedSizeRecvByteBufferAllocatorSizeIsConstant", testFixedSizeRecvByteBufferAllocatorSizeIsConstant),
+                ("testWeTolerateOutOfBandData", testWeTolerateOutOfBandData),
            ]
    }
 }

Tests/NIOTests/ChannelTests.swift

@@ -2776,7 +2776,29 @@ public final class ChannelTests: XCTestCase {
         XCTAssertEqual(1, b1.capacity)
         XCTAssertEqual(1, b2.capacity)
         XCTAssertEqual(1, allocator.capacity)
+    }
 
+    func testWeTolerateOutOfBandData() {
+        func runTest() throws {
+            try withCrossConnectedRawSocketToChannel { socketFD, channel in
+                let channelRead = channel.eventLoop.makePromise(of: Void.self)
+                XCTAssertNoThrow(try channel.pipeline.addHandler(FulfillOnFirstEventHandler(channelReadPromise: channelRead)).wait())
+                var data = [UInt8(ascii: "x")]
+                try channel.remoteAddress!.withSockAddr { sockAddrPtr, sockAddrSize in
+                    for _ in 0..<100 {
+                    XCTAssertNoThrow(try Posix.sendto(descriptor: socketFD,
+                                                      pointer: &data,
+                                                      size: 1,
+                                                      flags: Posix.MSG_OOB,
+                                                      destinationPtr: sockAddrPtr,
+                                                      destinationSize: .init(sockAddrSize)))
+                    }
+                    Thread.sleep(until: .init(timeIntervalSinceNow: 0.01))
+                    XCTAssertFalse(channelRead.futureResult.isFulfilled)
+                }
+            }
+        }
+        XCTAssertNoThrow(try runTest())
     }
 }
 

Tests/NIOTests/TestUtils.swift

@@ -653,3 +653,34 @@ extension EventLoopFuture {
         }
     }
 }
+
+func withCrossConnectedRawSocketToChannel<R>(file: StaticString = #file,
+                                             line: UInt = #line,
+                                             _ body: (CInt, Channel) throws -> R) throws -> R? {
+    let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    defer {
+        XCTAssertNoThrow(try group.syncShutdownGracefully(), file: file, line: line)
+    }
+
+    // socketpair doesn't work for AF_INET (at least on Darwin) so let's build it ourselves.
+    let serverSocket = try ServerSocket(protocolFamily: AF_INET, setNonBlocking: false)
+    defer {
+        XCTAssertNoThrow(try serverSocket.close(), file: file, line: line)
+    }
+    try serverSocket.bind(to: .init(ipAddress: "127.0.0.1", port: 0))
+    try serverSocket.listen()
+
+    let clientChannel = ClientBootstrap(group: group).connect(to: try serverSocket.localAddress())
+    let acceptedSocket = try serverSocket.accept()! /* only returns nil if in non-blocking mode */
+    defer {
+        try? acceptedSocket.close()
+    }
+    let channel = try clientChannel.wait()
+    defer {
+        XCTAssertNoThrow(try channel.syncCloseAcceptingAlreadyClosed(), file: file, line: line)
+    }
+
+    return try acceptedSocket.withUnsafeFileDescriptor { acceptedFD in
+        try body(acceptedFD, channel)
+    }
+}