feat(add): withAsync

Package.resolved

@@ -1,25 +1,14 @@
 {
-  "object": {
-    "pins": [
-      {
-        "package": "RxSwift",
-        "repositoryURL": "https://github.com/ReactiveX/RxSwift.git",
-        "state": {
-          "branch": null,
-          "revision": "b4307ba0b6425c0ba4178e138799946c3da594f8",
-          "version": "6.5.0"
-        }
-      },
-      {
-        "package": "SwiftDocCPlugin",
-        "repositoryURL": "https://github.com/apple/swift-docc-plugin",
-        "state": {
-          "branch": null,
-          "revision": "3303b164430d9a7055ba484c8ead67a52f7b74f6",
-          "version": "1.0.0"
-        }
+  "pins" : [
+    {
+      "identity" : "rxswift",
+      "kind" : "remoteSourceControl",
+      "location" : "https://github.com/ReactiveX/RxSwift.git",
+      "state" : {
+        "revision" : "b4307ba0b6425c0ba4178e138799946c3da594f8",
+        "version" : "6.5.0"
       }
-    ]
-  },
-  "version": 1
+    }
+  ],
+  "version" : 2
 }

Sources/RxConcurrency/RxSwift+Completable.swift

@@ -20,10 +20,28 @@ public extension PrimitiveSequenceType where Trait == CompletableTrait, Element
                     completable(.error(error))
                 }
             }
-
+
+            return Disposables.create {
+                task.cancel()
+            }
+        }
+    }
+
+    static func async<T: AnyObject>(with object: T, handler: @escaping (T) async throws -> Void) -> Completable {
+        Completable.create { completable in
+            let task = Task { [weak object] in
+                guard let object = object else { throw Error.unknown }
+                do {
+                    try await handler(object)
+                    completable(.completed)
+                } catch {
+                    completable(.error(error))
+                }
+            }
+
             return Disposables.create {
                 task.cancel()
             }
         }
     }
-}
+}

Sources/RxConcurrency/RxSwift+Observable.swift

@@ -5,6 +5,10 @@
 import Foundation
 import RxSwift
 
+enum Error: Swift.Error {
+    case unknown
+}
+
 public extension ObservableType {
     /// Create observable that emit value from async throws function.
     ///
@@ -20,10 +24,28 @@ public extension ObservableType {
                     observer.on(.error(error))
                 }
             }
-
+
+            return Disposables.create {
+                task.cancel()
+            }
+        }
+    }
+
+    static func async<T: AnyObject>(with object: T, handler: @escaping (T) async throws -> Element) -> Observable<Element> {
+        Observable<Element>.create { observer in
+            let task = Task { [weak object] in
+                guard let object = object else { throw Error.unknown }
+                do {
+                    observer.on(.next(try await handler(object)))
+                    observer.on(.completed)
+                } catch {
+                    observer.on(.error(error))
+                }
+            }
+
             return Disposables.create {
                 task.cancel()
             }
         }
     }
-}
+}

Sources/RxConcurrency/RxSwift+Single.swift

@@ -19,7 +19,24 @@ public extension PrimitiveSequenceType where Trait == SingleTrait {
                     single(.failure(error))
                 }
             }
-
+
+            return Disposables.create {
+                task.cancel()
+            }
+        }
+    }
+
+    static func async<T: AnyObject>(with object: T, handler: @escaping (T) async throws -> Element) -> Single<Element> {
+        Single<Element>.create { single in
+            let task = Task { [weak object] in
+                guard let object = object else { throw Error.unknown }
+                do {
+                    single(.success(try await handler(object)))
+                } catch {
+                    single(.failure(error))
+                }
+            }
+
             return Disposables.create {
                 task.cancel()
             }

Tests/RxConcurrencyTests/RxSwift+ObservableTests.swift

@@ -64,4 +64,4 @@ final class RxSwiftObservableTests: XCTestCase {
     }
 }
 
-extension String: Error {}
+extension String: Swift.Error {}