Support asynchronous services (#68)

-Service callbacks can now return promises. Result
 will be sent when promise resolves.
-Adds tests for asynchronous services.
-Service clients whose connections are closed while a call
 is in progress will be rejected.

Author: chris-smith

src/lib/ServiceClient.js

@@ -235,7 +235,14 @@ class ServiceClient extends EventEmitter {
     call.serviceClient.write(serializedRequest);
 
     return new Promise((resolve, reject) => {
+      const closeHandler = () => {
+        this._log.debug('Service %s client disconnected during call!', this.getService());
+        reject(new Error('Connection was closed'));
+      }
+
       call.serviceClient.$deserializeStream.once('message', (msg, success) => {
+        call.serviceClient.removeListener('close', closeHandler);
+
         if (success) {
           resolve(this._messageHandler.Response.deserialize(msg));
         }
@@ -245,6 +252,9 @@ class ServiceClient extends EventEmitter {
           reject(error);
         }
       });
+
+      // if the connection closes while waiting for a response, reject the request
+      call.serviceClient.on('close', closeHandler);
     });
   }
 

src/lib/ServiceServer.js

@@ -154,22 +154,29 @@ class ServiceServer extends EventEmitter {
 
     // call service callback
     let resp = new this._messageHandler.Response();
-    let success = this._requestCallback(req, resp);
+    let result = this._requestCallback(req, resp);
+    Promise.resolve(result)
+    .then((success) => {
+      // client should already have been closed, so if we got here just cut out early
+      if (this.isShutdown()) {
+        return;
+      }
 
-    const serializeResponse = TcprosUtils.serializeServiceResponse(
-      this._messageHandler.Response,
-      resp,
-      success
-    );
+      const serializeResponse = TcprosUtils.serializeServiceResponse(
+        this._messageHandler.Response,
+        resp,
+        success
+      );
 
-    // send service response
-    client.write(serializeResponse);
+      // send service response
+      client.write(serializeResponse);
 
-    if (!client.$persist) {
-      this._log.debug('Closing non-persistent client');
-      client.end();
-      delete this._clients[client.name];
-    }
+      if (!client.$persist) {
+        this._log.debug('Closing non-persistent client');
+        client.end();
+        delete this._clients[client.name];
+      }
+    });
   }
 
   _register() {

test/xmlrpcTest.js

@@ -812,6 +812,25 @@ describe('Protocol Test', () => {
       })
     });
 
+    it('Asynchronous Call and Response', (done) => {
+      const nh = rosnodejs.nh;
+      const serv = nh.advertiseService(service, srvType, (req, resp) => {
+        return Promise.resolve(true);
+      });
+
+      const client = nh.serviceClient(service, srvType);
+      nh.waitForService(service)
+      .then(() => {
+        return client.call({});
+      })
+      .then(() => {
+        done();
+      })
+      .catch((err) => {
+        throwNext(err);
+      })
+    });
+
     it('Service Failure', (done) => {
       const nh = rosnodejs.nh;
       const serv = nh.advertiseService(service, srvType, (req, resp) => {
@@ -881,6 +900,52 @@ describe('Protocol Test', () => {
       setTimeout(done, 500);
     });
 
+    it('Service Shutdown Handling Call', function(done) {
+      this.slow(1600);
+
+      const nh = rosnodejs.nh;
+      const serv = nh.advertiseService(service, srvType, (req, resp) => {
+        serv.shutdown();
+
+        return true;
+      });
+
+      const client = nh.serviceClient(service, srvType);
+      nh.waitForService(service)
+      .then(() => {
+        return client.call({});
+      })
+      .catch((err) => {
+        expect(err.message).to.equal('Connection was closed');
+        done();
+      });
+    });
+
+    it('Service Shutdown Handling Asynchronous Call', function(done) {
+      this.slow(1600);
+
+      const nh = rosnodejs.nh;
+      const serv = nh.advertiseService(service, srvType, (req, resp) => {
+
+        return new Promise((resolve) => {
+          setTimeout(() => {
+            serv.shutdown();
+            resolve(true);
+          }, 0);
+        });
+      });
+
+      const client = nh.serviceClient(service, srvType);
+      nh.waitForService(service)
+      .then(() => {
+        return client.call({});
+      })
+      .catch((err) => {
+        expect(err.message).to.equal('Connection was closed');
+        done();
+      });
+    });
+
     it('Service Unregistered During Call', (done) => {
       // simulate a service disconnecting between the lookupService call to ROS Master
       // and the connection to the service node's TCPROS server