Topic.toStream

src/RosLibNode.js

@@ -4,5 +4,6 @@
 var assign = require('object-assign');
 
 module.exports = assign(require('./RosLib'), {
-  Ros: require('./node/RosTCP.js')
+  Ros: require('./node/RosTCP.js'),
+  Topic: require('./node/TopicStream')
 });

src/node/TopicStream.js

@@ -0,0 +1,18 @@
+var Topic = require('../core/Topic');
+var ReadableStream = require('stream').Readable;
+
+Topic.prototype.toStream = function() {
+    var stream = new ReadableStream({
+        objectMode: true
+    });
+    stream._read = function() {};
+
+    this.subscribe(function(message) {
+        stream.push(message);
+    });
+    this.on('unsubscribe', stream.push.bind(stream, null));
+
+    return stream;
+};
+
+module.exports = Topic;

test/examples/fibonacci.example.js

@@ -27,21 +27,21 @@ describe('Fibonacci Example', function() {
 
         // Print out their output into the terminal.
         var items = [
-            {"sequence": [0, 1, 1]},
-            {"sequence": [0, 1, 1, 2]},
-            {"sequence": [0, 1, 1, 2, 3]},
-            {"sequence": [0, 1, 1, 2, 3, 5]},
-            {"sequence": [0, 1, 1, 2, 3, 5, 8]},
-            {"sequence": [0, 1, 1, 2, 3, 5, 8, 13]},
-            {"sequence": [0, 1, 1, 2, 3, 5, 8, 13, 21]}
+            {'sequence': [0, 1, 1]},
+            {'sequence': [0, 1, 1, 2]},
+            {'sequence': [0, 1, 1, 2, 3]},
+            {'sequence': [0, 1, 1, 2, 3, 5]},
+            {'sequence': [0, 1, 1, 2, 3, 5, 8]},
+            {'sequence': [0, 1, 1, 2, 3, 5, 8, 13]},
+            {'sequence': [0, 1, 1, 2, 3, 5, 8, 13, 21]}
         ];
         goal.on('feedback', function(feedback) {
             console.log('Feedback:', feedback);
             expect(feedback).to.eql(items.shift());
         });
         goal.on('result', function(result) {
             console.log('Result:', result);
-            expect(result).to.eql({"sequence": [0, 1, 1, 2, 3, 5, 8, 13, 21]});
+            expect(result).to.eql({'sequence': [0, 1, 1, 2, 3, 5, 8, 13, 21]});
             done();
         });
 

test/examples/topic-listener.example.js

@@ -0,0 +1,69 @@
+var expect = require('chai').expect;
+var ROSLIB = require('../..');
+
+describe('Topics Example', function() {
+    this.timeout(1000);
+
+    var ros = new ROSLIB.Ros({
+        url: 'ws://localhost:9090'
+    });
+
+    var example = ros.Topic({
+        name: '/test_topic',
+        messageType: 'std_msgs/String'
+    });
+
+    function format(msg) {
+        return {data: msg};
+    }
+    var messages = ['1', '2', '3', '4'].map(format);
+
+    function createAndStreamTopic(topicName) {
+        var topic = ros.Topic({
+            name: topicName,
+            messageType: 'std_msgs/String'
+        });
+        var idx = 0;
+
+        function emit() {
+            setTimeout(function() {
+                topic.publish(messages[idx++]);
+                if (idx < messages.length) {
+                    emit();
+                } else {
+                    topic.unsubscribe();
+                    topic.unadvertise();
+                }
+            }, 50);
+        }
+        emit();
+
+        return topic;
+    }
+
+
+    it('Listening to a topic & unsubscribes', function(done) {
+        var topic = createAndStreamTopic('/echo/test');
+        var expected = messages.slice();
+
+        topic.subscribe(function(message) {
+            expect(message).to.be.instanceof(ROSLIB.Message);
+            expect(message).to.be.eql(expected.shift());
+        });
+
+        topic.on('unsubscribe', done);
+    });
+
+    if (ROSLIB.Topic.prototype.toStream) {
+        it('Topic.toStream()', function(done) {
+            var stream = createAndStreamTopic('/echo/test-stream').toStream();
+            var expected = messages.slice();
+
+            expect(stream).to.be.instanceof(require('stream'));
+            stream.on('data', function(message) {
+                expect(message).to.be.eql(expected.shift());
+            });
+            stream.on('end', done);
+        });
+    }
+});