Fixed format of 'callBroadcast' method

Author: wokim

amqp-rpc/amqp-rpc-tests.ts

@@ -39,4 +39,42 @@ rpc.call<any>('withoutCB', {}, function (msg) {
   console.log('withoutCB results:', msg);  //output: please run function without cb parameter
 });
 
-rpc.call<any>('withoutCB', {}); //output message on server side console
+rpc.call<any>('withoutCB', {}); //output message on server side console
+
+import os = require('os');
+interface State {
+  type: string;
+}
+
+var counter = 0;
+rpc.onBroadcast<State>('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<State>(
+  '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);
+      }
+    }
+  });

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<T>(cmd: string, params: T, cb?: Callback, context?: any, options?: CallOptions): string;
     on<T>(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<T>(cmd: string, params: T, options?: BroadcastOptions): void;
+    onBroadcast<T>(cmd: string, cb?: (params?: T, cb?: CallbackWithError) => void, context?: any, options?: any): boolean;
     offBroadcast(cmd: string): boolean;
   }
-}
+}