terminate pub/sub operations

Author: sevenbitbyte

src/comms/peer-comms.js

@@ -18,6 +18,25 @@ const HOST_SESSION_STATES = {
   CLIENT_CLOSED: 'CLIENT_CLOSED'
 }
 
+
+function truncateString(str, num) {
+
+  if(!str){return ''}
+  
+  if(typeof str != 'string'){
+    str = str.toString()
+  }
+
+  let length = str.length
+
+  if (str.length <= num) {
+    return str
+  }
+  return str.slice(0, num) + '...' + (length-num) + 'more bytes'
+}
+
+
+
 class PeerComms extends SocketComms {
   constructor({remoteIdentity, discoverRemoteIdentity, host, party, socket, ...options}){
     super({remoteIdentity, discoverRemoteIdentity, party, ...options})
@@ -58,7 +77,7 @@ class PeerComms extends SocketComms {
 
       let response = null
       let request = await this.decrypt( {data: message}, this.remoteIdentity )
-      debug('handleHostCall', request)
+      debug('handleHostCall', truncateString(request, 1024))
 
       let inputValidated
 
@@ -85,8 +104,8 @@ class PeerComms extends SocketComms {
         throw inputValidated.error
       }
 
-      debug('original input ->', typeof request, request)
-      debug('validated input ->', inputValidated)
+      //debug('original input ->', typeof request, request)
+      //debug('validated input ->', inputValidated)
       const op = new HostOp({ msg: message, input: inputValidated.value })
 
       /*debug('session id : ', op.input.session, this.session)
@@ -109,6 +128,8 @@ class PeerComms extends SocketComms {
           ...op.result 
         }
 
+        debug(response.id, response.state, response.stats.duration_ms, 'ms')
+
         this.send(response)
       })
 
@@ -133,6 +154,8 @@ class PeerComms extends SocketComms {
     )
   }
 
+
+
   async handleAuthTimeout(){
     clearTimeout(this._host_auth_timeout)
     this._host_auth_timeout = null
@@ -144,7 +167,7 @@ class PeerComms extends SocketComms {
   }
 
   async handleMessage(message){
-    debug('handleMessage', message.toString())
+    debug('handleMessage', truncateString(message.toString(), 1024) )
 
     this.onmessage({data: message})
   }
@@ -159,7 +182,7 @@ class PeerComms extends SocketComms {
 
     let callOp = new SocketOp( 'peer-call', { endpoint: path, data }, this )
 
-    debug('running peer-call endpoint =', path, data)
+    debug('running peer-call endpoint =', path, truncateString(data, 1024))
 
     const reply = await callOp.run()
 
@@ -212,11 +235,11 @@ class PeerComms extends SocketComms {
 
   async authorizeOperation(op) {
 
-    debug('Here\'s op', op)
+    //debug('Here\'s op', op)
 
-    debug('Here state : ', this.state)
+    //debug('state : ', this.state)
 
-    console.log(op.input)
+    //console.log(op.input)
 
     if (op.op === 'auth' && this.state === PeerComms.STATES.AUTH_REQUIRED) {
 
@@ -231,24 +254,40 @@ class PeerComms extends SocketComms {
 
       if(this.party.topics){
         await this.party.topics.advertise(this, op.input.topic)
+        op.setState(HostOp.STATES.Finished_Success)
+      }
+      else{
+        op.setState(HostOp.STATES.Finished_Fail)
       }
 
     } else if (op.op === 'subscribe' && this.state === PeerComms.STATES.AUTHED) {
 
       if(this.party.topics){
         await  this.party.topics.subscribe.bind(this.party.topics)(this, op.input.topic)
+        op.setState(HostOp.STATES.Finished_Success)
+      }
+      else{
+        op.setState(HostOp.STATES.Finished_Fail)
       }
 
     } else if (op.op === 'unsubscribe' && this.state === PeerComms.STATES.AUTHED) {
 
       if(this.party.topics){
         await this.party.topics.unsubscribe(this, op.input.topic)
+        op.setState(HostOp.STATES.Finished_Success)
+      }
+      else{
+        op.setState(HostOp.STATES.Finished_Fail)
       }
 
     } else if (op.op === 'publish' && this.state === PeerComms.STATES.AUTHED) {
 
       if(this.party.topics){
         await this.party.topics.publish(this, op.input.topic, op.input.msg)
+        op.setState(HostOp.STATES.Finished_Success)
+      }
+      else{
+        op.setState(HostOp.STATES.Finished_Fail)
       }
 
     } else {
@@ -283,7 +322,7 @@ class PeerComms extends SocketComms {
       debug('calling runner')
 
       if(op.input.endpoint == 'api-v2-peer-bouncer'){
-        debug('ask->',op.input.data)
+        debug('ask->', truncateString(op.input.data, 1024))
         op.result = {result: await this.party.handleCall(op.input.data) }
 
         op.setState(HostOp.STATES.Finished_Success)

src/topics/host-topic.js

@@ -16,6 +16,9 @@ class HostTopic {
 
     this.subscribers = new Map()
     this.advertisers = new Map()
+
+    this.sending = false
+    this.sendRequest = false
   }
 
   subscribe(node){
@@ -54,22 +57,58 @@ class HostTopic {
       throw new Error('published called before advertise')
     }
 
-    this.lastMessage = data
-    this.lastMessageTime = Date.now()
+    if(!this.sending && this.sendRequest){
+      
+      this.sendRequest=false
+
+    }else  if(this.sending){
+      if(!this.sendRequest){
+        this.sendRequest = true
+      }
+
+      debug('throttle')
+      this.lastMessage = data
+      this.lastMessageTime = Date.now()
+
+      return
+
+    }else if(!this.sendRequest && !this.sending) {
+
+      this.lastMessage = data
+      this.lastMessageTime = Date.now()
+    }
+
+    
 
     for(let node of this.subscribers){
 
-      if(node[0] == sender.uuid){
+      if(sender && node[0] == sender.uuid){
         debug('publish skip', node[0])
         continue
       }
 
-      sends.push(node[1].send(this, data, sender))
+      //await node[1].send(this, this.lastMessage, sender)
+
+      sends.push(node[1].send(this, this.lastMessage, sender))
     }
 
     if(sends.length > 0){
       debug('publishing', this.path)
-      await Promise.all(sends)
+      this.sending = true
+
+      try{
+        await Promise.all(sends)
+        this.sending = false
+
+        if(this.sendRequest){
+          debug('send requested')
+          //setTimeout(this.publish.bind(this),1)
+        }
+      }
+      catch(err){
+        this.sending = false
+        throw err
+      }
     }
   }
 }

src/topics/local-topic-host.js

@@ -106,7 +106,7 @@ class LocalTopicHost {
   async destroyNode(peer){
     const node = this.getNodeByPeer(peer)
 
-    debug('destroying node', node.uuid, node)
+    debug('destroying node', node.uuid)
 
     node.destroy()
 

src/topics/peer-node.js

@@ -14,6 +14,9 @@ class PeerNode {
   }
 
   async send(topic, data, sender=null){
+
+    if(!data){return}
+
     if(this.subscriptions.has(topic.path)){
 
       debug(' publish to node', this.uuid)
@@ -26,6 +29,14 @@ class PeerNode {
         msg: data
       })
 
+      /*await this.peer.socket.send(JSON.stringify({
+        op: 'publish',
+        id: 'publish:'+this.peer.opId,
+        topic: topic.path,
+        sender: { uuid: this.uuid, identity: this.peer.remoteIdentity },
+        msg: data
+      }))*/
+
     }
   }