feat: Add detached parameter to AsyncSequence.asObservable()

Author: suojae

RxSwift/Observable+Concurrency.swift

@@ -58,22 +58,25 @@ 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.detached {
+            let taskBlock = {
                 do {
                     for try await value in self {
                         observer.onNext(value)
                     }
-
                     observer.onCompleted()
                 } catch is CancellationError {
                     observer.onCompleted()
                 } catch {
                     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

@@ -102,7 +102,7 @@ extension ObservableConcurrencyTests {
         let expectation = XCTestExpectation(description: "Observable completes")
 
         DispatchQueue.main.async {
-            let observable = asyncSequence.asObservable()
+            let observable = asyncSequence.asObservable(detached: true)
 
             var threadIsNotMain = false
             var values = [Int]()
@@ -141,7 +141,7 @@ extension ObservableConcurrencyTests {
             var values = [Int]()
             var executionThreads = Set<String>()
 
-            _ = asyncSequence.asObservable().subscribe(
+            _ = asyncSequence.asObservable(detached: true).subscribe(
                 onNext: { value in
                     values.append(value)
                     let threadName = Thread.current.description
@@ -178,7 +178,7 @@ extension ObservableConcurrencyTests {
         var threadIsNotMain = false
 
         DispatchQueue.main.async {
-            _ = asyncSequence.asObservable().subscribe(
+            _ = asyncSequence.asObservable(detached: true).subscribe(
                 onNext: { value in
                     values.append(value)
                     threadIsNotMain = !Thread.isMainThread