fix remote document change notifications

Author: sevenbitbyte

examples/test-loopback-party.js

@@ -4,6 +4,15 @@ const WRTC = require('wrtc')
 const BouncerModel = require('@dataparty/bouncer-model/dist/bouncer-model.json')
 const Dataparty = require('../src')
 
+
+async function getUser(party, name) {
+  return (await party.find()
+    .type('user')
+    .where('name').equals(name)
+    .exec())[0]
+}
+
+
 async function main(){
   const dbPath = await fs.mkdtemp('/tmp/tingo-party')
   const configPath = await fs.mkdtemp('/tmp/tingo-party-config')
@@ -18,11 +27,13 @@ async function main(){
     config
   })
 
+  hostLocal.topics = new Dataparty.LocalTopicHost()
+
   let loopback = new Dataparty.Comms.LoopbackChannel()
 
   let comms1 = new Dataparty.Comms.LoopbackComms({
     host: true,
-    channel: loopback.peer1
+    channel: loopback.port1
   })
 
   let peer1 = new Dataparty.PeerParty({
@@ -33,7 +44,9 @@ async function main(){
   })
 
 
-  let comms2 = new Dataparty.Comms.LoopbackComms({ channel: loopback.peer2, session: 'foobar' })
+
+
+  let comms2 = new Dataparty.Comms.LoopbackComms({ channel: loopback.port2, session: 'foobar' })
 
   let peer2 = new Dataparty.PeerParty({
     comms: comms2,
@@ -71,16 +84,36 @@ async function main(){
 
 
   if(!user){
-    debug('creating document')
+    console.log('peer2 creating document')
     user = await peer2.createDocument('user', {name: 'tester', created: (new Date()).toISOString() })
   }
   else{
-    debug('loaded document')
+    console.log('peer2 loaded document')
   }
-    
+  
+    console.log(user.data)
+    console.log('hash',user.hash)
+
+  console.log('peer1 find document by new field value')
+  let userFind = await getUser(peer1,'tester')
+  console.log(userFind.data)
+  console.log('hash',userFind.hash)
+
+  user.on('change', (obj)=>{ console.log('peer2 remote event [document.on(change)]', obj ) })
+  user.on('update', (obj)=>{ console.log('peer2 event [document.on(update)]')})
+  user.on('value', (doc)=>{ console.log('peer2 event [document.on(value)]') })
+  user.on('remove', (obj)=>{ console.log('peer2 event [document.on(remove)]') })
+
+  await user.watch()
+
+  console.log('\npeer1 changing document field')
+  userFind.data.name = 'renamed-tester'
+  await userFind.save()
+
+  console.log(userFind.data)
+  console.log('hash',userFind.hash)
 
-  console.log(user.data)
-  process.exit()
+  await userFind.remove()
 }
 
 

src/bouncer/idb.js

@@ -60,6 +60,7 @@ class IDb extends EventEmitter {
 
   emitChange(msg, change){
     const { type, id, revision } = msg.$meta
+    debug('emitChange', `${type}:${id}`)
     this.emit(
       `${type}:${id}`,
       {

src/comms/isocket-comms.js

@@ -126,7 +126,7 @@ class ISocketComms extends EventEmitter {
             debug(msg.id)
 
             if(msg.op != 'publish'){
-                debug('emit id')
+                debug('emit id', msg.id)
                 comm.emit(msg.id, msg)
             } else {
                 debug('emit message')
@@ -136,7 +136,11 @@ class ISocketComms extends EventEmitter {
     }
 
     send(input){
-        debug('send - ', input)
+        debug('send - ', typeof input, input)
+
+        if(typeof input != 'object'){
+            input = JSON.parse(input)
+        }
 
         const content = new Message({msg: input})
 

src/comms/loopback-channel-port.js

@@ -1,5 +1,5 @@
 const debug=require('debug')('dataparty.comms.loopback-channel-port')
-
+const EventEmitter = require("eventemitter3")
 
 class LoopbackChannelPort {
     constructor(peer, name){

src/comms/loopback-channel.js

@@ -17,7 +17,7 @@ module.exports = class LoopbackChannel {
     this.port1 = new LoopbackChannelPort(undefined, '1')
 
     //! The second channel peer
-    this.peer2 = new LoopbackChannelPort(this.port1, '2')
+    this.port2 = new LoopbackChannelPort(this.port1, '2')
 
     this.port1.peer = this.port2
   }

src/comms/ros-shim.js

@@ -27,8 +27,6 @@ class RosShim extends ROSLIB.Ros {
 
     handleMessage(message) {
 
-      console.log(message)
-
         if (message.op === 'publish') {
           debug('publish op')
           this.emit(message.topic, message.msg);

src/party/idocument.js

@@ -239,6 +239,10 @@ class IDocument extends EventEmitter {
       .id(this.id)
       .exec()
       .then(docs => {
+        if(docs.length==0){
+          return
+        }
+
         this._data = docs[0].data
 
         debug('pull found', docs)
@@ -268,8 +272,7 @@ class IDocument extends EventEmitter {
 
     if (this.watchSub){ return }
 
-    const socket = await this.party.socket()
-    const ros = socket.ros
+    const ros = this.party.comms.ros
     const watchPath = '/dataparty/document/' + this.type + '/' + this.id
 
     debug('watch document', watchPath)
@@ -355,7 +358,7 @@ class IDocument extends EventEmitter {
       case 'update':
       case 'create':
 
-        if(this.followcache){ 
+        if(this.followcache && event.event !== 'remove'){ 
           this._data = newMsg 
 
           /**

src/party/loki-cache.js

@@ -32,7 +32,6 @@ module.exports = class LokiCache extends EventEmitter {
   }
 
   remove(type, id){
-    console('cache remove')
     debug('remove', type, id)
     var collection = this.db.getCollection(type)
 

src/topics/host-topic.js

@@ -1,7 +1,7 @@
 'use strict'
 
 const Path = require('path')
-const debug = require('debug')('dataparty.topics.peer-node')
+const debug = require('debug')('dataparty.topics.host-topic')
 
 class HostTopic {
   constructor(path, type){

src/topics/local-topic-host.js

@@ -77,16 +77,20 @@ class LocalTopicHost {
 
     if(topic.path.indexOf('/dataparty/document/') != -1 && !exists){
       const [arg0, arg1, docType, docId] = topic.path.substr(1).split('/')
+      debug('is document watcher', docType+':'+docId)
 
-      peer.hostParty.db.on(docType+':'+docId, async (event)=>{
+      peer.party.hostParty.db.on(docType+':'+docId, async (event)=>{
         await this.handleDocChange(topic.path, event)
       })
     }
   }
 
   async handleDocChange(path, event){
+    debug('handleDocChange', path)
     const topic = this.getTopic(path,false)
 
+    debug('\ttopic',topic)
+
     if(!topic){return}
 
     const [arg0, arg1, docType, docId] = topic.path.substr(1).split('/')
@@ -96,7 +100,7 @@ class LocalTopicHost {
         id: event.msg.id,
         type: event.msg.type,
         revision: event.msg.revision,
-        operationType: event.change
+        operationType: event.event
       })
     }
   }