wip prototyping queue, tasks and timer

Author: tegefaulkes

package-lock.json

@@ -105,6 +105,7 @@
     "pako": "^1.0.11",
     "prompts": "^2.4.1",
     "readable-stream": "^3.6.0",
+    "real-cancellable-promise": "^1.1.1",
     "resource-counter": "^1.2.4",
     "threads": "^1.6.5",
     "utp-native": "^2.5.3",
@@ -117,7 +118,7 @@
     "@types/google-protobuf": "^3.7.4",
     "@types/jest": "^28.1.3",
     "@types/nexpect": "^0.4.31",
-    "@types/node": "^16.11.7",
+    "@types/node": "^16.11.49",
     "@types/node-forge": "^0.10.4",
     "@types/pako": "^1.0.2",
     "@types/prompts": "^2.0.13",

package.json

@@ -0,0 +1,4 @@
+// let's attempt the cancellable one as well
+// it requires the promise
+// we can avoid needing to use this in EFS for now
+// it's specific to PK

src/contexts/decorators/cancellable.ts

@@ -0,0 +1,18 @@
+import * as contextsUtils from '../utils';
+
+/**
+ * Context parameter decorator
+ * It is only allowed to be used once
+ */
+function context(target: Object, key: string | symbol, index: number) {
+  const targetName = (target['name'] ?? target.constructor.name);
+  const method = target[key];
+  if (contextsUtils.contexts.has(method)) {
+    throw new TypeError(
+      `\`${targetName}.${key.toString()}\` redeclares \`@context\` decorator`
+    );
+  }
+  contextsUtils.contexts.set(method, index);
+}
+
+export default context;

src/contexts/decorators/context.ts

@@ -0,0 +1,4 @@
+export { default as context } from './context';
+// export { default as cancellable }, * from './cancellable';
+export { default as timed } from './timed';
+// export { default as transactional }, * from './transactional';

src/contexts/decorators/index.ts

@@ -0,0 +1,128 @@
+import * as contextsUtils from '../utils';
+import * as contextsErrors from '../errors';
+import Timer from '../../timer/Timer';
+import * as timerErrors from '../../timer/errors';
+import {
+  AsyncFunction,
+  GeneratorFunction,
+  AsyncGeneratorFunction
+} from '../../utils';
+
+/**
+ * Timed method decorator
+ */
+function timed(delay: number = Infinity) {
+  return (
+    target: any,
+    key: string | symbol,
+    descriptor: TypedPropertyDescriptor<(...params: any[]) => any>
+  ): TypedPropertyDescriptor<(...params: any[]) => any>  => {
+    const targetName = (target['name'] ?? target.constructor.name);
+    const f = descriptor['value'];
+    if (typeof f !== 'function') {
+      throw new TypeError(`\`${targetName}.${key.toString()}\` is not a function`);
+    }
+    const contextIndex = contextsUtils.contexts.get(target[key]);
+    if (contextIndex == null) {
+      throw new TypeError(`\`${targetName}.${key.toString()}\` does not have a \`@context\` parameter decorator`);
+    }
+    const wrap = (that: any, params: Array<any>) => {
+      const context = params[contextIndex];
+      if (context !== undefined && (typeof context !== 'object' || context === null)) {
+        throw new TypeError(
+          `\`${targetName}.${key.toString()}\` decorated \`@context\` parameter is not a context object`
+        );
+      }
+      if (context?.timer !== undefined && !(context.timer instanceof Timer)) {
+        throw new TypeError(
+          `\`${targetName}.${key.toString()}\` decorated \`@context\` parameter's \`timer\` property is not an instance of \`Timer\``
+        );
+      }
+      if (context?.signal !== undefined && !(context.signal instanceof AbortSignal)) {
+        throw new TypeError(
+          `\`${targetName}.${key.toString()}\` decorated \`@context\` parameter's \`signal\` property is not an instance of \`AbortSignal\``
+        );
+      }
+      // Now `context: { timer: Timer | undefined; signal: AbortSignal | undefined } | undefined`
+      if (
+        context === undefined ||
+        context.timer === undefined && context.signal === undefined
+      ) {
+        const abortController = new AbortController();
+        const timer = new Timer({
+          delay,
+          handler: () => void abortController.abort(new contextsErrors.ErrorContextsTimerExpired)
+        });
+        params[contextIndex] = (context !== undefined) ? context : {};
+        params[contextIndex].signal = abortController.signal;
+        params[contextIndex].timer = timer;
+        const result = f.apply(that, params);
+        timer.catch((e) => {
+          // Ignore cancellation
+          if (!(e instanceof timerErrors.ErrorTimerCancelled)) {
+            throw e;
+          }
+        });
+        timer.cancel();
+        return result;
+      } else if (
+        context.timer === undefined &&
+        context.signal instanceof AbortSignal
+      ) {
+        const abortController = new AbortController();
+        const timer = new Timer({
+          delay,
+          handler: () => void abortController.abort(new contextsErrors.ErrorContextsTimerExpired)
+        });
+        context.signal.onabort = () => void abortController.abort(context.signal.reason);
+        params[contextIndex].signal = abortController.signal;
+        params[contextIndex].timer = timer;
+        const result = f.apply(that, params);
+        timer.catch((e) => {
+          // Ignore cancellation
+          if (!(e instanceof timerErrors.ErrorTimerCancelled)) {
+            throw e;
+          }
+        });
+        timer.cancel();
+        return result;
+      } else if (
+        context.timer instanceof Timer &&
+        context.signal === undefined
+      ) {
+        const abortController = new AbortController();
+        context.timer.then(() => void abortController.abort(new contextsErrors.ErrorContextsTimerExpired));
+        params[contextIndex].signal = abortController.signal;
+        return f.apply(that, params);
+      } else if (
+        context.timer instanceof Timer && context.signal instanceof AbortSignal
+      ) {
+        return f.apply(that, params);
+      }
+    };
+    if (f instanceof AsyncFunction) {
+      descriptor['value'] = async function (...params) {
+        return wrap(this, params);
+      };
+    } else if (f instanceof GeneratorFunction) {
+      descriptor['value'] = function* (...params) {
+        return yield* wrap(this, params);
+      };
+    } else if (f instanceof AsyncGeneratorFunction) {
+      descriptor['value'] = async function* (...params) {
+        return yield* wrap(this, params);
+      };
+    } else {
+      descriptor['value'] = function (...params) {
+        return wrap(this, params);
+      };
+    }
+    // Preserve the name
+    Object.defineProperty(descriptor['value'], 'name', {
+      value: (typeof key === 'symbol') ? `[${key.description}]` : key
+    });
+    return descriptor;
+  };
+}
+
+export default timed;

src/contexts/decorators/timed.ts

@@ -0,0 +1,13 @@
+import { ErrorPolykey, sysexits } from '../errors';
+
+class ErrorContexts<T> extends ErrorPolykey<T> {}
+
+class ErrorContextsTimerExpired<T> extends ErrorContexts<T> {
+  static description = 'Aborted due to timer expiration';
+  exitCode = sysexits.UNAVAILABLE;
+}
+
+export {
+  ErrorContexts,
+  ErrorContextsTimerExpired
+};

src/contexts/decorators/transactional.ts

@@ -0,0 +1,4 @@
+export * from './decorators';
+export * from './utils';
+export * as types from './types';
+export * as errors from './errors';

src/contexts/errors.ts

@@ -0,0 +1,20 @@
+import type { DBTransaction } from '@matrixai/db';
+import type Timer from '../timer/Timer';
+
+type ContextCancellable = {
+  signal: AbortSignal;
+};
+
+type ContextTimed = ContextCancellable & {
+  timer: Timer;
+};
+
+type ContextTransactional = {
+  tran: DBTransaction;
+};
+
+export type {
+  ContextCancellable,
+  ContextTimed,
+  ContextTransactional
+};