Pubsub commands working (WIP)

Author: haadcode

src/api/pubsub.js

@@ -3,9 +3,9 @@
 const promisify = require('promisify-es6')
 const bs58 = require('bs58')
 var Base64 = require('js-base64').Base64
+var Stream = require('stream')
 // const Wreck = require('wreck')
 var http = require('http')
-var Writable = require('stream').Writable;
 
 module.exports = (send) => {
   const api = {
@@ -19,48 +19,28 @@ module.exports = (send) => {
       }
 
 
-      // var ws = Writable();
-      var Stream = require('stream');
-      var rs = new Stream;
-      rs.readable = true;
-
-      // ws._write = function (chunk, enc, next) {
-      //   // console.log('>', d.toString())
-      //   var parsed = JSON.parse(chunk)
-      //   parsed.from = bs58.encode(parsed.from)
-      //   parsed.data = Base64.decode(parsed.data)
-      //   parsed.seqno = Base64.decode(parsed.seqno)
-      //   console.log(parsed)
-      //   next();
-      // };
+      var rs = new Stream()
+      rs.readable = true
 
       console.log('Sub', topic)
       http.get({
         host: 'localhost',
         port: 5001,
         path: '/api/v0/pubsub/sub/' + topic
       }, function (response) {
-        // Continuously update stream with data
-        // var body = ''
-        // response.pipe
         response.on('data', function (d) {
-          // console.log("chunk", d.toString())
           var parsed = JSON.parse(d)
-          parsed.from = bs58.encode(parsed.from)
-          parsed.data = Base64.decode(parsed.data)
-          parsed.seqno = Base64.decode(parsed.seqno)
-          // console.log(parsed)
-          rs.emit('data', parsed)
-          // rs.push(JSON.stringify(parsed))
-          // ws.write(d)
-          // body += d
+
+          // skip "double subscription" error
+          if(!parsed.Message) {
+            parsed.from = bs58.encode(parsed.from)
+            parsed.data = Base64.decode(parsed.data)
+            parsed.seqno = Base64.decode(parsed.seqno)
+            rs.emit('data', parsed)
+          }
         })
         response.on('end', function () {
           rs.emit('end')
-          // Data reception is done, do whatever with it!
-          // var parsed = JSON.parse(body)
-          // console.log(parsed)
-          // callback(null, parsed)
         })
 
         callback(null, rs)
@@ -74,7 +54,6 @@ module.exports = (send) => {
       //     return callback(err)
       //   }
 
-      //   console.log("THIS IS THE CALLBACK")
       //   callback(null, response.pipe(ndjson.parse()))
       //   // callback(null, result) // result is a Stream
       // })

test/ipfs-api/pubsub.spec.js

@@ -35,9 +35,7 @@ describe('.pubsub', () => {
 
   it.only('sub', (done) => {
     ipfs.pubsub.sub('testi1', (err, result) => {
-      // console.log('RESULT1', err, result)
       expect(err).to.not.exist
-      // expect(result.length).to.equal(1)
       result.on('data', function (d) {
         // console.log("-->", d)
         expect(d.data).to.equal('hi')