Merge pull request #12 from chfritz/Issue#6_generate_message_handlers_automatically

Issue#6 generate message handlers automatically

Author: chris-smith

example2.js

@@ -0,0 +1,87 @@
+'use strict';
+
+let rosnodejs = require('./index.js');
+// const std_msgs = rosnodejs.require('std_msgs').msg;
+// const SetBool = rosnodejs.require('std_srvs').srv.SetBool;
+
+rosnodejs.use(['std_msgs/String'], 
+              ['std_srvs/SetBool'], function() {
+
+  const msg = new (rosnodejs.message('std_msgs/String'))(
+    { data: "howdy" });
+
+  rosnodejs.initNode('/my_node')
+    .then((rosNode) => {
+
+      // EXP 1) Service Server
+      let service = rosNode.advertiseService({
+        service: '/set_bool',
+        type: 'std_srvs/SetBool'
+      }, (req, resp) => {
+        console.log('Handling request! ' + JSON.stringify(req));
+        resp.success = !req.data;
+        resp.message = 'Inverted!';
+        return true;
+      });
+
+      // EXP 2) Service Client
+      setTimeout(function() {
+        let serviceClient = rosNode.serviceClient({
+          service: '/set_bool',
+          type: 'std_srvs/SetBool'
+        });
+        rosNode.waitForService(serviceClient.getService(), 2000)
+          .then((available) => {
+            if (available) {
+              const request = 
+                new (rosnodejs.serviceRequest('std_srvs/SetBool'))({ 
+                  data: false });
+              serviceClient.call(request, (resp) => {
+                console.log('Service response ' + JSON.stringify(resp));
+              });
+            } else {
+              console.log('Service not available');     
+            }
+          });
+      }, 1000); // wait a second before calling our service
+
+      // EXP 3) Params
+      rosNode.setParam('~junk', {'hi': 2}).then(() => {
+        rosNode.getParam('~junk').then((val) => { 
+          console.log('Got Param!!! ' + JSON.stringify(val)); 
+        });
+      });
+
+      // // EXP 4) Publisher
+      let pub = rosNode.advertise({
+        topic: '/my_topic',
+        type: 'std_msgs/String',
+        queueSize: 1,
+        latching: true,
+        throttleMs: 9
+      });
+
+      let msgStart = 'my message ';
+      let iter = 0;
+      setInterval(() => {
+        msg.data = msgStart + iter
+        pub.publish(msg);
+        ++iter;
+        if (iter > 200) {
+          iter = 0;
+        }
+      }, 5);
+
+      // EXP 5) Subscriber
+      let sub = rosNode.subscribe({
+        topic: '/my_topic',
+        type: 'std_msgs/String',
+        queueSize: 1,
+        throttleMs: 1000},
+        (data) => {
+          console.log('SUB DATA ', data, data.data);
+        });
+    });
+  
+});
+

index.js

@@ -21,6 +21,7 @@
 
 const netUtils = require('./utils/network_utils.js');
 const msgUtils = require('./utils/message_utils.js');
+const messages = require('./utils/messages.js');
 msgUtils.findMessageFiles();
 
 // these will be modules, they depend on logger which isn't initialized yet
@@ -136,6 +137,58 @@ let Rosnodejs = {
     return pack;
   },
 
+  /** create message classes and services classes for all the given
+   * types before calling callback */
+  use(messages, services, callback) {
+    const self = this;
+    this._useMessages(messages, function() {
+      self._useServices(services, callback);
+    });
+  },
+
+  /** create message classes for all the given types */
+  _useMessages(types, callback) {
+    var Messages = [];
+    types.forEach(function(type) { 
+      messages.getMessage(type, function(error, Message) {
+        Messages.push(Message);
+        if (Messages.length === types.length) {
+          callback();
+        }
+      });
+    });
+  },
+
+  /** create message classes for all the given types */
+  _useServices(types, callback) {
+    var count = types.length;
+    types.forEach(function(type) { 
+      messages.getServiceRequest(type, function() {
+        messages.getServiceResponse(type, function() {
+          if (--count == 0) {
+            callback();
+          }
+        });
+      });
+    });
+  },
+
+  /** get message definition class from registry. Do not generate it
+   * from .msg file if it doesn't already exist. This is mostly
+   * because that would need to be async right now and we want a sync
+   * method. */
+  message(type) {
+    return messages.getFromRegistry(type, "message");
+  },
+
+  serviceRequest(type) {
+    return messages.getFromRegistry(type, "request");
+  },
+
+  serviceResponse(type) {
+    return messages.getFromRegistry(type, "response");
+  },
+
   /**
    * @return {NodeHandle} for initialized node
    */

lib/Publisher.js

@@ -8,11 +8,11 @@
  *    You may obtain a copy of the License at
  *    http://www.apache.org/licenses/LICENSE-2.0
  *
- *    Unless required by applicable law or agreed to in writing, software
- *    distributed under the License is distributed on an "AS IS" BASIS,
- *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *    See the License for the specific language governing permissions and
- *    limitations under the License.
+ *    Unless required by applicable law or agreed to in writing,
+ *    software distributed under the License is distributed on an "AS
+ *    IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *    express or implied. See the License for the specific language
+ *    governing permissions and limitations under the License.
  */
 
 "use strict";
@@ -48,9 +48,11 @@ class Publisher extends EventEmitter {
     }
 
     /**
-     * throttleMs interacts with queueSize to determine when to send messages.
+     * throttleMs interacts with queueSize to determine when to send
+     * messages.
      *  < 0  : send immediately - no interaction with queue
-     *  >= 0 : place event at end of event queue to publish message after minimum delay (MS)
+     * >= 0 : place event at end of event queue to publish message
+         after minimum delay (MS)
      */
     if (options.hasOwnProperty('throttleMs')) {
       this._throttleMs = options.throttleMs;
@@ -118,7 +120,8 @@ class Publisher extends EventEmitter {
   }
 
   /**
-   * Schedule the msg for publishing - or publish immediately if we're supposed to
+   * Schedule the msg for publishing - or publish immediately if we're
+   * supposed to
    * @param msg {object} object type matching this._type
    * @param [throttleMs] {number} optional override for publisher setting
    */
@@ -149,7 +152,8 @@ class Publisher extends EventEmitter {
       if (this._pubTimeout === null) {
         let now = Date.now();
         if (this._pubTime !== null) {
-          // check how long to throttle for based on the last time we published
+          // check how long to throttle for based on the last time we
+          // published
           if (now - this._pubTime > throttleMs) {
             throttleMs = 0;
           }
@@ -180,11 +184,16 @@ class Publisher extends EventEmitter {
       try {
         let bufferInfo = {buffer: [], length: 0};
         // serialize pushes buffers onto buffInfo.buffer in order
-        // concat them, and preprend the byte length to the message before sending
+        // concat them, and preprend the byte length to the message
+        // before sending
+        // console.log("_publish", this._messageHandler, msg);
         bufferInfo = this._messageHandler.serialize(msg, bufferInfo);
+        // bufferInfo = msg.serialize();
+        // console.log("_publish", bufferInfo);
 
         // prepend byte length to message
-        let serialized = Serialize(Buffer.concat(bufferInfo.buffer, bufferInfo.length));
+        let serialized = Serialize(
+          Buffer.concat(bufferInfo.buffer, bufferInfo.length));
         Object.keys(this._subClients).forEach((client) => {
           this._subClients[client].write(serialized);
         });
@@ -197,7 +206,7 @@ class Publisher extends EventEmitter {
         }
       }
       catch (err) {
-        this._log.warn('Error when publishing message ' + err);
+        this._log.warn('Error when publishing message ', err.stack);
       }
     });
 
@@ -212,13 +221,16 @@ class Publisher extends EventEmitter {
       NetworkUtils.getFreePort()
       .then((port) => {
         let server = net.createServer((subscriber) => {
-          let subName = subscriber.remoteAddress + ":" + subscriber.remotePort;
+          let subName = subscriber.remoteAddress + ":" 
+            + subscriber.remotePort;
           subscriber.name = subName;
-          this._log.debug('Publisher ' + this.getTopic() + ' got connection from ' + subName);
+          this._log.debug('Publisher ' + this.getTopic() 
+                          + ' got connection from ' + subName);
 
-          // subscriber will send us tcpros handshake before we can start publishing
-          // to it.
-          subscriber.$handshake = this._handleHandshake.bind(this, subscriber);
+          // subscriber will send us tcpros handshake before we can
+          // start publishing to it.
+          subscriber.$handshake = 
+            this._handleHandshake.bind(this, subscriber);
 
           // handshake will be TCPROS encoded, so use a DeserializeStream to
           // handle any chunking
@@ -233,7 +245,8 @@ class Publisher extends EventEmitter {
           }
 
           subscriber.on('close', () => {
-            this._log.info('Publisher ' + this.getTopic() + ' client ' + subscriber.name + ' disconnected!');
+            this._log.info('Publisher ' + this.getTopic() + ' client ' 
+                           + subscriber.name + ' disconnected!');
             delete this._subClients[subscriber.name];
           });
 
@@ -271,15 +284,24 @@ class Publisher extends EventEmitter {
   _handleHandshake(subscriber, data) {
     if (!subscriber.$initialized) {
       let header = TcprosUtils.parseSubHeader(data);
-      let valid = TcprosUtils.validateSubHeader(header, this.getTopic(), this.getType(), this._messageHandler.md5sum());
+      let valid = TcprosUtils.validateSubHeader(
+        header, this.getTopic(), this.getType(), 
+        this._messageHandler.md5sum());
       if (valid !== null) {
-        this._log.error('Unable to validate connection header ' + JSON.stringify(header));
+        this._log.error('Unable to validate connection header ' 
+                        + JSON.stringify(header));
         subscriber.write(Serialize(valid));
         return;
       }
-      this._log.debug('Pub ' + this.getTopic() + ' got connection header ' + JSON.stringify(header));
-
-      let respHeader = TcprosUtils.createPubHeader(this._nodeHandle.getNodeName(), this._messageHandler.md5sum(), this.getType(), this.getLatching());
+      this._log.debug('Pub ' + this.getTopic() 
+                      + ' got connection header ' + JSON.stringify(header));
+
+      let respHeader = 
+        TcprosUtils.createPubHeader(
+          this._nodeHandle.getNodeName(), 
+          this._messageHandler.md5sum(), 
+          this.getType(), 
+          this.getLatching());
       subscriber.write(respHeader);
 
       if (this._lastSentMsg !== null) {
@@ -293,7 +315,8 @@ class Publisher extends EventEmitter {
       this.emit('connection', subscriber.name);
     }
     else {
-      this._log.error('Got message from subscriber after handshake - what gives!!');
+      this._log.error(
+        'Got message from subscriber after handshake - what gives!!');
     }
   }
 
@@ -310,7 +333,8 @@ class Publisher extends EventEmitter {
       }
     })
     .catch((err, resp) => {
-      this._log.error('reg pub err ' + err + ' resp: ' + JSON.stringify(resp));
+      this._log.error('reg pub err ' + err + ' resp: ' 
+                      + JSON.stringify(resp));
     })
   }
 

lib/ServiceClient.js

@@ -77,7 +77,10 @@ class ServiceClient extends EventEmitter {
     .then((resp) => {
       let serviceUri = resp[2];
       // connect to the service
-      return this._connectToService(NetworkUtils.getAddressAndPortFromUri(serviceUri), request);
+      return this._connectToService(
+        NetworkUtils.getAddressAndPortFromUri(serviceUri), 
+        request
+      );
     })
     .then((msg) => {
       responseCallback(msg);

lib/ServiceServer.js

@@ -39,7 +39,9 @@ class ServiceServer extends EventEmitter {
 
     this._nodeHandle = nodeHandle;
 
-    this._log = log.createLogger({name: 'srvServer' + this.getService()});
+    this._log = log.createLogger({
+      name: 'srvServer' + this.getService()
+    });
 
     this._requestCallback = callback;
 

package.json

@@ -21,6 +21,9 @@
   "dependencies": {
     "moment": "^2.12.0",
     "portscanner": "^1.0.0",
-    "xmlrpc": "^1.3.1"
+    "xmlrpc": "^1.3.1",
+    "walker"        : "1.0.7",
+    "md5"           : "2.1.0",
+    "async"         : "0.1.22"
   }
 }