Embedded: getOption(ChannelOptions.allowRemoteHalfClosure) should not fatalError

Sources/NIOEmbedded/AsyncTestingChannel.swift

@@ -170,6 +170,16 @@ public final class NIOAsyncTestingChannel: Channel {
     /// - note: An ``NIOAsyncTestingChannel`` starts _inactive_ and can be activated, for example by calling `connect`.
     public var isActive: Bool { return channelcore.isActive }
 
+    /// - see: `ChannelOptions.Types.AllowRemoteHalfClosureOption`
+    public var allowRemoteHalfClosure: Bool {
+        get {
+            return channelcore.allowRemoteHalfClosure
+        }
+        set {
+            channelcore.allowRemoteHalfClosure = newValue
+        }
+    }
+
     /// - see: `Channel.closeFuture`
     public var closeFuture: EventLoopFuture<Void> { return channelcore.closePromise.futureResult }
 
@@ -362,7 +372,7 @@ public final class NIOAsyncTestingChannel: Channel {
             try self._readFromBuffer(buffer: &self.channelcore.outboundBuffer)
         }
     }
-    
+
     /// This method is similar to ``NIOAsyncTestingChannel/readOutbound(as:)`` but will wait if the outbound buffer is empty.
     /// If available, this method reads one element of type `T` out of the ``NIOAsyncTestingChannel``'s outbound buffer. If the
     /// first element was of a different type than requested, ``WrongTypeError`` will be thrown, if there
@@ -412,7 +422,7 @@ public final class NIOAsyncTestingChannel: Channel {
             try self._readFromBuffer(buffer: &self.channelcore.inboundBuffer)
         }
     }
-    
+
     /// This method is similar to ``NIOAsyncTestingChannel/readInbound(as:)`` but will wait if the inbound buffer is empty.
     /// If available, this method reads one element of type `T` out of the ``NIOAsyncTestingChannel``'s inbound buffer. If the
     /// first element was of a different type than requested, ``WrongTypeError`` will be thrown, if there
@@ -499,7 +509,7 @@ public final class NIOAsyncTestingChannel: Channel {
             throw error
         }
     }
-    
+
 
     @inlinable
     func _readFromBuffer<T>(buffer: inout CircularBuffer<NIOAny>) throws -> T? {
@@ -510,7 +520,7 @@ public final class NIOAsyncTestingChannel: Channel {
         }
         return try self._cast(buffer.removeFirst(), to: T.self)
     }
-    
+
     @inlinable
     func _cast<T>(_ element: NIOAny, to: T.Type = T.self) throws -> T {
         guard let t = self._channelCore.tryUnwrapData(element, as: T.self) else {
@@ -532,8 +542,12 @@ public final class NIOAsyncTestingChannel: Channel {
 
     @inlinable
     internal func setOptionSync<Option: ChannelOption>(_ option: Option, value: Option.Value) {
-        // No options supported
-        fatalError("no options supported")
+        if option is ChannelOptions.Types.AllowRemoteHalfClosureOption {
+            self.allowRemoteHalfClosure = value as! Bool
+            return
+        }
+        // No other options supported
+        fatalError("option not supported")
     }
 
     /// - see: `Channel.getOption`
@@ -551,6 +565,9 @@ public final class NIOAsyncTestingChannel: Channel {
         if option is ChannelOptions.Types.AutoReadOption {
             return true as! Option.Value
         }
+        if option is ChannelOptions.Types.AllowRemoteHalfClosureOption {
+            return self.allowRemoteHalfClosure as! Option.Value
+        }
         fatalError("option \(option) not supported")
     }
 

Sources/NIOEmbedded/Embedded.swift

@@ -255,8 +255,18 @@ class EmbeddedChannelCore: ChannelCore {
         }
     }
 
+    var allowRemoteHalfClosure: Bool {
+        get {
+            return self._allowRemoteHalfClosure.load(ordering: .sequentiallyConsistent)
+        }
+        set {
+            self._allowRemoteHalfClosure.store(newValue, ordering: .sequentiallyConsistent)
+        }
+    }
+
     private let _isOpen = ManagedAtomic(true)
     private let _isActive = ManagedAtomic(false)
+    private let _allowRemoteHalfClosure = ManagedAtomic(false)
 
     let eventLoop: EventLoop
     let closePromise: EventLoopPromise<Void>
@@ -281,7 +291,7 @@ class EmbeddedChannelCore: ChannelCore {
     /// Contains the flushed items that went into the `Channel` (and on a regular channel would have hit the network).
     @usableFromInline
     var outboundBuffer: CircularBuffer<NIOAny> = CircularBuffer()
-    
+
     /// Contains observers that want to consume the first element that would be appended to the `outboundBuffer`
     @usableFromInline
     var outboundBufferConsumer: Deque<(NIOAny) -> Void> = []
@@ -294,7 +304,7 @@ class EmbeddedChannelCore: ChannelCore {
     /// regular `Channel` these items would be lost.
     @usableFromInline
     var inboundBuffer: CircularBuffer<NIOAny> = CircularBuffer()
-    
+
     /// Contains observers that want to consume the first element that would be appended to the `inboundBuffer`
     @usableFromInline
     var inboundBufferConsumer: Deque<(NIOAny) -> Void> = []
@@ -551,6 +561,16 @@ public final class EmbeddedChannel: Channel {
     /// - note: An `EmbeddedChannel` starts _inactive_ and can be activated, for example by calling `connect`.
     public var isActive: Bool { return channelcore.isActive }
 
+    /// - see: `ChannelOptions.Types.AllowRemoteHalfClosureOption`
+    public var allowRemoteHalfClosure: Bool {
+        get {
+            return channelcore.allowRemoteHalfClosure
+        }
+        set {
+            channelcore.allowRemoteHalfClosure = newValue
+        }
+    }
+
     /// - see: `Channel.closeFuture`
     public var closeFuture: EventLoopFuture<Void> { return channelcore.closePromise.futureResult }
 
@@ -749,7 +769,7 @@ public final class EmbeddedChannel: Channel {
         let handlers = handler.map { [$0] } ?? []
         self.init(handlers: handlers, loop: loop)
     }
-    
+
     /// Create a new instance.
     ///
     /// During creation it will automatically also register itself on the `EmbeddedEventLoop`.
@@ -776,8 +796,12 @@ public final class EmbeddedChannel: Channel {
 
     @inlinable
     internal func setOptionSync<Option: ChannelOption>(_ option: Option, value: Option.Value) {
-        // No options supported
-        fatalError("no options supported")
+        if option is ChannelOptions.Types.AllowRemoteHalfClosureOption {
+            self.allowRemoteHalfClosure = value as! Bool
+            return
+        }
+        // No other options supported
+        fatalError("option not supported")
     }
 
     /// - see: `Channel.getOption`
@@ -791,6 +815,9 @@ public final class EmbeddedChannel: Channel {
         if option is ChannelOptions.Types.AutoReadOption {
             return true as! Option.Value
         }
+        if option is ChannelOptions.Types.AllowRemoteHalfClosureOption {
+            return self.allowRemoteHalfClosure as! Option.Value
+        }
         fatalError("option \(option) not supported")
     }
 

Tests/NIOEmbeddedTests/AsyncTestingChannelTests.swift

@@ -69,7 +69,7 @@ class AsyncTestingChannelTests: XCTestCase {
             XCTAssertNoThrow(try channel.pipeline.removeHandler(name: "handler2").wait())
         }
     }
-    
+
     func testWaitForInboundWrite() throws {
         guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { throw XCTSkip() }
         XCTAsyncTest {
@@ -79,14 +79,14 @@ class AsyncTestingChannelTests: XCTestCase {
                 try await XCTAsyncAssertEqual(try await channel.waitForInboundWrite(), 2)
                 try await XCTAsyncAssertEqual(try await channel.waitForInboundWrite(), 3)
             }
-            
+
             try await channel.writeInbound(1)
             try await channel.writeInbound(2)
             try await channel.writeInbound(3)
             try await task.value
         }
     }
-    
+
     func testWaitForMultipleInboundWritesInParallel() throws {
         guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { throw XCTSkip() }
         XCTAsyncTest {
@@ -101,14 +101,14 @@ class AsyncTestingChannelTests: XCTestCase {
                     try await task3.value,
                 ]), [1, 2, 3])
             }
-            
+
             try await channel.writeInbound(1)
             try await channel.writeInbound(2)
             try await channel.writeInbound(3)
             try await task.value
         }
     }
-    
+
     func testWaitForOutboundWrite() throws {
         guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { throw XCTSkip() }
         XCTAsyncTest {
@@ -118,14 +118,14 @@ class AsyncTestingChannelTests: XCTestCase {
                 try await XCTAsyncAssertEqual(try await channel.waitForOutboundWrite(), 2)
                 try await XCTAsyncAssertEqual(try await channel.waitForOutboundWrite(), 3)
             }
-            
+
             try await channel.writeOutbound(1)
             try await channel.writeOutbound(2)
             try await channel.writeOutbound(3)
             try await task.value
         }
     }
-    
+
     func testWaitForMultipleOutboundWritesInParallel() throws {
         guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { throw XCTSkip() }
         XCTAsyncTest {
@@ -140,7 +140,7 @@ class AsyncTestingChannelTests: XCTestCase {
                     try await task3.value,
                 ]), [1, 2, 3])
             }
-            
+
             try await channel.writeOutbound(1)
             try await channel.writeOutbound(2)
             try await channel.writeOutbound(3)
@@ -575,6 +575,35 @@ class AsyncTestingChannelTests: XCTestCase {
         }.wait()
     }
 
+    func testGetChannelOptionAutoReadIsSupported() throws {
+        guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { throw XCTSkip() }
+        let channel = NIOAsyncTestingChannel()
+        try channel.testingEventLoop.submit {
+            let options = channel.syncOptions
+            XCTAssertNotNil(options)
+            // Unconditionally returns true.
+            XCTAssertEqual(try options?.getOption(ChannelOptions.autoRead), true)
+        }.wait()
+    }
+
+    func testSetGetChannelOptionAllowRemoteHalfClosureIsSupported() throws {
+        guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { throw XCTSkip() }
+        let channel = NIOAsyncTestingChannel()
+        try channel.testingEventLoop.submit {
+            let options = channel.syncOptions
+            XCTAssertNotNil(options)
+
+            // allowRemoteHalfClosure should be false by default
+            XCTAssertEqual(try options?.getOption(ChannelOptions.allowRemoteHalfClosure), false)
+
+            channel.allowRemoteHalfClosure = true
+            XCTAssertEqual(try options?.getOption(ChannelOptions.allowRemoteHalfClosure), true)
+
+            XCTAssertNoThrow(try options?.setOption(ChannelOptions.allowRemoteHalfClosure, value: false))
+            XCTAssertEqual(try options?.getOption(ChannelOptions.allowRemoteHalfClosure), false)
+        }.wait()
+    }
+
     func testSecondFinishThrows() throws {
         guard #available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *) else { throw XCTSkip() }
         XCTAsyncTest {

Tests/NIOEmbeddedTests/EmbeddedChannelTest.swift

@@ -69,50 +69,50 @@ class ChannelLifecycleHandler: ChannelInboundHandler {
 }
 
 class EmbeddedChannelTest: XCTestCase {
-    
+
     func testSingleHandlerInit() {
         class Handler: ChannelInboundHandler {
             typealias InboundIn = Never
         }
-        
+
         let channel = EmbeddedChannel(handler: Handler())
         XCTAssertNoThrow(try channel.pipeline.handler(type: Handler.self).wait())
     }
-    
+
     func testSingleHandlerInitNil() {
         class Handler: ChannelInboundHandler {
             typealias InboundIn = Never
         }
-        
+
         let channel = EmbeddedChannel(handler: nil)
         XCTAssertThrowsError(try channel.pipeline.handler(type: Handler.self).wait()) { e in
             XCTAssertEqual(e as? ChannelPipelineError, .notFound)
         }
     }
-    
+
     func testMultipleHandlerInit() {
         class Handler: ChannelInboundHandler, RemovableChannelHandler {
             typealias InboundIn = Never
             let identifier: String
-            
+
             init(identifier: String) {
                 self.identifier = identifier
             }
         }
-        
+
         let channel = EmbeddedChannel(
             handlers: [Handler(identifier: "0"), Handler(identifier: "1"), Handler(identifier: "2")]
         )
         XCTAssertNoThrow(XCTAssertEqual(try channel.pipeline.handler(type: Handler.self).wait().identifier, "0"))
         XCTAssertNoThrow(try channel.pipeline.removeHandler(name: "handler0").wait())
-        
+
         XCTAssertNoThrow(XCTAssertEqual(try channel.pipeline.handler(type: Handler.self).wait().identifier, "1"))
         XCTAssertNoThrow(try channel.pipeline.removeHandler(name: "handler1").wait())
-        
+
         XCTAssertNoThrow(XCTAssertEqual(try channel.pipeline.handler(type: Handler.self).wait().identifier, "2"))
         XCTAssertNoThrow(try channel.pipeline.removeHandler(name: "handler2").wait())
     }
-    
+
     func testWriteOutboundByteBuffer() throws {
         let channel = EmbeddedChannel()
         var buf = channel.allocator.buffer(capacity: 1024)
@@ -481,13 +481,27 @@ class EmbeddedChannelTest: XCTestCase {
         XCTAssertEqual(invocations, 1)
     }
 
-    func testSyncOptionsAreSupported() throws {
+    func testGetChannelOptionAutoReadIsSupported() {
         let channel = EmbeddedChannel()
         let options = channel.syncOptions
         XCTAssertNotNil(options)
         // Unconditionally returns true.
         XCTAssertEqual(try options?.getOption(ChannelOptions.autoRead), true)
-        // (Setting options isn't supported.)
+    }
+
+    func testSetGetChannelOptionAllowRemoteHalfClosureIsSupported() {
+        let channel = EmbeddedChannel()
+        let options = channel.syncOptions
+        XCTAssertNotNil(options)
+
+        // allowRemoteHalfClosure should be false by default
+        XCTAssertEqual(try options?.getOption(ChannelOptions.allowRemoteHalfClosure), false)
+
+        channel.allowRemoteHalfClosure = true
+        XCTAssertEqual(try options?.getOption(ChannelOptions.allowRemoteHalfClosure), true)
+
+        XCTAssertNoThrow(try options?.setOption(ChannelOptions.allowRemoteHalfClosure, value: false))
+        XCTAssertEqual(try options?.getOption(ChannelOptions.allowRemoteHalfClosure), false)
     }
 
     func testLocalAddress0() throws {

docker/docker-compose.2004.56.yaml

@@ -23,8 +23,8 @@ services:
     environment:
       - MAX_ALLOCS_ALLOWED_10000000_asyncsequenceproducer=22
       - MAX_ALLOCS_ALLOWED_1000000_asyncwriter=1000050
-      - MAX_ALLOCS_ALLOWED_1000_addHandlers=44050
-      - MAX_ALLOCS_ALLOWED_1000_addHandlers_sync=38050
+      - MAX_ALLOCS_ALLOWED_1000_addHandlers=45050
+      - MAX_ALLOCS_ALLOWED_1000_addHandlers_sync=39050
       - MAX_ALLOCS_ALLOWED_1000_addRemoveHandlers_handlercontext=8050
       - MAX_ALLOCS_ALLOWED_1000_addRemoveHandlers_handlername=8050
       - MAX_ALLOCS_ALLOWED_1000_addRemoveHandlers_handlertype=8050
@@ -33,8 +33,9 @@ services:
       - MAX_ALLOCS_ALLOWED_1000_copying_bytebufferview_to_array=1050
       - MAX_ALLOCS_ALLOWED_1000_copying_circularbuffer_to_array=1050
       - MAX_ALLOCS_ALLOWED_1000_getHandlers=8050
-      - MAX_ALLOCS_ALLOWED_1000_getHandlers_sync=35
+      - MAX_ALLOCS_ALLOWED_1000_getHandlers_sync=36
       - MAX_ALLOCS_ALLOWED_1000_reqs_1_conn=26400
+      - MAX_ALLOCS_ALLOWED_1000_rst_connections=146000
       - MAX_ALLOCS_ALLOWED_1000_tcpbootstraps=4050
       - MAX_ALLOCS_ALLOWED_1000_tcpconnections=151050
       - MAX_ALLOCS_ALLOWED_1000_udp_reqs=6050
@@ -65,7 +66,7 @@ services:
       - MAX_ALLOCS_ALLOWED_read_10000_chunks_from_file=140050
       - MAX_ALLOCS_ALLOWED_schedule_10000_tasks=60100
       - MAX_ALLOCS_ALLOWED_schedule_and_run_10000_tasks=60050
-      - MAX_ALLOCS_ALLOWED_scheduling_10000_executions=88
+      - MAX_ALLOCS_ALLOWED_scheduling_10000_executions=86
       - MAX_ALLOCS_ALLOWED_udp_1000_reqs_1_conn=6200
       - MAX_ALLOCS_ALLOWED_udp_1_reqs_1000_conn=161050
       - FORCE_TEST_DISCOVERY=--enable-test-discovery