feat(core): adds resolvableWait and safeTrigger utils

rafamel · Oct 28, 2019 · dd35d51 · dd35d51
1 parent ffeb61b
commit dd35d51
Show file tree

Hide file tree

Showing 8 changed files with 107 additions and 17 deletions.
diff --git a/packages/core/package-lock.json b/packages/core/package-lock.json
diff --git a/packages/core/package.json b/packages/core/package.json
@@ -81,7 +81,8 @@
     "ajv": "^6.10.2",
     "camelcase": "^5.3.1",
     "json-schema-to-typescript": "^7.1.0",
-    "lodash.isequal": "^4.5.0"
+    "lodash.isequal": "^4.5.0",
+    "promist": "^1.0.0"
   },
   "peerDependencies": {
     "rxjs": "6.x"

diff --git a/packages/core/src/utils/index.ts b/packages/core/src/utils/index.ts
@@ -1 +1,3 @@
+export * from './resolvable-wait';
 export * from './safe-promise';
+export * from './safe-trigger';
diff --git a/packages/core/src/utils/resolvable-wait.ts b/packages/core/src/utils/resolvable-wait.ts
@@ -0,0 +1,24 @@
+import { deferred } from 'promist';
+
+export interface ResolvableWait {
+  promise: Promise<void>;
+  resolve: () => void;
+}
+
+export function resolvableWait(ms: number): ResolvableWait {
+  const promise = deferred<void>();
+  let timer: null | NodeJS.Timer = setTimeout(resolve, ms);
+
+  function resolve(): void {
+    if (!timer) return;
+
+    clearTimeout(timer);
+    timer = null;
+    promise.resolve();
+  }
+
+  return {
+    promise: promise.then((x) => x),
+    resolve: resolve
+  };
+}
diff --git a/packages/core/src/utils/safe-promise.ts b/packages/core/src/utils/safe-promise.ts
@@ -1,5 +1,6 @@
 import { Observable } from 'rxjs';
 import { take } from 'rxjs/operators';
+import { safeTrigger } from './safe-trigger';
 
 /**
  * Converts an *Observable* into a *Promise* that will throw if the observable
@@ -10,15 +11,24 @@ export function toSafePromise<T>(observable: Observable<T>): Promise<T> {
     const subscription = observable.pipe(take(1)).subscribe({
       next(value) {
         resolve(value);
-        setTimeout(() => subscription.unsubscribe, 0);
+        safeTrigger(
+          () => Boolean(subscription),
+          () => subscription.unsubscribe()
+        );
       },
       error(err) {
         reject(err);
-        setTimeout(() => subscription.unsubscribe, 0);
+        safeTrigger(
+          () => Boolean(subscription),
+          () => subscription.unsubscribe()
+        );
       },
       complete() {
         reject(Error(`Source completed before emitting a result`));
-        setTimeout(() => subscription.unsubscribe, 0);
+        safeTrigger(
+          () => Boolean(subscription),
+          () => subscription.unsubscribe()
+        );
       }
     });
   });

diff --git a/packages/core/src/utils/safe-trigger.ts b/packages/core/src/utils/safe-trigger.ts
@@ -0,0 +1,17 @@
+import { waitUntil } from 'promist';
+
+/**
+ * Will execute `test` forever until it doesn't throw or reject and returns true, and only then will it execute `fn`.
+ */
+export function safeTrigger(
+  test: () => boolean | Promise<boolean>,
+  fn: () => void
+): void {
+  waitUntil(async () => {
+    try {
+      return await test();
+    } catch (err) {
+      return false;
+    }
+  }).then(() => fn());
+}
diff --git a/packages/intercepts/package.json b/packages/intercepts/package.json
@@ -59,7 +59,6 @@
     "url": "https://github.com/rafamel/karmic/issues"
   },
   "devDependencies": {
-    "@karmic/core": "0.0.0",
     "@pika/pack": "^0.4.0",
     "@pika/plugin-build-web": "^0.6.1",
     "@pika/plugin-standard-pkg": "^0.6.1",
@@ -73,11 +72,11 @@
     "typescript": "^3.6.4"
   },
   "dependencies": {
+    "@karmic/core": "0.0.0",
     "ajv": "^6.10.2",
     "errorish": "^0.4.0"
   },
   "peerDependencies": {
-    "@karmic/core": "0.0.0",
     "rxjs": "6.x"
   },
   "@pika/pack": {

diff --git a/packages/intercepts/src/logging.ts b/packages/intercepts/src/logging.ts
@@ -2,7 +2,8 @@ import {
   intercept,
   PublicError,
   InterceptImplementation,
-  ServiceKind
+  ServiceKind,
+  safeTrigger
 } from '@karmic/core';
 import { tap } from 'rxjs/operators';
 import { Observable } from 'rxjs';
@@ -133,11 +134,11 @@ export function logging(
           error: opts.skip.includes('error')
             ? (err: Error) => {
                 self.error(err);
-                setTimeout(() => subscription.unsubscribe(), 0);
+                unsubscribe();
               }
             : (err: Error) => {
                 self.error(err);
-                setTimeout(() => subscription.unsubscribe(), 0);
+                unsubscribe();
                 fn({
                   status: 'error',
                   ...base,
@@ -148,11 +149,11 @@ export function logging(
           complete: opts.skip.includes('complete')
             ? () => {
                 self.complete();
-                setTimeout(() => subscription.unsubscribe(), 0);
+                unsubscribe();
               }
             : () => {
                 self.complete();
-                setTimeout(() => subscription.unsubscribe(), 0);
+                unsubscribe();
                 fn({
                   status: 'complete',
                   ...base,
@@ -162,10 +163,17 @@ export function logging(
               }
         });
 
+        function unsubscribe(): void {
+          return safeTrigger(
+            () => Boolean(subscription),
+            () => subscription.unsubscribe()
+          );
+        }
+
         return opts.skip.includes('unsubscribe')
-          ? () => subscription.unsubscribe()
+          ? unsubscribe
           : () => {
-              subscription.unsubscribe();
+              unsubscribe();
               fn({
                 status: 'unsubscribe',
                 ...base,