Worker-based Async API with await/Promise support (or plain callbacks) (#6)

* fix lint errors in eslint config :)

* index.d.s: add /* eslint-disable @typescript-eslint/triple-slash-reference */

* improve eslint calling script

* add async-worker version of the SRT JS API with await/Promise support
+ potentially out-of-order RPC-back/result-dequeuing is possible
with a type of call-ID generator and callback-map that we prototyped
(but not used atm since the worker is only doing sync/blocking internally
with the current SRT lib binding).

* add various examples for async API ("classic" callbacks / Promise / await)

* add .eslintignore (should go with commit where we just call "eslint .")

* index.js: add missing semi

* async.js: fix method name litteral in epollUWait

* async.js: allow for accept method to use a timeout opt (defaults to false),
and a custom timeout value option (default to default timeout),
which can potentially be set differently than the general default timeout.
+ make the timeout value a static class property so that it can be
user-defined module-load wide. defaults to constant in module top scope.

* async.js: fix for a rejected promise, make sure we don't resolve anymore
+ add a custom timeout value argument to _createAsyncWorkPromise

* async.js: rm an experiment

* fix a lint error in example

* add types for async api and fix some details on binding API types

* export async API to index

* async.js: add some missing docs for private methods

* include async types in index

.eslintignore

@@ -0,0 +1,4 @@
+build
+deps
+node_modules
+tsc-lib

.eslintrc.js

@@ -1,25 +1,25 @@
 module.exports = {
-    root: true,
-    "env": {
-        "commonjs": true,
-        "es2020": true,
-        "node": true,
-        "jasmine": true
-    },
-    "extends": [
-        "eslint:recommended",
-        "plugin:@typescript-eslint/recommended",
-        "plugin:jasmine/recommended"
-    ],
-    "parser": "@typescript-eslint/parser",
-    "parserOptions": {
-        "ecmaVersion": 11
-    },
-    "plugins": ["@typescript-eslint", "jasmine"],
-    "rules": {
-        "indent": ["error", 2],
-        "semi": "error",
-        "@typescript-eslint/no-var-requires": 0,
-        "no-constant-condition": 0
-    }
+  root: true,
+  "env": {
+    "commonjs": true,
+    "es2020": true,
+    "node": true,
+    "jasmine": true
+  },
+  "extends": [
+    "eslint:recommended",
+    "plugin:@typescript-eslint/recommended",
+    "plugin:jasmine/recommended"
+  ],
+  "parser": "@typescript-eslint/parser",
+  "parserOptions": {
+    "ecmaVersion": 11
+  },
+  "plugins": ["@typescript-eslint", "jasmine"],
+  "rules": {
+    "indent": ["error", 2],
+    "semi": "error",
+    "@typescript-eslint/no-var-requires": 0,
+    "no-constant-condition": 0
+  }
 };

examples/async-srt-await.js

@@ -0,0 +1,25 @@
+const { AsyncSRT } = require('../src/async.js');
+
+const asyncSrt = new AsyncSRT();
+
+(async function() {
+  const socket = await asyncSrt.createSocket(false);
+  console.log('createSocket() result:', socket);
+  let result = await asyncSrt.bind(socket, "0.0.0.0", 1234);
+  console.log('bind() result:', result);
+  result = await asyncSrt.listen(socket, 2);
+  console.log('listen() result:', result);
+
+  awaitConnections(socket);
+
+})();
+
+async function awaitConnections(socket) {
+  console.log('Awaiting incoming client connection ...');
+  const fd = await asyncSrt.accept(socket);
+  console.log('New incoming client fd:', fd);
+}
+
+setInterval(() => {
+  console.log('Doing other stuff in the meantime ... :)');
+}, 1000);

examples/async-srt-promises.js

@@ -0,0 +1,32 @@
+"use strict";
+
+const { AsyncSRT } = require('../src/async.js');
+
+const asyncSrt = new AsyncSRT();
+
+let mySocket;
+
+asyncSrt.createSocket(false)
+  .catch((err) => console.error(err))
+  .then((result) => {
+    console.log('createSocket:', result);
+    mySocket = result;
+    return result;
+  })
+  .then((socket) => asyncSrt.bind(socket, "0.0.0.0", 1234))
+  .then((result) => {
+    if (result !== 0) {
+      throw new Error('Failed bind');
+    }
+    console.log('Bind success');
+    return asyncSrt.listen(mySocket, 2);
+  })
+  .then((result) => {
+    if (!result) {
+      console.log("Listen success");
+    } else {
+      throw new Error('SRT listen error: ' + result);
+    }
+  });
+
+

examples/async-srt.js

@@ -0,0 +1,25 @@
+"use strict";
+
+const { AsyncSRT } = require('../src/async.js');
+
+const asyncSrt = new AsyncSRT();
+
+asyncSrt.createSocket(false, (result) => {
+  console.log('createSocket:', result);
+  const socket = result;
+  asyncSrt.bind(socket, "0.0.0.0", 1234, (result) => {
+    if (result !== 0) {
+      console.log('Failed bind');
+    } else {
+      console.log('Bind success');
+      asyncSrt.listen(socket, 2, (result) => {
+        if (!result) {
+          console.log("Listen success");
+        } else {
+          console.log(result);
+        }
+      });
+    }
+  });
+});
+

index.d.ts

@@ -1,5 +1,7 @@
+/* eslint-disable @typescript-eslint/triple-slash-reference */
 /// <reference path="./types/srt-api.d.ts" />
 /// <reference path="./types/srt-stream.d.ts" />
 /// <reference path="./types/srt-server.d.ts" />
+/// <reference path="./types/srt-api-async.d.ts" />
 
 export * from "srt";

index.js

@@ -1,10 +1,12 @@
-const LIB = require('./build/Release/node_srt.node');
+const { SRT } = require('./build/Release/node_srt.node');
 const Server = require('./src/server.js');
 const { SRTReadStream, SRTWriteStream } = require('./src/stream.js');
+const { AsyncSRT } = require('./src/async');
 
 module.exports = {
-  SRT: LIB.SRT,
-  Server: Server,
+  SRT,
+  Server,
   SRTReadStream,
-  SRTWriteStream
-}
+  SRTWriteStream,
+  AsyncSRT,
+};

package.json

@@ -9,7 +9,7 @@
     "rebuild": "node-gyp rebuild",
     "clean": "node-gyp clean",
     "test": "$(npm bin)/jasmine",
-    "lint": "eslint src examples types --ext .js --ext .ts",
+    "lint": "eslint . --ext .js --ext .ts",
     "check-tsc": "./node_modules/.bin/tsc examples/srt.ts --outDir ./tsc-lib",
     "postversion": "git push && git push --tags"
   },

src/async-worker.js

@@ -0,0 +1,41 @@
+const {
+  Worker, isMainThread, parentPort, workerData
+} = require('worker_threads');
+
+const { SRT } = require('../build/Release/node_srt.node');
+
+if (isMainThread) {
+  throw new Error("Worker module can not load on main thread");
+}
+
+(function run() {
+  const libSRT = new SRT();
+  parentPort.on('message', (data) => {
+    if (!data.method) {
+      throw new Error('Worker message needs `method` property');
+    }
+    /*
+    if (!data.workId) {
+      throw new Error('Worker message needs `workId` property');
+    }
+    */
+    let result = libSRT[data.method].apply(libSRT, data.args);
+    // TODO: see if we can do this using SharedArrayBuffer for example,
+    //       or just leveraging Transferable objects capabilities ... ?
+    // FIXME: Performance ... ?
+    if (result instanceof Buffer) {
+      const buf = Buffer.allocUnsafe(result.length);
+      result.copy(buf);
+      result = buf;
+    }
+    parentPort.postMessage({
+      // workId: data.workId,
+      timestamp: data.timestamp,
+      result
+    });
+  });
+})();
+
+
+
+

src/async.js

@@ -0,0 +1,242 @@
+const {
+  Worker, isMainThread, parentPort, workerData
+} = require('worker_threads');
+
+const path = require('path');
+const {performance} = require("perf_hooks");
+
+const DEFAULT_PROMISE_TIMEOUT_MS = 3000;
+
+/*
+const WORK_ID_GEN_MOD = 0xFFF;
+*/
+
+class AsyncSRT {
+
+  /**
+   * @static
+   * @type {number} Promise-timeout in millis
+   */
+  static TimeoutMs = DEFAULT_PROMISE_TIMEOUT_MS;
+
+  constructor() {
+    this._worker = new Worker(path.resolve(__dirname, './async-worker.js'));
+    this._worker.on('message', this._onWorkerMessage.bind(this));
+    /*
+    this._workIdGen = 0;
+    this._workCbMap = new Map();
+    */
+    this._workCbQueue = [];
+  }
+
+  /**
+   * @private
+   * @param {*} data
+   */
+  _onWorkerMessage(data) {
+    const resolveTime = performance.now();
+    const {timestamp, result, workId} = data;
+
+    const callback = this._workCbQueue.shift();
+    callback(result);
+  }
+
+  /**
+   * @private
+   * @param {string} method
+   * @param {Array<any>} args
+   * @param {Function} callback
+   */
+  _postAsyncWork(method, args, callback) {
+    const timestamp = performance.now();
+
+    // not really needed atm,
+    // only if the worker spawns async jobs itself internally
+    // and thus the queuing order of jobs would not be preserved
+    // across here and the worker side.
+    /*
+    if (this._workCbMap.size >= WORK_ID_GEN_MOD - 1) {
+      throw new Error('Can`t post more async work: Too many awaited callbacks unanswered in queue');
+    }
+    const workId = this._workIdGen;
+    this._workIdGen = (this._workIdGen + 1) % WORK_ID_GEN_MOD;
+    this._workCbMap.set(workId, callback);
+    */
+
+    this._workCbQueue.push(callback);
+    this._worker.postMessage({method, args, /*workId,*/ timestamp});
+  }
+
+  /**
+   * @private
+   * @param {string} method
+   * @param {Array<any>} args optional
+   * @param {Function} callback optional
+   */
+  _createAsyncWorkPromise(method, args = [], callback = null, useTimeout = true, timeoutMs = AsyncSRT.TimeoutMs) {
+    return new Promise((resolve, reject) => {
+      let timeout;
+      let rejected = false;
+      const onResult = (result) => {
+        // Q: signal somehow to app that timed-out call has had result after all? (only in case of using Promise..?)
+        if (rejected) {
+          // The reject thing only makes sense for Promise,
+          // and users can manage this aspect themselves when using plain callbacks.
+          if (callback) callback(result);
+          return;
+        } else if (useTimeout) clearTimeout(timeout);
+        resolve(result);
+        if (callback) callback(result); // NOTE: the order doesn't matter for us,
+        //      but intuitively the promise result should probably be resolved first.
+      };
+      if (useTimeout) {
+        timeout = setTimeout(() => {
+          reject(new Error('Timeout exceeded while awaiting result from worker running native-addon module functions'));
+          rejected = true;
+        }, timeoutMs);
+      }
+      this._postAsyncWork(method, args, onResult);
+    });
+  }
+
+  /**
+   *
+   * @param {boolean} sender
+   * @returns SRTSOCKET identifier (integer value) or -1 (SRT_ERROR)
+   */
+  createSocket(sender, callback) {
+    return this._createAsyncWorkPromise("createSocket", [sender], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   * @param address
+   * @param port
+   */
+  bind(socket, address, port, callback) {
+    return this._createAsyncWorkPromise("bind", [socket, address, port], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   * @param backlog
+   */
+  listen(socket, backlog, callback) {
+    return this._createAsyncWorkPromise("listen", [socket, backlog], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   * @param host
+   * @param port
+   */
+  connect(socket, host, port, callback) {
+    return this._createAsyncWorkPromise("connect", [socket, host, port], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   * @returns File descriptor of incoming connection pipe
+   */
+  accept(socket, callback, useTimeout = false, timeoutMs = AsyncSRT.TimeoutMs) {
+    return this._createAsyncWorkPromise("accept", [socket], callback, useTimeout, timeoutMs);
+  }
+
+  /**
+   *
+   * @param socket
+   */
+  close(socket, callback) {
+    return this._createAsyncWorkPromise("close", [socket], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   * @param chunkSize
+   * @returns {Promise<Buffer>}
+   */
+  read(socket, chunkSize, callback) {
+    return this._createAsyncWorkPromise("read", [socket, chunkSize], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   * @param {Buffer} chunk
+   */
+  write(socket, chunk, callback) {
+    // TODO: see if we can do this using SharedArrayBuffer for example,
+    //       or just leveraging Transferable objects capabilities ... ?
+    // FIXME: Performance ... ?
+    const buf = Buffer.allocUnsafe(chunk.length);
+    chunk.copy(buf);
+    chunk = buf;
+    return this._createAsyncWorkPromise("write", [socket, chunk], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   * @param option
+   * @param value
+   */
+  setSockOpt(socket, option, value, callback) {
+    return this._createAsyncWorkPromise("setSockOpt", [socket, option, value], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   * @param option
+   */
+  getSockOpt(socket, option, callback) {
+    return this._createAsyncWorkPromise("getSockOpt", [socket, option], callback);
+  }
+
+  /**
+   *
+   * @param socket
+   */
+  getSockState(socket, callback) {
+    return this._createAsyncWorkPromise("getSockState", [socket], callback);
+  }
+
+  /**
+   * @returns epid
+   */
+  epollCreate(callback) {
+    return this._createAsyncWorkPromise("epollCreate", [], callback);
+  }
+
+  /**
+   *
+   * @param epid
+   * @param socket
+   * @param events
+   */
+  epollAddUsock(epid, socket, events, callback) {
+    return this._createAsyncWorkPromise("epollAddUsock", [epid, socket, events], callback);
+  }
+
+  /**
+   *
+   * @param epid
+   * @param msTimeOut
+   */
+  epollUWait(epid, msTimeOut, callback) {
+    return this._createAsyncWorkPromise("epollUWait", [epid, msTimeOut], callback);
+  }
+}
+
+module.exports = {AsyncSRT};
+
+
+
+
+
+

types/srt-api-async.d.ts

@@ -0,0 +1,107 @@
+declare module "srt" {
+
+  type AsyncSRTCallback<T> = (result: T) => void;
+
+  class AsyncSRT {
+
+    static TimeoutMs: number;
+
+    /**
+     *
+     * @param sender
+     * @returns SRTSOCKET identifier (integer value)
+     */
+    createSocket(sender: boolean, callback?: AsyncSRTCallback<number>): Promise<number>
+
+    /**
+     *
+     * @param socket
+     * @param address
+     * @param port
+     */
+    bind(socket: number, address: string, port: number, callback?: AsyncSRTCallback<SRTResult>): Promise<SRTResult>
+
+    /**
+     *
+     * @param socket
+     * @param backlog
+     */
+    listen(socket: number, backlog: number, callback?: AsyncSRTCallback<SRTResult>): Promise<SRTResult>
+
+    /**
+     *
+     * @param socket
+     * @param host
+     * @param port
+     */
+    connect(socket: number, host: string, port: number, callback?: AsyncSRTCallback<SRTResult>): Promise<SRTResult>
+
+    /**
+     *
+     * @param socket
+     * @returns File descriptor of incoming connection pipe
+     */
+    accept(socket: number, callback?: AsyncSRTCallback<SRTFileDescriptor>): Promise<SRTFileDescriptor>
+
+    /**
+     *
+     * @param socket
+     */
+    close(socket: number, callback?: AsyncSRTCallback<SRTResult>): Promise<SRTResult>
+
+    /**
+     *
+     * @param socket
+     * @param chunkSize
+     */
+    read(socket: number, chunkSize: number, callback?: AsyncSRTCallback<SRTReadReturn>): Promise<SRTReadReturn>
+
+    /**
+     *
+     * @param socket
+     * @param chunk
+     */
+    write(socket: number, chunk: Buffer, callback?: AsyncSRTCallback<SRTResult>): Promise<SRTResult>
+
+    /**
+     *
+     * @param socket
+     * @param option
+     * @param value
+     */
+    setSockOpt(socket: number, option: SRTSockOpt, value: SRTSockOptValue, callback?: AsyncSRTCallback<SRTResult>): Promise<SRTResult>
+
+    /**
+     *
+     * @param socket
+     * @param option
+     */
+    getSockOpt(socket: number, option: SRTSockOpt, callback?: AsyncSRTCallback<SRTSockOptValue>): Promise<SRTSockOptValue>
+
+    /**
+     *
+     * @param socket
+     */
+    getSockState(socket: number, callback?: AsyncSRTCallback<SRTSockStatus>): Promise<SRTSockStatus>
+
+    /**
+     * @returns epid
+     */
+    epollCreate(callback?: AsyncSRTCallback<number>): Promise<number>
+
+    /**
+     *
+     * @param epid
+     * @param socket
+     * @param events
+     */
+    epollAddUsock(epid: number, socket: number, events: number, callback?: AsyncSRTCallback<SRTResult>): Promise<SRTResult>
+
+    /**
+     *
+     * @param epid
+     * @param msTimeOut
+     */
+    epollUWait(epid: number, msTimeOut: number, callback?: AsyncSRTCallback<SRTEpollEvent[]>): Promise<SRTEpollEvent[]>
+  }
+}

types/srt-api.d.ts

@@ -48,7 +48,7 @@ declare module "srt" {
      * @param socket
      * @param chunkSize
      */
-    read(socket: number, chunkSize: number): Buffer
+    read(socket: number, chunkSize: number): SRTReadReturn
 
     /**
      *
@@ -63,14 +63,14 @@ declare module "srt" {
      * @param option
      * @param value
      */
-    setSockOpt(socket: number, option: number, value: SRTSocketOptValue): number
+    setSockOpt(socket: number, option: SRTSockOpt, value: SRTSockOptValue): SRTResult
 
     /**
      *
      * @param socket
      * @param option
      */
-    getSockOpt(socket: number, option: number): SRTSocketOptValue
+    getSockOpt(socket: number, option: SRTSockOpt): SRTSockOptValue
 
     /**
      *
@@ -96,17 +96,19 @@ declare module "srt" {
      * @param epid
      * @param msTimeOut
      */
-    epollUWait(epid: number, msTimeOut: number): SRTPollingEvent[]
+    epollUWait(epid: number, msTimeOut: number): SRTEpollEvent[]
   }
 
-  interface SRTPollingEvent {
+  interface SRTEpollEvent {
     socket: SRTFileDescriptor
     events: number
   }
 
+  type SRTReadReturn = Buffer | SRTResult.SRT_ERROR | null
+
   type SRTFileDescriptor = number;
 
-  type SRTSocketOptValue = boolean | number | string
+  type SRTSockOptValue = boolean | number | string
 
   enum SRTResult {
     SRT_ERROR = -1,