fix(#2553) : AsyncSequence.asObservable() runs on background thread

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file.
 * Provides `Infallible` versions of `combineLatest` without `resultSelector` requirement.
 * Provides `Infallible` versions of `CombineLatest+Collection` helpers.
 * Explicitly declare `APPLICATION_EXTENSION_API_ONLY` for CocoaPods 
+* Add a new parameter `detached: Bool = false` to the `AsyncSequence.asObservable` function to allow iterating over the `AsyncSequence` on a background thread. 
 
 ## 6.5.0
 

RxSwift/Observable+Concurrency.swift

@@ -58,9 +58,9 @@ public extension AsyncSequence {
     /// values of the asynchronous sequence's type
     ///
     /// - returns: An `Observable` of the async sequence's type
-    func asObservable() -> Observable<Element> {
+    func asObservable(detached: Bool = false) -> Observable<Element> {
         Observable.create { observer in
-            let task = Task {
+            let taskBlock: @Sendable () async -> Void = {
                 do {
                     for try await value in self {
                         observer.onNext(value)
@@ -73,7 +73,11 @@ public extension AsyncSequence {
                     observer.onError(error)
                 }
             }
-
+
+            let task: Task<Void, Never> = detached
+                ? Task.detached(operation: taskBlock)
+                : Task(operation: taskBlock)
+
             return Disposables.create { task.cancel() }
         }
     }

Tests/RxSwiftTests/Observable+ConcurrencyTests.swift

@@ -90,5 +90,101 @@ extension ObservableConcurrencyTests {
         task.cancel()
     }
 
+    func testAsyncSequenceToObservableRunsOnBackgroundThread() async throws {
+
+        let asyncSequence = AsyncStream<Int> { continuation in
+            for i in 1...5 {
+                continuation.yield(i)
+            }
+            continuation.finish()
+        }
+
+        let expectation = XCTestExpectation(description: "Observable completes")
+
+        DispatchQueue.main.async {
+            let observable = asyncSequence.asObservable(detached: true)
+
+            var threadIsNotMain = false
+            var values = [Int]()
+
+            _ = observable.subscribe(
+                onNext: { value in
+                    values.append(value)
+                    threadIsNotMain = !Thread.isMainThread
+                },
+                onCompleted: {
+                    XCTAssertEqual(values, [1, 2, 3, 4, 5])
+                    XCTAssertTrue(threadIsNotMain, "AsyncSequence.asObservable should not run on main thread")
+                    expectation.fulfill()
+                }
+            )
+        }
+
+        await fulfillment(of: [expectation], timeout: 5.0)
+    }
+
+    func testAsyncSequenceToObservableWithSleep() async throws {
+        let asyncSequence = AsyncStream<Int> { continuation in
+            Task {
+                for i in 1...3 {
+                    try? await Task.sleep(nanoseconds: 100_000_000)
+                    continuation.yield(i)
+                }
+                continuation.finish()
+            }
+        }
+
+        let expectation = XCTestExpectation(description: "Observable with sleep completes")
+
+        DispatchQueue.main.async {
+            let startTime = Date()
+            var values = [Int]()
+            var executionThreads = Set<String>()
+
+            _ = asyncSequence.asObservable(detached: true).subscribe(
+                onNext: { value in
+                    values.append(value)
+                    let threadName = Thread.current.description
+                    executionThreads.insert(threadName)
+                },
+                onCompleted: {
+                    let duration = Date().timeIntervalSince(startTime)
+                    XCTAssertGreaterThanOrEqual(duration, 0.3)
+                    XCTAssertEqual(values, [1, 2, 3])
+                    XCTAssertFalse(executionThreads.contains(where: { $0.contains("main") }))
+
+                    expectation.fulfill()
+                }
+            )
+        }
+
+        await fulfillment(of: [expectation], timeout: 5.0)
+    }
+
+    func testAsyncSequenceToObservableWithError() async throws {
+        struct TestError: Error {}
+
+        let asyncSequence = AsyncThrowingStream<Int, Error> { continuation in
+            for i in 1...3 {
+                continuation.yield(i)
+            }
+            continuation.finish(throwing: TestError())
+        }
+
+        let expectation = XCTestExpectation(description: "Observable with error completes")
+        var receivedError: Error?
+
+        _ = asyncSequence.asObservable(detached: true).subscribe(
+            onNext: { _ in },
+            onError: { error in
+                receivedError = error
+                expectation.fulfill()
+            }
+        )
+
+        await fulfillment(of: [expectation], timeout: 5.0)
+        XCTAssertTrue(receivedError is TestError)
+    }
+
 }
 #endif