Merge branch 'multipleClients' of https://github.com/maxired/rosnodejs

Conflicts: lib/master.js lib/topic.js
baalexander · May 3, 2012 · 221d667 · 221d667
2 parents 7dee9b8 + af26a76
commit 221d667
Show file tree

Hide file tree

Showing 5 changed files with 178 additions and 160 deletions.
diff --git a/lib/master.js b/lib/master.js
@@ -5,14 +5,14 @@ var xmlrpc      = require('xmlrpc')
 var master = exports
 
 master.registerPublisher = function(publisher, callback) {
-  var masterUri   = environment.getRosMasterUri()
-    , callerUri   = publisher.callerUri
-    , callerId    = getGraphResourceName(publisher.callerId)
-    , topicName   = getGraphResourceName(publisher.topic)
-    , messageType = publisher.messageType
-    , params      = [callerId, topicName, messageType, callerUri]
-    , client      = xmlrpc.createClient(masterUri)
-    ;
+ var masterUri  = environment.getRosMasterUri()
+  , callerUri   = publisher.callerUri
+  , callerId    = getGraphResourceName(publisher.callerId)
+  , topicName   = getGraphResourceName(publisher.topic)
+  , messageType = publisher.messageType
+  , params      = [callerId, topicName, messageType, callerUri]
+  , client      = xmlrpc.createClient(masterUri)
+  ;
 
   client.methodCall('registerPublisher', params, function(error, value) {
     parseResponse(error, value, callback);

diff --git a/lib/tcpros.js b/lib/tcpros.js
@@ -23,19 +23,18 @@ TCPROS.prototype.createPublisher = function() {
   portscanner.findAPortNotInUse(9000, null, hostname, function(error, port) {
 
     var server = net.createServer(function(socket) {
-      // that.sockets.push(socket);
-      // var removeSocketFromSockets = function(socket) {
-      //   var index = that.sockets.indexOf(socket);
-      //   if (index>-1) {
-      //     that.sockets.splice(index, 1);
-      //   }
-      // }
-      that.socket = socket;
+       that.sockets.push(socket);
+       var removeSocketFromSockets = function(socket) {
+         var index = that.sockets.indexOf(socket);
+         if (index>-1) {
+           that.sockets.splice(index, 1);
+         }
+       }
       socket.on('end', function() {
-        // removeSocketFromSockets(socket);
+         removeSocketFromSockets(socket);
       });
       socket.on('error', function() {
-        // removeSocketFromSockets(socket);
+         removeSocketFromSockets(socket);
       });
 
       socket.on('data', function(data) {
@@ -100,23 +99,25 @@ TCPROS.prototype.createSubscriber = function(port, host, subscriber) {
 };
 
 TCPROS.prototype.publish = function(message) {
-  var publish = function() {
+  var that=this;
+  var publish = function(message) {
     var messageBuffer = serializeMessage(message);
     this.socket.write(messageBuffer);
   };
 
-  if (!this.socket) {
-    this.once('connect', publish);
+  if (!this.sockets.length) {
+    this.once('connect', function(){
+      //that.publish(message);
+    });
   }
   else {
-    publish();
+    var self=this;
+    async.forEach( this.sockets, function(socket, callback){
+      var messageBuffer = serializeMessage(message);
+      socket.write(messageBuffer);
+      callback();
+    });
   }
-  // var self=this;
-  // async.forEach( sockets, function(socket, callback){
-  //   var messageBuffer = serializeMessage(message);
-  //   socket.write(messageBuffer);
-  //   callback();
-  // }, callback);
 };
 
 function deserializeConnectionHeader(buffer) {
@@ -129,12 +130,12 @@ function deserializeConnectionHeader(buffer) {
     var fieldLength = buffer.readUInt32LE(bufferOffset);
     bufferOffset += 4;
     var fieldStart      = bufferOffset
-      , fieldEnd        = fieldStart + fieldLength
-      , field           = buffer.toString('utf8', fieldStart, fieldEnd)
-      , fieldComponents = field.split('=')
-      , fieldName       = fieldComponents[0]
-      , fieldValue      = fieldComponents[1]
-      ;
+    , fieldEnd        = fieldStart + fieldLength
+    , field           = buffer.toString('utf8', fieldStart, fieldEnd)
+    , fieldComponents = field.split('=')
+    , fieldName       = fieldComponents[0]
+    , fieldValue      = fieldComponents[1]
+    ;
     bufferOffset += fieldLength;
 
     if (fieldName === 'callerid') {
@@ -250,10 +251,10 @@ function serializeInnerMessage(message, buffer, bufferOffset) {
 
 function deserializeMessage(buffer, messageType) {
   var message            = new messageType()
-    , headerLength       = 0
-    , isConnectionHeader = false
-    , messageLength      = buffer.readUInt32LE(0)
-    , bufferOffset       = 4
+  , headerLength       = 0
+  , isConnectionHeader = false
+  , messageLength      = buffer.readUInt32LE(0)
+  , bufferOffset       = 4
 
   // If the initial length is less than the buffer, then a connection header was
   // sent first as part of this buffer.
@@ -289,9 +290,9 @@ function deserializeInnerMessage(message, buffer, bufferOffset) {
     }
     else if (fieldsUtil.isArray(field.type)) {
       var array     = []
-        , arraySize = buffer.readUInt32LE(bufferOffset)
-        , arrayType = fieldsUtil.getTypeOfArray(field.type)
-        ;
+      , arraySize = buffer.readUInt32LE(bufferOffset)
+      , arrayType = fieldsUtil.getTypeOfArray(field.type)
+      ;
       bufferOffset += 4;
 
       for (var i = 0; i < arraySize; i++) {