Some bugs, updates, start adding tests

Creates basic xmlrpc tests for subs, pubs, services.
Keep trying to make xmlrpc calls if connection refused.

Author: chris-smith

lib/MasterApiClient.js

@@ -20,6 +20,7 @@
 let xmlrpc = require('xmlrpc');
 let networkUtils = require('../utils/network_utils.js');
 let logger = require('../utils/logger.js');
+let xmlrpcUtils = require('../utils/xmlrpc_utils.js');
 
 //-----------------------------------------------------------------------
 
@@ -35,6 +36,10 @@ class MasterApiClient {
 		return this._xmlrpcClient;
 	}
 
+	_call(method, data, resolve, reject) {
+		xmlrpcUtils.call(this.getXmlrpcClient(), method, data, resolve, reject, this._log);
+	}
+
 	registerService(callerId, service, serviceUri, uri) {
 		let data = [
 			callerId,
@@ -44,14 +49,7 @@ class MasterApiClient {
 		];
 
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('registerService', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('registerService', data, resolve, reject);
 		});
 	}
 
@@ -63,14 +61,7 @@ class MasterApiClient {
 		];
 
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('unregisterService', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('unregisterService', data, resolve, reject);
 		});
 	}
 
@@ -82,14 +73,7 @@ class MasterApiClient {
 			uri
 		];
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('registerSubscriber', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('registerSubscriber', data, resolve, reject);
 		});
 	}
 
@@ -100,14 +84,7 @@ class MasterApiClient {
 			uri
 		];
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('unregisterSubscriber', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('unregisterSubscriber', data, resolve, reject);
 		});
 	}
 
@@ -119,14 +96,7 @@ class MasterApiClient {
 			uri
 		];
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('registerPublisher', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('registerPublisher', data, resolve, reject);
 		});
 	}
 
@@ -137,28 +107,14 @@ class MasterApiClient {
 			uri
 		];
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('unregisterPublisher', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('unregisterPublisher', data, resolve, reject);
 		});
 	}
 
 	lookupNode(callerId, nodeName) {
 		let data = [callerId, nodeName];
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('lookupNode', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('lookupNode', data, resolve, reject);
 		});
 	}
 
@@ -177,31 +133,16 @@ class MasterApiClient {
 	getUri(callerId) {
 		let data = [callerId];
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('getUri', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('getUri', data, resolve, reject);
 		});
 	}
 
 	lookupService(callerId, service) {
 		let data = [callerId, service];
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('lookupService', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+			this._call('lookupService', data, resolve, reject);
 		});
 	}
-
 };
 
 //-----------------------------------------------------------------------

lib/ParamServerApiClient.js

@@ -16,29 +16,29 @@
  */
 
 'use strict';
+const logger = require('../utils/logger.js');
+const xmlrpcUtils = require('../utils/xmlrpc_utils.js');
 
 //-----------------------------------------------------------------------
 
 class ParamServerApiClient {
   constructor(xmlrpcClient) {
+    this._log = logger.createLogger({name: 'ParamClient'});
     this._xmlrpcClient = xmlrpcClient;
   }
 
+  _call(method, data, resolve, reject) {
+		xmlrpcUtils.call(this._xmlrpcClient, method, data, resolve, reject, this._log);
+	}
+
   deleteParam(callerId, key) {
     let data = [
 			callerId,
 			key
 		];
 
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('deleteParam', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+      this._call('deleteParam', data, resolve, reject);
 		});
   }
 
@@ -50,14 +50,7 @@ class ParamServerApiClient {
 		];
 
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('setParam', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-					resolve(resp);
-				}
-			});
+      this._call('setParam', data, resolve, reject);
 		});
   }
 
@@ -68,15 +61,10 @@ class ParamServerApiClient {
 		];
 
 		return new Promise((resolve, reject) => {
-			this._xmlrpcClient.methodCall('getParam', data, (err, resp) => {
-				if (err || resp[0] !== 1) {
-					reject(err, resp);
-				}
-				else {
-          // resp[2] is the actual parameter value, and presumably all anyone cares about
-					resolve(resp[2]);
-				}
-			});
+      this._call('getParam', data, (resp) => {
+        // resp[2] is the actual parameter value, and presumably all anyone cares about
+        resolve(resp[2]);
+      }, reject);
 		});
   }
 

lib/Publisher.js

@@ -102,6 +102,10 @@ class Publisher extends EventEmitter {
 		return this._latching;
 	}
 
+	getNumSubscribers() {
+		return Object.keys(this._subClients).length;
+	}
+
 	disconnect() {
 		Object.keys(this._subClients).forEach((clientId) => {
 			const client = this._subClients[clientId];
@@ -232,6 +236,14 @@ class Publisher extends EventEmitter {
 						this._log.debug('Publisher ' + this.getTopic() + ' client ' + subscriber.name + ' disconnected!');
 						delete this._subClients[subscriber.name];
 					});
+
+					subscriber.on('end', () => {
+						this._log.debug('Sub %s sent END', subscriber.name);
+					});
+
+					subscriber.on('error', () => {
+						this._log.debug('Sub %s had error', subscriber.name);
+					});
 				}).listen(port);
 
 				// it's possible the port was taken before we could use it
@@ -294,7 +306,7 @@ class Publisher extends EventEmitter {
 			if (code === 1) {
 				// registration worked
 				this._ready = true;
-				this.emit('ready');
+				this.emit('registered');
 			}
 		})
 		.catch((err, resp) => {

lib/RosNode.js

@@ -126,13 +126,13 @@ class RosNode extends EventEmitter {
 		}
 	}
 
-	unadvertiseService(serviceNam) {
+	unadvertiseService(service) {
 		const server = this._services[service];
 		if (server) {
 			// FIXME: we'll definitely need this when we support
 			// persistent service connections.
 			// service.disconnect();
-			this.unregisterService(service);
+			this.unregisterService(service, server.getServiceUri());
 			delete this._services[service];
 		}
 	}
@@ -253,6 +253,7 @@ class RosNode extends EventEmitter {
 					this._slaveApiServer.on('getBusInfo', this._handleGetBusInfo.bind(this));
 					this._slaveApiServer.on('getBusStats', this._handleGetBusStats.bind(this));
           callback(port);
+					this.emit('slaveApiSetupComplete', port);
         });
 
         server.on('error', (err) => {
@@ -310,6 +311,7 @@ class RosNode extends EventEmitter {
 		let topic = params[1];
 		let sub = this._subscribers[topic];
 		if (sub) {
+			this._log.debug('got sub for topic ' + topic);
 			let pubs = params[2];
 			sub.requestTopicFromPubs(params[2]);
 			let resp = [
@@ -320,6 +322,7 @@ class RosNode extends EventEmitter {
 			callback(null, resp);
 		}
 		else {
+			this._log.info('Don\'t have topic ' + topic);
 			let resp = [
 				0,
 				'Don\'t have topic ' + topic,

lib/ServiceServer.js

@@ -120,7 +120,7 @@ class ServiceServer extends EventEmitter {
       return this._nodeHandle.registerService(this.getService(), NetworkUtils.formatServiceUri(port));
     })
     .then((resp) => {
-      this._log.debug('register service resp! ' + resp);
+      this.emit('registered');
     });
   }
 
@@ -156,6 +156,7 @@ class ServiceServer extends EventEmitter {
 
         server.on('listening', () => {
           this._log.debug('Listening on port ' + port);
+		  this._port = port;
           callback(port);
         });
       })

lib/Subscriber.js

@@ -91,6 +91,10 @@ class Subscriber extends EventEmitter {
 		return this._type;
 	}
 
+	getNumPublishers() {
+		return Object.keys(this._pubClients).length;
+	}
+
 	/**
 	 * Send a topic request to each of the publishers we haven't connected to yet
 	 * @param pubs {Array} array of uris of nodes that are publishing this topic
@@ -133,15 +137,18 @@ class Subscriber extends EventEmitter {
 	_register() {
 		this._nodeHandle.registerSubscriber(this._topic, this._type)
 		.then((resp) => {
-			// handle response from register subscriber call
-			let code = resp[0];
-			let msg = resp[1];
-			let pubs = resp[2];
-			if ( code === 1 && pubs.length > 0) {
-				// this means we're ok and that publishers already exist on this topic
-				// we should connect to them
-				this.requestTopicFromPubs(pubs);
-			}
+		  // handle response from register subscriber call
+		  let code = resp[0];
+		  let msg = resp[1];
+		  let pubs = resp[2];
+		  if ( code === 1 ) {
+            if (pubs.length > 0) {
+  			  // this means we're ok and that publishers already exist on this topic
+  			  // we should connect to them
+  			  this.requestTopicFromPubs(pubs);
+            }
+            this.emit('registered');
+		  }
 		})
 		.catch((err, resp) => {
 			this._log.warn('Error during subscriber ' + this.getTopic() + ' registration.');
@@ -182,11 +189,19 @@ class Subscriber extends EventEmitter {
 			client.$initialized = true;
 
 			client.on('close', () => {
+				this._log.debug('Pub %s closed', client.name);
 				this._log.debug('Subscriber ' + this.getTopic() + ' client ' + client.name + ' disconnected!');
 				delete this._nodeMap[client.nodeUri];
 				delete this._pubClients[client.name];
 				this._log.debug(JSON.stringify(this._pubClients));
 			});
+
+			client.on('end', () => {
+				this._log.debug('Pub %s sent END', client.name);
+			});
+			client.on('error', () => {
+				this._log.debug('Pub sent END');
+			});
 		}
 		else {
 			// deserialize message