closes #87

Author: sevenbitbyte

examples/test-unix-socket-peer.js

@@ -0,0 +1,162 @@
+const Path = require('path')
+const debug = require('debug')('test.server-db')
+const Dataparty = require('../src')
+const dataparty_crypto = require('@dataparty/crypto')
+
+class ExampleService extends Dataparty.IService {
+  constructor(opts){
+    super(opts)
+
+    this.addMiddleware(Dataparty.middleware_paths.pre.decrypt)
+    this.addMiddleware(Dataparty.middleware_paths.pre.validate)
+
+    this.addMiddleware(Dataparty.middleware_paths.post.validate)
+    this.addMiddleware(Dataparty.middleware_paths.post.encrypt)
+
+    this.addEndpoint(Dataparty.endpoint_paths.echo)
+    this.addEndpoint(Dataparty.endpoint_paths.secureecho)
+    this.addEndpoint(Dataparty.endpoint_paths.identity)
+    this.addEndpoint(Dataparty.endpoint_paths.version)
+
+    this.addSchema(Path.join(__dirname, './party/schema/basic_types.js'))
+    this.addSchema(Path.join(__dirname, './party/schema/user.js'))
+  }
+
+}
+
+async function main(){
+
+  const socketFilePath = '/tmp/party.sock'
+
+  
+  const service = new ExampleService({ name: '@dataparty/example', version: '0.0.1' })
+  const build = await service.compile(Path.join(__dirname,'/dataparty'), true)
+  
+  const serviceName = build.package.name
+  const basePath = '/data/dataparty/'
+  const servicePath = Path.join(basePath, serviceName.replace('/','-'))
+
+  let config = new Dataparty.Config.JsonFileConfig({ basePath: servicePath })
+  config.touchDir('/tingo')
+
+  const dbPath = Path.join(servicePath, '/tingo')
+
+  let party = new Dataparty.TingoParty({
+    config,
+    path: dbPath,
+    model: build
+  })
+
+  const live = new Dataparty.IService(build.package, build)
+
+  
+  const runner = new Dataparty.ServiceRunnerNode({
+    party,
+    service: live,
+    sendFullErrors: false,
+    useNative: false
+  })
+
+  const runnerRouter = new Dataparty.RunnerRouter(runner)
+  
+  
+  const host = new Dataparty.ServiceHost({
+    runner: runnerRouter,
+    trust_proxy: true,
+    wsEnabled: true,
+    listenUri: 'file://'+socketFilePath,
+    wsUpgradePath: '/'
+  })
+  
+  debug(runner.party.identity)
+  await party.start()
+  await runnerRouter.start()
+  await host.start()
+
+  console.log('started')
+
+
+  const remoteIdentity = party.identity
+
+  debug('remoteIdentity', remoteIdentity)
+
+  let clientParty = new Dataparty.PeerParty({
+    comms: new Dataparty.Comms.WebsocketComms({
+      uri: 'ws+unix://'+socketFilePath,
+      remoteIdentity,
+      session: 'foobar'
+    }),
+    model: build,
+    config: new Dataparty.Config.MemoryConfig()
+  })
+
+  async function exitHandler(){
+
+    //! We must explictly stop server to clean up socket file
+
+    await clientParty.stop()
+
+    await host.stop()
+    process.exit()
+  }
+
+  process.on('exit', exitHandler)
+  process.on('SIGINT', exitHandler)
+
+
+
+  await clientParty.start()
+
+  debug('client waiting for auth')
+  await clientParty.comms.authorized()
+  debug('client authorized')
+
+  const remoteVersion = await clientParty.comms.call('version')
+  const remoteId = await clientParty.comms.call('identity')
+
+  debug('version', remoteVersion)
+  debug('identity', remoteId)
+
+  let user = (await clientParty.find()
+  .type('user')
+  .where('name').equals('tester')
+  .exec())[0]
+
+
+  if(!user){
+    debug('client creating document')
+    user = await clientParty.createDocument('user', {name: 'tester', created: (new Date()).toISOString() })
+  }
+  else{
+    debug('client loaded document')
+  }
+
+  console.log(user.data)
+
+  user.on('change', (obj)=>{ console.log('client remote event [document.on(change)]', obj.operationType ) })
+  user.on('update', (obj)=>{ console.log('client event [document.on(update)]')})
+  user.on('value', (doc)=>{ console.log('client event [document.on(value)]') })
+  user.on('remove', (obj)=>{ console.log('client event [document.on(remove)]') })
+
+  await user.watch()
+
+  let localUser = (await party.find()
+  .type('user')
+  .where('name').equals('tester')
+  .exec())[0]
+
+  console.log('\nserver changing document field')
+  localUser.data.name = 'renamed-tester'
+  await localUser.save()
+
+  console.log(localUser.data)
+  console.log('hash',localUser.hash)
+
+  await localUser.remove()
+}
+
+
+
+main().catch(err=>{
+  console.error(err)
+})

package.json

@@ -76,7 +76,7 @@
     "dom-storage": "^2.1.0",
     "eventemitter3": "^4.0.7",
     "express": "^4.17.1",
-    "express-list-routes": "^0.1.4",
+    "express-list-routes": "^1.1.9",
     "git-repo-info": "^2.1.1",
     "joi": "^17.9.1",
     "joi-objectid": "^4.0.2",

src/comms/isocket-comms.js

@@ -22,7 +22,7 @@ const IParty = require('../party/iparty')
  */
 
 class ISocketComms extends EventEmitter {
-    constructor({session, party, remoteIdentity, discoverRemoteIdentity}){
+    constructor({session, uri, party, remoteIdentity, discoverRemoteIdentity}){
         super()
         this.uri = uri
         this.session = session

src/comms/websocket-shim.js

@@ -2,6 +2,8 @@
 const debug = require('debug')('dataparty.comms.websocket-shim')
 const EventEmitter = require('eventemitter3')
 
+const WebSocket = global.WebSocket ? global.WebSocket : require('ws')
+
 class WebsocketShim extends EventEmitter {
     constructor(conn){
       super()

src/party/iparty.js

@@ -70,6 +70,10 @@ class IParty {
     debug('\tDocument Classes', this.factory.getTypes())
   }
 
+  async stop(){
+    this.comms.close()
+  }
+
   /**
    * @async
    * @method module:Party.IParty.createDocument

src/service/service-host.js

@@ -12,8 +12,10 @@ const reach = require('../utils/reach')
 
 const ServiceHostWebsocket = require('./service-host-websocket')
 
-const Pify = async (p)=>{
-  return await p
+const Pify = (p)=>{
+  return new Promise((resolve,reject)=>{
+    p(resolve)
+  })
 }
 
 class ServiceHost {
@@ -140,24 +142,31 @@ class ServiceHost {
       //Setup router
       this.apiApp.use(this.runner.onRequest.bind(this.runner))
 
-      if(debug.enabled){ expressListRoutes('API:', this.router ) }
+      if(debug.enabled){ expressListRoutes(this.router ) }
     }
 
+    let backlog = 511
     let listenPort = this.apiServerUri.port
     let listenHost = this.apiServerUri.hostname
 
     if(this.apiServerUri.protocol == 'http:'){
 
+      debug('http server')
+
       //! Handle http
       this.apiServer = http.createServer(this.apiApp)
 
     } else if(this.apiServerUri.protocol == 'https:'){
 
+      debug('http server')
+
       //! Handle https
       this.apiServer = https.createServer(this.apiApp)
 
     } else if(this.apiServerUri.protocol == 'file:'){
 
+      debug('unix socket server')
+
       //! Handle unix socket
       listenHost = null
       listenPort = this.apiServerUri.pathname
@@ -167,7 +176,10 @@ class ServiceHost {
 
 
     await new Promise((resolve,reject)=>{
-      this.apiServer.listen(listenPort, listenHost, resolve)
+
+      debug('listening', this.apiServerUri.toString())
+
+      this.apiServer.listen(listenPort, listenHost==null ? backlog : listenHost, resolve)
     })
 
     clearTimeout(this.errorHandlerTimer)
@@ -217,16 +229,19 @@ class ServiceHost {
    * @async
    */
   async stop(){
-    debug('stopping server')
-
+    
     if(!this.apiServer || !this.apiServer.listening){
       return
     }
-
+    
+    debug('stopping server')
+    
     clearTimeout(this.errorHandlerTimer)
     this.errorHandlerTimer = null
 
-    await (Pify(this.apiServer.close)())
+    await new Promise((resolve,reject)=>{
+      this.apiServer.close(resolve)
+    })
 
     debug('stopped server')
   }