local topic host

sevenbitbyte · sevenbitbyte · commit e76e5227a37f · 2023-03-14T12:54:04.000Z
diff --git a/examples/test-peer-party.js b/examples/test-peer-party.js
@@ -42,6 +42,8 @@ async function main(){
   await peer1.loadIdentity()
   await peer2.loadIdentity()
 
+  console.log('peer identity', peer1.identity)
+
   //peer1.comms.remoteIdentity = peer2.identity
   peer2.comms.remoteIdentity = peer1.identity
 
diff --git a/examples/test-service-node-host.js b/examples/test-service-node-host.js
@@ -31,13 +31,16 @@ async function main(){
 
   const path = '/data/datparty/srv-party'
 
+
   let party = new Dataparty.TingoParty({
     path,
     model: BouncerClientModels,
     serverModels: BouncerServerModels,
     config: new Dataparty.Config.JsonFileConfig({basePath: '/data/datparty/'})
   })
 
+  party.topics = new Dataparty.LocalTopicHost()
+
   const service = new ExampleService({ name: '@dataparty/example', version: '0.0.1' })
 
   const build = await service.compile(Path.join(__dirname,'/dataparty'), true)
@@ -46,13 +49,14 @@ async function main(){
 
   const runner = new Dataparty.ServiceRunnerNode({
     party, service,
-    sendFullErrors: false
+    sendFullErrors: false,
+    useNative: false
   })
   
   const host = new Dataparty.ServiceHost({
     runner,
     trust_proxy: true,
-    wsEnabled: true
+    wsEnabled: true,
   })
 
   await party.start()
diff --git a/package.json b/package.json
@@ -123,7 +123,9 @@
     "minami": "^1.2.3",
     "mongodb-client-encryption": "^2.2.1",
     "parcel": "^2.3.1",
+    "path-browserify": "^1.0.1",
     "process": "^0.11.10",
+    "punycode": "2.3.0",
     "querystring-es3": "^0.2.1",
     "stream-browserify": "^3.0.0",
     "stream-http": "^3.2.0",
diff --git a/public/example.html b/public/example.html
@@ -5,4 +5,47 @@
 </script>
 </head>
 
+<body>
+
+
+<script>
+
+async function init(){
+
+  //new Dataparty.Config.MemoryConfig()
+
+  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
+    })
+
+
+  window.party = party
+}
+
+init()
+
+
+</script>
+
+</body>
+
 </html>
diff --git a/src/comms/peer-comms.js b/src/comms/peer-comms.js
@@ -1,5 +1,5 @@
 const debug = require('debug')('dataparty.comms.peercomms')
-
+const uuidv4 = require('uuid/v4')
 const HttpMocks = require('node-mocks-http')
 
 const SocketOp = require('./op/socket-op')
@@ -22,6 +22,7 @@ class PeerComms extends SocketComms {
   constructor({remoteIdentity, discoverRemoteIdentity, host, party, socket, ...options}){
     super({remoteIdentity, discoverRemoteIdentity, party, ...options})
 
+    this.uuid = uuidv4()
     this.socket = socket || null
 
     this.host = host   //! Is comms host\
@@ -166,7 +167,7 @@ class PeerComms extends SocketComms {
       await this.socketInit()
     }
     
-    this.socket.on('close', this.onclose.bind(this))
+    this.socket.on('close', this.close.bind(this))
 
     if(this.host){
       debug('host mode comms')
@@ -186,12 +187,21 @@ class PeerComms extends SocketComms {
   }
 
   async stop(){
+    debug('stop')
     this.close()
   }
 
-  close(){
+  async close(){
+    debug('close')
+
+    if(this.party.topics){
+      await this.party.topics.destroyNode(this)
+    }
+
     debug('closing connection')
     this.socket.destroy()
+
+    this.onclose()
   }
 
 
@@ -210,7 +220,34 @@ class PeerComms extends SocketComms {
 
       return this.handleCallOp(op)
 
+    } else if (op.op === 'advertise' && this.state === PeerComms.STATES.AUTHED) {
+
+      if(this.party.topics){
+        await this.party.topics.advertise(this, op.topic)
+      }
+
+    } else if (op.op === 'subscribe' && this.state === PeerComms.STATES.AUTHED) {
+
+      if(this.party.topics){
+        await  this.party.topics.subscribe(this, op.topic)
+      }
+
+    } else if (op.op === 'unsubscribe' && this.state === PeerComms.STATES.AUTHED) {
+
+      if(this.party.topics){
+        await this.party.topics.unsubscribe(this, op.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)
+      }
+
     } else {
+      debug('⚠️ op not implemented ⚠️')
+      debug(op.input)
+
       op.result='not implemented'
       op.setState(HostOp.STATES.Finished_Fail)
     }
diff --git a/src/comms/websocket-comms.js b/src/comms/websocket-comms.js
@@ -1,6 +1,6 @@
 const debug = require('debug')('dataparty.comms.websocket')
 
-const WebSocket = require('ws')
+const WebSocket = global.WebSocket ? global.WebSocket : require('ws')
 const EventEmitter = require('eventemitter3')
 
 
diff --git a/src/index-browser.js b/src/index-browser.js
@@ -1,5 +1,6 @@
 const Comms = require('./comms')
 const Party = require('./party/index-browser')
+const Topics = require('./topics')
 
 const MemoryConfig = require('./config/memory')
 const LocalStorageConfig = require('./config/local-storage')
@@ -13,7 +14,8 @@ const Config = {
 let lib = {
   Comms,
   Config,
-  ...Party
+  ...Party,
+  ...Topics
 }
 
 
diff --git a/src/index-embedded.js b/src/index-embedded.js
@@ -1,11 +1,13 @@
 const Comms = require('./comms')
 const Config = require('./config')
 const Party = require('./party/index-embedded')
+const Topics = require('./topics')
 const Bouncer = require('./bouncer')
 
 module.exports = {
   Comms,
   Config,
   Bouncer,
-  ...Party
+  ...Party,
+  ...Topics
 }
diff --git a/src/index.js b/src/index.js
@@ -1,6 +1,7 @@
 const Comms = require('./comms')
 const Config = require('./config')
 const Party = require('./party')
+const Topics = require('./topics')
 const Bouncer = require('./bouncer')
 const Service = require('./service')
 const Sandbox = require('./sandbox')
@@ -10,6 +11,7 @@ module.exports = {
   Config,
   Bouncer,
   ...Party,
+  ...Topics,
   ...Service,
-  ...Sandbox
-}
+  ...Sandbox,
+}
diff --git a/src/party/peer/peer-party.js b/src/party/peer/peer-party.js
@@ -5,6 +5,8 @@ const debug = require('debug')('dataparty.peer-party')
 const Qb = require('../qb')
 const IParty = require('../iparty')
 
+const LocalTopicHost = require('../../topics/local-topic-host')
+
 /**
  * @class 
  * @alias module:dataparty.LocalParty
@@ -29,8 +31,10 @@ class PeerParty extends IParty {
     this.hostRunner = null
 
     if(this.comms.host){
+      debug('host')
       this.hostParty = hostParty
       this.hostRunner = hostRunner
+      this.topics = new LocalTopicHost()
     }
   }
 
diff --git a/src/service/endpoints/echo.js b/src/service/endpoints/echo.js
@@ -1,6 +1,4 @@
 const Joi = require('@hapi/joi')
-const Hoek = require('@hapi/hoek')
-const {Message, Routines} = require('@dataparty/crypto')
 const debug = require('debug')('dataparty.endpoint.echo')
 
 const IEndpoint = require('../iendpoint')
diff --git a/src/service/service-host-websocket.js b/src/service/service-host-websocket.js
@@ -102,6 +102,7 @@ class ServiceHostWebsocket{
 
     let peer = new PeerParty({
       hostParty: this.runner.party,
+      hostRunner: this.runner,
       model: this.runner.party.factory.model,
       config: this.runner.party.config,
       comms: new Comms.WebsocketComms({
diff --git a/src/service/service-runner-node.js b/src/service/service-runner-node.js
@@ -1,7 +1,6 @@
 const Path = require('path')
 const Joi = require('@hapi/joi')
 const Hoek = require('@hapi/hoek')
-const {VM, VMScript} = require('vm2')
 const Debug = require('debug')
 const debug = Debug('dataparty.service.runner-node')
 const EndpointContext = require('./endpoint-context')
@@ -66,7 +65,7 @@ class ServiceRunnerNode {
       return
     }
 
-    debug('loadTask', name)
+    debug('loadTask', name, 'useNative =',this.useNative)
 
     let dt = new DeltaTime().start()
     
@@ -116,7 +115,7 @@ class ServiceRunnerNode {
       return
     }
 
-    debug('loadEndpoint', name)
+    debug('loadEndpoint', name, 'useNative =',this.useNative)
 
     let dt = new DeltaTime().start()
     
@@ -168,7 +167,7 @@ class ServiceRunnerNode {
       return this.middleware[type][name]
     }
 
-    debug('loadMiddleware', type, name)
+    debug('loadMiddleware', type, name, 'useNative =',this.useNative)
 
     let dt = new DeltaTime().start()
     const build = Hoek.reach(this.service, `compiled.middleware.${type}.${name}`)
diff --git a/src/topics/host-topic.js b/src/topics/host-topic.js
@@ -0,0 +1,69 @@
+'use strict'
+
+const Path = require('path')
+const debug = require('debug')('dataparty.topics.peer-node')
+
+class HostTopic {
+  constructor(path, type){
+    this.path = Path.normalize(path)
+    this.type = type
+    this.created = Date.now()
+
+    this.lastMessage = null
+    this.lastMessageTime = null
+
+    this.subscribers = new Map()
+    this.advertisers = new Map()
+  }
+
+  subscribe(node){
+    if(!this.subscribers.get(node.uuid)){
+      this.subscribers.add(node.uuid, node)
+    }
+  }
+
+  unsubscribe(node){
+    if(this.subscribers.get(node.uuid)){
+      this.subscribers.delete(node.uuid)
+    }
+  }
+
+  advertise(node){
+    if(!this.advertisers.get(node.uuid)){
+      this.advertisers.add(node.uuid, node)
+    }
+  }
+
+  unadvertise(node){
+    if(this.advertisers.get(node.uuid)){
+      this.advertisers.delete(node.uuid)
+    }
+  }
+
+  canPublish(sender){
+    return this.advertisers.has(sender.uuid)
+  }
+
+  async publish(data, sender=null){
+
+    let sends = []
+
+    if(sender!=null && !this.canPublish(sender)){
+      throw new Error('published called before advertise')
+    }
+
+    this.lastMessage = data
+    this.lastMessageTime = Date.now()
+
+    for(let node of this.subscribers){
+      sends.push(node[1].send(this, data, sender))
+    }
+
+    if(sends.length > 0){
+      debug('publishing', this.path)
+      await Promise.all(sends)
+    }
+  }
+}
+
+module.exports = HostTopic
diff --git a/src/topics/index.js b/src/topics/index.js
@@ -0,0 +1,3 @@
+module.exports = {
+  LocalTopicHost: require('./local-topic-host')
+}
diff --git a/src/topics/local-topic-host.js b/src/topics/local-topic-host.js
diff --git a/src/topics/peer-node.js b/src/topics/peer-node.js
