generally working

Author: sevenbitbyte

public/example.html

@@ -12,33 +12,52 @@
 
 async function init(){
 
-  //new Dataparty.Config.MemoryConfig()
-
-  config = new Dataparty.Config.LocalStorageConfig({
+  config = new Dataparty.Config.MemoryConfig({
     basePath:'demo',
     cloud: {
       uri: 'http://172.16.3.1:4001'
     }
   })
 
+  /*config = new Dataparty.Config.LocalStorageConfig({
+    basePath:'demo',
+    cloud: {
+      uri: 'http://172.16.3.1:4001'
+    }
+  })*/
+
 
   const remoteIdentity = await Dataparty.Comms.RestComms.HttpGet( config.read('cloud.uri') + '/identity')
 
   console.log('cloud.uri -',config.read('cloud.uri'))
   console.log('\tremoteIdentity', remoteIdentity)
 
   let party = new Dataparty.PeerParty({
-      comms: new Dataparty.Comms.WebsocketComms({
-        uri:'ws://172.16.3.1:4001/ws',
-        discoverRemoteIdentity: false,
-        remoteIdentity: remoteIdentity,
-        session: Math.random().toString(36).slice(2)
-      }),
-      config: config
-    })
-
+    comms: new Dataparty.Comms.WebsocketComms({
+      uri:'ws://172.16.3.1:4001/ws',
+      discoverRemoteIdentity: false,
+      remoteIdentity: remoteIdentity,
+      session: Math.random().toString(36).slice(2)
+    }),
+    config: config
+  })
 
   window.party = party
+  await party.start()
+
+  await party.comms.authorized()
+
+
+  timeTopic = new party.ROSLIB.Topic({
+    ros : party.comms.ros,
+    name : '/time',
+    messageType: 'number'
+  })
+
+  timeTopic.subscribe((msg)=>{
+    console.log(name, msg)
+  })
+
 }
 
 init()

src/comms/host/host-protocol-scheme.js

@@ -49,7 +49,7 @@ const UNADVERTISE_OP = Joi.object().keys({
 const SUBSCRIBE_OP = Joi.object().keys({
   id: ID_SCHEME.required(),
   op: Joi.string().valid('subscribe').required(),
-  type: Joi.string().required(),
+  type: Joi.string(),//.required(),
   topic: Joi.string().required(),
   compression: Joi.string(),
   queue_length: Joi.number(),

src/comms/peer-comms.js

@@ -57,12 +57,17 @@ class PeerComms extends SocketComms {
     try{
 
       let response = null
-      const request = await this.decrypt( {data: message}, this.remoteIdentity )
+      let request = await this.decrypt( {data: message}, this.remoteIdentity )
       debug('handleHostCall', request)
 
       let inputValidated
 
       if (this.state === PeerComms.STATES.AUTHED) {
+
+        if(typeof request != 'object'){
+          request = JSON.parse(request)
+        }
+
         debug('handling authed call')
         inputValidated = HostProtocolScheme.ANY_OP.validate(request)
       } else if (this.state === PeerComms.STATES.AUTH_REQUIRED) {
@@ -167,7 +172,7 @@ class PeerComms extends SocketComms {
       await this.socketInit()
     }
 
-    this.socket.on('close', this.close.bind(this))
+    this.socket.on('close', this.stop.bind(this))
 
     if(this.host){
       debug('host mode comms')
@@ -192,7 +197,7 @@ class PeerComms extends SocketComms {
   }
 
   async close(){
-    debug('close')
+    debug('close', this.uuid)
 
     if(this.party.topics){
       await this.party.topics.destroyNode(this)
@@ -211,6 +216,8 @@ class PeerComms extends SocketComms {
 
     debug('Here state : ', this.state)
 
+    console.log(op.input)
+
     if (op.op === 'auth' && this.state === PeerComms.STATES.AUTH_REQUIRED) {
 
       debug('handling auth op')
@@ -223,25 +230,25 @@ class PeerComms extends SocketComms {
     } else if (op.op === 'advertise' && this.state === PeerComms.STATES.AUTHED) {
 
       if(this.party.topics){
-        await this.party.topics.advertise(this, op.topic)
+        await this.party.topics.advertise(this, op.input.topic)
       }
 
     } else if (op.op === 'subscribe' && this.state === PeerComms.STATES.AUTHED) {
 
       if(this.party.topics){
-        await  this.party.topics.subscribe(this, op.topic)
+        await  this.party.topics.subscribe.bind(this.party.topics)(this, op.input.topic)
       }
 
     } else if (op.op === 'unsubscribe' && this.state === PeerComms.STATES.AUTHED) {
 
       if(this.party.topics){
-        await this.party.topics.unsubscribe(this, op.topic)
+        await this.party.topics.unsubscribe(this, op.input.topic)
       }
 
     } else if (op.op === 'publish' && this.state === PeerComms.STATES.AUTHED) {
 
       if(this.party.topics){
-        await this.party.topics.publish(this, op.topic, op.msg)
+        await this.party.topics.publish(this, op.input.topic, op.input.msg)
       }
 
     } else {

src/comms/websocket-comms.js

@@ -18,7 +18,7 @@ class WebsocketShim extends EventEmitter {
 
     this.conn.onopen = () => {
       debug('shim open')
-      this.emit('connect')
+      setTimeout(()=>{this.emit('connect')}, 1)
     }
 
     this.conn.onclose = () => {

src/party/iparty.js

@@ -1,6 +1,8 @@
 const debug = require('debug')('dataparty.iparty')
 const dataparty_crypto = require('@dataparty/crypto')
 
+const ROSLIB = require('roslib')
+
 const Query = require('./query.js')
 const IDocument = require('./idocument')
 const DocumentFactory = require('./document-factory')

src/party/peer/peer-party.js

@@ -34,7 +34,7 @@ class PeerParty extends IParty {
       debug('host')
       this.hostParty = hostParty
       this.hostRunner = hostRunner
-      this.topics = new LocalTopicHost()
+      this.topics = hostParty.topics || new LocalTopicHost()
     }
   }
 

src/topics/host-topic.js

@@ -9,6 +9,8 @@ class HostTopic {
     this.type = type
     this.created = Date.now()
 
+    debug('new topic', path, '(', type, ')')
+
     this.lastMessage = null
     this.lastMessageTime = null
 
@@ -18,7 +20,7 @@ class HostTopic {
 
   subscribe(node){
     if(!this.subscribers.get(node.uuid)){
-      this.subscribers.add(node.uuid, node)
+      this.subscribers.set(node.uuid, node)
     }
   }
 
@@ -30,7 +32,7 @@ class HostTopic {
 
   advertise(node){
     if(!this.advertisers.get(node.uuid)){
-      this.advertisers.add(node.uuid, node)
+      this.advertisers.set(node.uuid, node)
     }
   }
 
@@ -56,6 +58,12 @@ class HostTopic {
     this.lastMessageTime = Date.now()
 
     for(let node of this.subscribers){
+
+      if(node[0] == sender.uuid){
+        debug('publish skip', node[0])
+        continue
+      }
+
       sends.push(node[1].send(this, data, sender))
     }
 

src/topics/local-topic-host.js

@@ -13,30 +13,35 @@ class LocalTopicHost {
     this.topicsByPath = new Map()
   }
 
-  async getTopic(path, create=true){
+  getTopic(path, create=true){
     const normalized = Path.normalize(path)
     let topic = this.topicsByPath.get(normalized)
 
+    //debug('get topic', topic, normalized)
+
     if(!topic && create == true){
-      topic = new HostTopic(normalized)
-      this.topicsByPath.add(normalized, topic)
+      //topic = 
+      this.topicsByPath.set(normalized, new HostTopic(normalized))
+
+      //debug('set topic', this.topicsByPath.get(normalized), normalized)
+      return this.topicsByPath.get(normalized)
     }
 
     return topic
   }
 
-  async getNodeByUuid(uuid, peer){
+  getNodeByUuid(uuid, peer){
     let node = this.nodesByUuid.get(uuid)
 
     if(!node && peer){
       node = new PeerNode(peer)
-      this.nodesByUuid.add(uuid, node)
+      this.nodesByUuid.set(uuid, node)
     }
 
     return node
   }
 
-  async getNodeByPeer(peer){
+  getNodeByPeer(peer){
     return this.getNodeByUuid(peer.uuid, peer)
   }
 
@@ -51,6 +56,7 @@ class LocalTopicHost {
   }
 
   async subscribe(peer, path){
+    debug('sub', path)
     const topic = this.getTopic(path)
     const node = this.getNodeByPeer(peer)
 
@@ -100,13 +106,11 @@ class LocalTopicHost {
   async destroyNode(peer){
     const node = this.getNodeByPeer(peer)
 
-    debug('destroying node', node.uuid)
+    debug('destroying node', node.uuid, node)
 
     node.destroy()
 
     this.nodesByUuid.delete(node.uuid)
-
-    node=null
   }
 
   async cleanUpTopics(){

src/topics/peer-node.js

@@ -31,13 +31,13 @@ class PeerNode {
 
   advertise(topic){
     if(!this.advertisements.get(topic.path)){
-      this.advertisements.add(topic.path, topic)
+      this.advertisements.set(topic.path, topic)
     }
   }
 
   subscribe(topic){
     if(!this.subscriptions.get(topic.path)){
-      this.subscriptions.add(topic.path, topic)
+      this.subscriptions.set(topic.path, topic)
     }
   }