diff --git a/.eslintignore b/.eslintignore new file mode 100644 index 0000000..29b8616 --- /dev/null +++ b/.eslintignore @@ -0,0 +1,4 @@ +build +deps +node_modules +tsc-lib diff --git a/.eslintrc.js b/.eslintrc.js index 76da8ea..02825dd 100644 --- a/.eslintrc.js +++ b/.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 + } }; diff --git a/examples/async-srt-await.js b/examples/async-srt-await.js new file mode 100644 index 0000000..4c381f3 --- /dev/null +++ b/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); diff --git a/examples/async-srt-promises.js b/examples/async-srt-promises.js new file mode 100644 index 0000000..62c820f --- /dev/null +++ b/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); + } + }); + + diff --git a/examples/async-srt.js b/examples/async-srt.js new file mode 100644 index 0000000..9f794e7 --- /dev/null +++ b/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); + } + }); + } + }); +}); + diff --git a/index.d.ts b/index.d.ts index 6eccc32..d7a0064 100644 --- a/index.d.ts +++ b/index.d.ts @@ -1,5 +1,7 @@ +/* eslint-disable @typescript-eslint/triple-slash-reference */ /// /// /// +/// export * from "srt"; diff --git a/index.js b/index.js index d364e9c..167c971 100644 --- a/index.js +++ b/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 -} \ No newline at end of file + SRTWriteStream, + AsyncSRT, +}; diff --git a/package.json b/package.json index 4c81249..d680859 100644 --- a/package.json +++ b/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" }, diff --git a/src/async-worker.js b/src/async-worker.js new file mode 100644 index 0000000..507fd97 --- /dev/null +++ b/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 + }); + }); +})(); + + + + diff --git a/src/async.js b/src/async.js new file mode 100644 index 0000000..404ce80 --- /dev/null +++ b/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} 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} 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} + */ + 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}; + + + + + + diff --git a/types/srt-api-async.d.ts b/types/srt-api-async.d.ts new file mode 100644 index 0000000..1a02341 --- /dev/null +++ b/types/srt-api-async.d.ts @@ -0,0 +1,107 @@ +declare module "srt" { + + type AsyncSRTCallback = (result: T) => void; + + class AsyncSRT { + + static TimeoutMs: number; + + /** + * + * @param sender + * @returns SRTSOCKET identifier (integer value) + */ + createSocket(sender: boolean, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + * @param address + * @param port + */ + bind(socket: number, address: string, port: number, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + * @param backlog + */ + listen(socket: number, backlog: number, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + * @param host + * @param port + */ + connect(socket: number, host: string, port: number, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + * @returns File descriptor of incoming connection pipe + */ + accept(socket: number, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + */ + close(socket: number, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + * @param chunkSize + */ + read(socket: number, chunkSize: number, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + * @param chunk + */ + write(socket: number, chunk: Buffer, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + * @param option + * @param value + */ + setSockOpt(socket: number, option: SRTSockOpt, value: SRTSockOptValue, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + * @param option + */ + getSockOpt(socket: number, option: SRTSockOpt, callback?: AsyncSRTCallback): Promise + + /** + * + * @param socket + */ + getSockState(socket: number, callback?: AsyncSRTCallback): Promise + + /** + * @returns epid + */ + epollCreate(callback?: AsyncSRTCallback): Promise + + /** + * + * @param epid + * @param socket + * @param events + */ + epollAddUsock(epid: number, socket: number, events: number, callback?: AsyncSRTCallback): Promise + + /** + * + * @param epid + * @param msTimeOut + */ + epollUWait(epid: number, msTimeOut: number, callback?: AsyncSRTCallback): Promise + } +} diff --git a/types/srt-api.d.ts b/types/srt-api.d.ts index 0ee237a..84c36cd 100644 --- a/types/srt-api.d.ts +++ b/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,