add timeout parameter to client socket

This commit is contained in:
nitowa
2021-07-28 02:07:21 +02:00
parent 9f53abe0e3
commit bf96f9007f
8 changed files with 54 additions and 22 deletions
+4 -3
View File
@@ -5,6 +5,7 @@ import { PromiseIO } from "./PromiseIO/Server";
import * as T from './Types'; import * as T from './Types';
import * as U from './Utils'; import * as U from './Utils';
import * as I from './Interfaces'; import * as I from './Interfaces';
import { BAD_CONFIG_PARAM, UNKNOWN_RPC_IDENTIFIER, UNKNOWN_RPC_SERVER } from './Strings';
export class RPCServer< export class RPCServer<
InterfaceT extends T.RPCInterface = T.RPCInterface, InterfaceT extends T.RPCInterface = T.RPCInterface,
@@ -45,7 +46,7 @@ export class RPCServer<
if (conf.errorHandler) conf.errorHandler(socket, error, rpcName, args) if (conf.errorHandler) conf.errorHandler(socket, error, rpcName, args)
else { else {
if (forward) { if (forward) {
socket.call("$UNKNOWNRPC$", error) socket.call(UNKNOWN_RPC_IDENTIFIER, error)
} else { } else {
throw error throw error
} }
@@ -96,7 +97,7 @@ export class RPCServer<
this.attach(undefined, options) this.attach(undefined, options)
} else { } else {
if (options) if (options)
console.warn("RPCServer options were passed to listen(..) after attach(..) was called. Please pass them to attach(..) instead. Ignoring.") console.warn(BAD_CONFIG_PARAM)
} }
this.pio.listen(port) this.pio.listen(port)
return this return this
@@ -114,7 +115,7 @@ export class RPCServer<
clientSocket.on("*", (packet) => { clientSocket.on("*", (packet) => {
if (!infos.some(i => i.uniqueName === packet.data[0])) { if (!infos.some(i => i.uniqueName === packet.data[0])) {
if (packet.data[0].startsWith('destroy_')) return if (packet.data[0].startsWith('destroy_')) return
this.errorHandler(clientSocket, new Error(`Unknown RPC ${packet.data[0]}`), packet.data[0], [...packet.data].splice(1), true) this.errorHandler(clientSocket, new Error(UNKNOWN_RPC_SERVER(packet.data[0])), packet.data[0], [...packet.data].splice(1), true)
} }
}) })
} }
+15 -6
View File
@@ -4,6 +4,7 @@ import { PromiseIOClient, defaultClientConfig } from './PromiseIO/Client'
import * as T from './Types'; import * as T from './Types';
import * as I from './Interfaces'; import * as I from './Interfaces';
import { stripAfterEquals, appendComma } from './Utils'; import { stripAfterEquals, appendComma } from './Utils';
import { SOCKET_NOT_CONNECTED, UNKNOWN_RPC_IDENTIFIER, USER_DEFINED_TIMEOUT } from './Strings';
/** /**
@@ -33,7 +34,7 @@ export class RPCSocket<Ifc extends T.RPCInterface = T.RPCInterface> implements I
*/ */
constructor(public port: number, public address: string, private conf: T.ClientConfig = defaultClientConfig) { constructor(public port: number, public address: string, private conf: T.ClientConfig = defaultClientConfig) {
Object.defineProperty(this, 'socket', { value: undefined, writable: true }) Object.defineProperty(this, 'socket', { value: undefined, writable: true })
this.hook("$UNKNOWNRPC$", (err) => this.handlers['error'].forEach(handler => handler(err))) this.hook(UNKNOWN_RPC_IDENTIFIER, (err) => this.handlers['error'].forEach(handler => handler(err)))
} }
/** /**
@@ -114,11 +115,19 @@ export class RPCSocket<Ifc extends T.RPCInterface = T.RPCInterface> implements I
* @param args other arguments * @param args other arguments
*/ */
public async call(rpcname: string, ...args: any[]): Promise<any> { public async call(rpcname: string, ...args: any[]): Promise<any> {
if (!this.socket) throw new Error("The socket is not connected! Use socket.connect() first") if (!this.socket) throw new Error(SOCKET_NOT_CONNECTED)
try { try {
const val = await this.socket.call.apply(this.socket, [rpcname, ...args]) if(!this.conf.callTimeoutMs || this.conf.callTimeoutMs <= 0)
return val return await this.socket.call.apply(this.socket, [rpcname, ...args])
else
return await Promise.race([
this.socket.call.apply(this.socket, [rpcname, ...args]),
new Promise((_, rej) => {
setTimeout(_ => rej(USER_DEFINED_TIMEOUT(this.conf.callTimeoutMs)), this.conf.callTimeoutMs)
})
])
} catch (e) { } catch (e) {
this.emit('error', e) this.emit('error', e)
throw e throw e
@@ -131,7 +140,7 @@ export class RPCSocket<Ifc extends T.RPCInterface = T.RPCInterface> implements I
* @param args other arguments * @param args other arguments
*/ */
public async fire(rpcname: string, ...args: any[]): Promise<void> { public async fire(rpcname: string, ...args: any[]): Promise<void> {
if (!this.socket) throw new Error("The socket is not connected! Use socket.connect() first") if (!this.socket) throw new Error(SOCKET_NOT_CONNECTED)
await this.socket.fire.apply(this.socket, [rpcname, ...args]) await this.socket.fire.apply(this.socket, [rpcname, ...args])
} }
@@ -181,7 +190,7 @@ export class RPCSocket<Ifc extends T.RPCInterface = T.RPCInterface> implements I
* Get a list of available RPCs from the server * Get a list of available RPCs from the server
*/ */
public async info(sesame?: string) { public async info(sesame?: string) {
if (!this.socket) throw new Error("The socket is not connected! Use socket.connect() first") if (!this.socket) throw new Error(SOCKET_NOT_CONNECTED)
return await this.socket.call('info', sesame) return await this.socket.call('info', sesame)
} }
-1
View File
@@ -1,5 +1,4 @@
import * as T from "./Types"; import * as T from "./Types";
import * as I from "./Interfaces"
/** /**
* Interface for all classes that export RPCs * Interface for all classes that export RPCs
+1
View File
@@ -10,6 +10,7 @@ export const defaultClientConfig: ClientConfig = {
reconnectionDelay: 200, reconnectionDelay: 200,
timeout: 450, timeout: 450,
reconnection: false, reconnection: false,
callTimeoutMs: 0,
} }
export class PromiseIOClient { export class PromiseIOClient {
+14
View File
@@ -0,0 +1,14 @@
export const USER_DEFINED_TIMEOUT = ms => `User defined socket timout after ${ms}ms`
export const SOCKET_NOT_CONNECTED = "The socket is not connected! Use socket.connect() first"
export const UNKNOWN_RPC_IDENTIFIER = "$UNKNOWNRPC$"
export const UNKNOWN_RPC_SERVER = name => `Unknown RPC ${name}`
export const RPC_BAD_TYPE = type => `Bad socketIORPC type ${type}`
export const CALL_NOT_FOUND = callName => `Call not found: ${callName}. ; Zone: <root> ; Task: Promise.then ; Value: Error: Call not found: ${callName}`
export const BAD_CONFIG_PARAM = "RPCServer options were passed to listen(..) after attach(..) was called. Please pass them to attach(..) instead. Ignoring."
export const RPC_NO_NAME = name => `
RPC did not provide a name.
\nUse 'funtion name(..){ .. }' syntax instead.
\n
\n<------------OFFENDING RPC:
\n${name}
\n>------------OFFENDING RPC`
+2 -1
View File
@@ -21,7 +21,8 @@ export type FrontEndHandlerType = {
'close' : () => void 'close' : () => void
} }
export type ClientConfig = SocketIOClient.ConnectOpts & { export type ClientConfig = SocketIOClient.ConnectOpts & {
protocol?: 'http' | 'https' protocol?: 'http' | 'https',
callTimeoutMs?: number
} }
export type ExporterArray<InterfaceT extends RPCInterface = RPCInterface> = I.RPCExporter<RPCInterface<InterfaceT>, keyof InterfaceT>[] export type ExporterArray<InterfaceT extends RPCInterface = RPCInterface> = I.RPCExporter<RPCInterface<InterfaceT>, keyof InterfaceT>[]
+4 -9
View File
@@ -3,6 +3,7 @@ import * as uuidv4 from "uuid/v4"
import * as T from "./Types"; import * as T from "./Types";
import * as I from "./Interfaces"; import * as I from "./Interfaces";
import { Socket } from "socket.io" import { Socket } from "socket.io"
import { CALL_NOT_FOUND, RPC_BAD_TYPE, RPC_NO_NAME } from "./Strings";
/** /**
* Translate an RPC to RPCInfo for serialization. * Translate an RPC to RPCInfo for serialization.
@@ -36,13 +37,7 @@ export const rpcToRpcinfo = (socket: I.Socket, rpc: T.RPC<any, any>, owner: stri
} }
} }
case "function": case "function":
if (!rpc.name) throw new Error(` if (!rpc.name) throw new Error(RPC_NO_NAME(rpc.toString()))
RPC did not provide a name.
\nUse 'funtion name(..){ .. }' syntax instead.
\n
\n<------------OFFENDING RPC:
\n${rpc.toString()}
\n>------------OFFENDING RPC`)
return { return {
owner: owner, owner: owner,
argNames: extractArgs(rpc), argNames: extractArgs(rpc),
@@ -51,7 +46,7 @@ RPC did not provide a name.
call: sesame ? async ($__sesame__$, ...args) => { if (sesame($__sesame__$)) return await rpc.apply({}, args); throw makeError(rpc.name) } : rpc, // check & remove sesame call: sesame ? async ($__sesame__$, ...args) => { if (sesame($__sesame__$)) return await rpc.apply({}, args); throw makeError(rpc.name) } : rpc, // check & remove sesame
} }
} }
throw new Error("Bad socketIORPC type " + typeof rpc) throw new Error(RPC_BAD_TYPE(typeof rpc))
} }
/** /**
@@ -141,7 +136,7 @@ const hookGenerator = (rpc: T.HookRPC<any, any>, errorHandler: T.ErrorHandler, s
return eval(hookStr) return eval(hookStr)
} }
const makeError = (callName: string) => new Error(`Call not found: ${callName}. ; Zone: <root> ; Task: Promise.then ; Value: Error: Call not found: ${callName}`) const makeError = (callName: string) => new Error(CALL_NOT_FOUND(callName))
/** /**
* Extract a string list of parameters from a function * Extract a string list of parameters from a function
+13 -1
View File
@@ -9,8 +9,12 @@ import * as fetch from 'node-fetch';
import { PromiseIO } from "../src/PromiseIO/Server"; import { PromiseIO } from "../src/PromiseIO/Server";
import { PromiseIOClient } from "../src/PromiseIO/Client"; import { PromiseIOClient } from "../src/PromiseIO/Client";
import { assert, expect } from 'chai'; import { assert, expect } from 'chai';
import { USER_DEFINED_TIMEOUT } from "../src/Strings";
var should = require('chai').should(); var should = require('chai').should();
var chai = require("chai");
var chaiAsPromised = require("chai-as-promised");
chai.use(chaiAsPromised);
const noop = (...args) => { } const noop = (...args) => { }
const add = (...args: number[]) => { return args.reduce((a, b) => a + b, 0) } const add = (...args: number[]) => { return args.reduce((a, b) => a + b, 0) }
@@ -311,6 +315,7 @@ describe('can attach multiple RPCServers to same http server', () => {
], ],
} }
] ]
const callTimeout = 100;
let client: RPCSocket, client2: RPCSocket, server: RPCServer, server2: RPCServer let client: RPCSocket, client2: RPCSocket, server: RPCServer, server2: RPCServer
@@ -334,7 +339,7 @@ describe('can attach multiple RPCServers to same http server', () => {
new RPCSocket(8080, 'localhost').connect().then(sock => { new RPCSocket(8080, 'localhost').connect().then(sock => {
client = sock client = sock
new RPCSocket(8080, 'localhost', { path: "test" }).connect().then(sock2 => { new RPCSocket(8080, 'localhost', { path: "test", callTimeoutMs: callTimeout }).connect().then(sock2 => {
client2 = sock2 client2 = sock2
done() done()
}) })
@@ -356,6 +361,13 @@ describe('can attach multiple RPCServers to same http server', () => {
const res2 = await client2['Grp2'].test() const res2 = await client2['Grp2'].test()
expect(res2).to.equal('/test') expect(res2).to.equal('/test')
}) })
it('server1 should answer after server2 is closed', async () => {
server2.close()
const res = await client['HelloWorldRPCGroup'].echo("test")
expect(res).to.equal('test')
return client2['Grp2'].test().should.eventually.be.rejectedWith(USER_DEFINED_TIMEOUT(callTimeout))
})
}) })
describe("can attach second RPCServer if first is already running", () => { describe("can attach second RPCServer if first is already running", () => {