feat: add observeAsync function

Author: djcsdy

index.ts

@@ -0,0 +1,62 @@
+import {Observable, Subscription} from "rxjs";
+
+export interface Sink<T> {
+    subscribed: () => boolean;
+    emit: (value: T) => void;
+    emitAll: (values: Observable<T>) => Promise<void>;
+}
+
+export class EmitAfterUnsubscribeError extends Error {
+    public constructor() {
+        super("Emitted a value after subscriber unsubscribed");
+    }
+}
+
+export function observeAsync<T>(action: (sink: Sink<T>) => Promise<void>): Observable<T> {
+    return new Observable<T>(subscriber => {
+        let subscribed = true;
+        const subscription = new Subscription(() => {
+            subscribed = false;
+        });
+
+        const emit = (value: T): void => {
+            if (subscribed) {
+                void subscriber.next(value);
+            } else {
+                throw new EmitAfterUnsubscribeError();
+            }
+        };
+
+        void action({
+            subscribed: () => subscribed,
+            emit,
+            emitAll: async values =>
+                new Promise((resolve, reject) => {
+                    const innerSubscription = values.subscribe({
+                        next: value => void emit(value),
+                        error: (error: unknown) => {
+                            subscriber.error(error);
+                            reject(error);
+                        },
+                        complete: () => {
+                            resolve();
+                            subscription.remove(innerSubscription);
+                        }
+                    });
+                    subscription.add(innerSubscription);
+                })
+        }).then(
+            () => {
+                subscriber.complete();
+                subscription.unsubscribe();
+            },
+            (error: unknown) => {
+                subscriber.error(error);
+                subscriber.complete();
+                subscription.unsubscribe();
+            }
+        );
+
+        return subscription;
+    });
+}

package.json

@@ -26,6 +26,7 @@
     "node": "^18 || ^20 || >=21"
   },
   "dependencies": {
+    "rxjs": "^6.0.0 || ^7.0.0",
     "tslib": "^2.6.2"
   },
   "devDependencies": {

tsconfig.json

@@ -1,7 +1,8 @@
 {
   "extends": "@softwareventures/tsconfig/commonjs",
   "compilerOptions": {
-    "types": []
+    "types": [],
+    "skipLibCheck": true
   },
   "exclude": ["**/*.test.ts", "**/test.ts"]
 }

yarn.lock

@@ -1036,6 +1036,7 @@ __metadata:
     inquirer: "npm:9.2.17"
     pinst: "npm:3.0.0"
     prettier: "npm:3.2.5"
+    rxjs: "npm:^7.8.1"
     semantic-release: "npm:23.0.6"
     ts-node: "npm:10.9.2"
     tslib: "npm:^2.6.2"