Add more methods to work with Isolated futures

Sources/NIOCore/EventLoop.swift

@@ -1107,6 +1107,22 @@ extension EventLoop {
         }
     }
 
+    /// Creates and returns a new isolated `EventLoopFuture` that is already marked as success. Notifications will be done using this `EventLoop` as execution `NIOThread`.
+    ///
+    /// - Parameters:
+    ///   - value: the value that is used by the `EventLoopFuture.Isolated`.
+    /// - Returns: a succeeded `EventLoopFuture.Isolated`.
+    @inlinable
+    @available(*, noasync)
+    public func makeSucceededIsolatedFuture<Success>(_ value: Success) -> EventLoopFuture<Success>.Isolated {
+        if Success.self == Void.self {
+            // The as! will always succeed because we previously checked that Success.self == Void.self.
+            return self.makeSucceededVoidFuture().assumeIsolated() as! EventLoopFuture<Success>.Isolated
+        } else {
+            return EventLoopFuture.Isolated(_wrapped: EventLoopFuture(eventLoop: self, isolatedValue: value))
+        }
+    }
+
     /// Creates and returns a new `EventLoopFuture` that is marked as succeeded or failed with the value held by `result`.
     ///
     /// - Parameters:

Sources/NIOCore/EventLoopFuture+AssumeIsolated.swift

@@ -389,6 +389,43 @@ extension EventLoopFuture {
             return next.futureResult.assumeIsolatedUnsafeUnchecked()
         }
 
+        /// When the current `EventLoopFuture<Value>.Isolated` is in an error state, run the provided callback, which
+        /// may recover from the error by returning an `EventLoopFuture<NewValue>.Isolated`. The callback is intended to potentially
+        /// recover from the error by returning a new `EventLoopFuture.Isolated` that will eventually contain the recovered
+        /// result.
+        ///
+        /// If the callback cannot recover it should return a failed `EventLoopFuture.Isolated`.
+        ///
+        /// - Note: The `Value` need not be `Sendable` since the isolation domains of this future and the future returned from the callback
+        /// must be the same
+        ///
+        /// - Parameters:
+        ///   - callback: Function that will receive the error value of this `EventLoopFuture.Isolated` and return
+        ///         a new value lifted into a new `EventLoopFuture.Isolated`.
+        /// - Returns: A future that will receive the recovered value.
+        @inlinable
+        @available(*, noasync)
+        public func flatMapError(
+            _ callback: @escaping (Error) -> EventLoopFuture<Value>.Isolated
+        ) -> EventLoopFuture<Value>.Isolated {
+            let next = EventLoopPromise<Value>.makeUnleakablePromise(eventLoop: self._wrapped.eventLoop)
+            let base = self._wrapped
+
+            base._whenCompleteIsolated {
+                switch base._value! {
+                case .success(let t):
+                    return next._setValue(value: .success(t))
+                case .failure(let e):
+                    let t = callback(e)
+                    t._wrapped.eventLoop.assertInEventLoop()
+                    return t._wrapped._addCallback {
+                        next._setValue(value: t._wrapped._value!)
+                    }
+                }
+            }
+            return next.futureResult.assumeIsolatedUnsafeUnchecked()
+        }
+
         /// When the current `EventLoopFuture<Value>` is fulfilled, run the provided callback, which
         /// performs a synchronous computation and returns either a new value (of type `NewValue`) or
         /// an error depending on the `Result` returned by the closure.
@@ -643,6 +680,11 @@ extension EventLoopPromise {
             self._wrapped = _wrapped
         }
 
+        @usableFromInline
+        var futureResult: EventLoopFuture<Value>.Isolated {
+            self._wrapped.futureResult.assumeIsolated()
+        }
+
         /// Deliver a successful result to the associated `EventLoopFuture<Value>` object.
         ///
         /// - Parameters:
@@ -708,3 +750,4 @@ extension EventLoopPromise {
 
 @available(*, unavailable)
 extension EventLoopPromise.Isolated: Sendable {}
+

Sources/NIOCore/EventLoopFuture.swift

@@ -437,6 +437,16 @@ public final class EventLoopFuture<Value> {
         self._callbacks = .init()
     }
 
+    /// A EventLoopFuture<Value> that has already succeeded with an isolated (not-necessarily-sendable) value
+    @inlinable
+    internal init(eventLoop: EventLoop, isolatedValue value: Value) {
+        eventLoop.assertInEventLoop()
+
+        self.eventLoop = eventLoop
+        self._value = .success(value)
+        self._callbacks = .init()
+    }
+
     /// A EventLoopFuture<Value> that has already failed
     @inlinable
     internal init(eventLoop: EventLoop, error: Error) {

Tests/NIOPosixTests/EventLoopFutureIsolatedTests.swift

@@ -288,6 +288,19 @@ final class EventLoopFutureIsolatedTest: XCTestCase {
                 }
                 XCTAssertEqual(r, originalValue.x - 1)
             }
+            throwingFuture.map { _ in 5 }.flatMapError { (error: any Error) -> EventLoopFuture<Int>.Isolated in
+                guard let error = error as? TestError, error == .error else {
+                    XCTFail("Invalid passed error: \(error)")
+                    return loop.makeSucceededIsolatedFuture(originalValue.x)
+                }
+                return loop.makeSucceededIsolatedFuture(originalValue.x - 2)
+            }.whenComplete { (result: Result<Int, any Error>) in
+                guard case .success(let r) = result else {
+                    XCTFail("Unexpected error")
+                    return
+                }
+                XCTAssertEqual(r, originalValue.x - 2)
+            }
 
             // This block handles unwrap.
             newFuture.map { x -> SuperNotSendable? in

Tests/NIOPosixTests/EventLoopFutureTest.swift

@@ -1110,6 +1110,30 @@ class EventLoopFutureTest: XCTestCase {
         XCTAssertEqual(try promise.futureResult.wait(), "yay")
     }
 
+    func testFutureFulfilledIfHasNonSendableResult() throws {
+        let eventLoop = EmbeddedEventLoop()
+        let f = EventLoopFuture(eventLoop: eventLoop, isolatedValue: NonSendableObject(value: 5))
+        XCTAssertTrue(f.isFulfilled)
+    }
+
+    func testSucceededIsolatedFutureIsCompleted() throws {
+        let group = EmbeddedEventLoop()
+        let loop = group.next()
+
+        let value = NonSendableObject(value: 4)
+
+        let future = loop.makeSucceededIsolatedFuture(value)
+
+        future.whenComplete { result in
+            switch result{
+            case .success(let nonSendableStruct):
+                XCTAssertEqual(nonSendableStruct, value)
+            case .failure(let error):
+                XCTFail("\(error)")
+            }
+        }
+    }
+
     func testPromiseCompletedWithFailedFuture() throws {
         let group = EmbeddedEventLoop()
         let loop = group.next()
@@ -1652,3 +1676,16 @@ class EventLoopFutureTest: XCTestCase {
         XCTAssertNotEqual(promise3, promise2)
     }
 }
+
+class NonSendableObject: Equatable {
+    var value: Int
+    init(value: Int) {
+        self.value = value
+    }
+
+    static func == (lhs: NonSendableObject, rhs: NonSendableObject) -> Bool {
+        lhs.value == rhs.value
+    }
+}
+@available(*, unavailable)
+extension NonSendableObject: Sendable {}