diff --git a/amqp-rpc/amqp-rpc-tests.ts b/amqp-rpc/amqp-rpc-tests.ts index 30d740d3e..71df096a8 100644 --- a/amqp-rpc/amqp-rpc-tests.ts +++ b/amqp-rpc/amqp-rpc-tests.ts @@ -39,4 +39,42 @@ rpc.call('withoutCB', {}, function (msg) { console.log('withoutCB results:', msg); //output: please run function without cb parameter }); -rpc.call('withoutCB', {}); //output message on server side console \ No newline at end of file +rpc.call('withoutCB', {}); //output message on server side console + +import os = require('os'); +interface State { + type: string; +} + +var counter = 0; +rpc.onBroadcast('getWorkerStat', function (params, cb) { + if (params && params.type == 'fullStat') { + cb(null, { + pid: process.pid, + hostname: os.hostname(), + uptime: process.uptime(), + counter: counter++ + }); + } + else { + cb(null, { counter: counter++ }) + } +}); + +var all_stats: any = {}; +rpc.callBroadcast( + 'getWorkerStat', + { type: 'fullStat' }, //request parameters + { //call options + ttl: 1000, //wait response time (1 seconds), after run onComplete + onResponse: function (err: any, stat: any) { //callback on each worker response + all_stats[stat.hostname + ':' + stat.pid] = stat; + }, + onComplete: function () { //callback on ttl expired + console.log('----------------------- WORKER STATISTICS ----------------------------------------'); + for (var worker in all_stats) { + var s: any = all_stats[worker]; + console.log(worker, '\tuptime=', s.uptime.toFixed(2) + ' seconds', '\tcounter=', s.counter); + } + } + }); \ No newline at end of file diff --git a/amqp-rpc/amqp-rpc.d.ts b/amqp-rpc/amqp-rpc.d.ts index a848fd4c4..a7334460f 100644 --- a/amqp-rpc/amqp-rpc.d.ts +++ b/amqp-rpc/amqp-rpc.d.ts @@ -35,7 +35,7 @@ declare module "amqp-rpc" { } export interface BroadcastOptions { - ttl?: boolean; + ttl?: number; onResponse?: any; context?: any; onComplete?: any; @@ -52,6 +52,10 @@ declare module "amqp-rpc" { (...args: any[]): void; } + export interface CallbackWithError { + (err: any, ...args: any[]): void; + } + export function factory(opt?: Options): amqpRPC; export class amqpRPC { @@ -61,8 +65,8 @@ declare module "amqp-rpc" { call(cmd: string, params: T, cb?: Callback, context?: any, options?: CallOptions): string; on(cmd: string, cb: (param?: T, cb?: Callback, info?: CommandInfo) => void, context?: any, options?: HandlerOptions): boolean; off(cmd: string): boolean; - callBroadcast(cmd: string, params: any, options: BroadcastOptions): void; - onBroadcast(cmd: string, cb: (err: any) => void, context: any, options?: any): boolean; + callBroadcast(cmd: string, params: T, options?: BroadcastOptions): void; + onBroadcast(cmd: string, cb?: (params?: T, cb?: CallbackWithError) => void, context?: any, options?: any): boolean; offBroadcast(cmd: string): boolean; } -} \ No newline at end of file +}