Merge pull request #44 from datapartyjs/service-task

Service tasks

Author: sevenbitbyte

examples/tasks/status-display.js

@@ -0,0 +1,63 @@
+const debug = require('debug')('dataparty.task.status-update')
+
+const ITask = require('../../src/service/itask')
+
+class StatusDisplayTask extends ITask {
+
+  constructor(options){
+    super({
+      name: StatusDisplayTask.name,
+      background: StatusDisplayTask.Config.background,
+      ...options
+    })
+
+    debug('new')
+
+    this.duration = 5000
+    this.timeout = null
+  }
+
+  static get Config(){
+    return {
+      background: true,
+      autostart: true
+    }
+  }
+ 
+  async exec(){
+
+    this.setTimer()
+
+    return this.detach()
+  }
+
+  setTimer(){
+    this.timeout = setTimeout(this.onTimeout.bind(this), this.duration)
+  }
+ 
+  onTimeout(){
+    this.timeout = null
+    
+    this.context.serviceRunner.taskRunner.printTaskLists()
+    debug('sleep complete')
+
+    this.setTimer()
+  }
+ 
+  stop(){
+    if(this.timeout !== null){
+      clearTimeout(this.timeout)
+      this.timeout = null
+    }
+  }
+
+  static get Name(){
+    return 'status-display'
+  }
+
+  static get Description(){
+    return 'Status Display'
+  }
+}
+
+module.exports = StatusDisplayTask

examples/test-service-tasks.js

@@ -0,0 +1,64 @@
+const Path = require('path')
+const debug = require('debug')('test.server-db')
+const Dataparty = require('../src')
+
+const BouncerServerModels = require('@dataparty/bouncer-model')
+const BouncerClientModels = require('@dataparty/bouncer-model/dist/bouncer-model.json')
+
+class ExampleTaskService extends Dataparty.IService {
+  constructor(opts){
+    super(opts)
+
+    this.addTask(Path.join(__dirname,'./tasks/status-display.js'))
+  }
+
+}
+
+async function main(){
+
+  
+  //const uri = 'mongodb://localhost:27017/server-party-test'
+  //debug('db location', uri)
+
+  const path = '/data/datparty/srv-party'
+
+  let party = new Dataparty.TingoParty({
+    path,
+    model: BouncerClientModels,
+    serverModels: BouncerServerModels,
+    config: new Dataparty.Config.JsonFileConfig({basePath: '/data/datparty/'})
+  })
+
+  const service = new ExampleTaskService({ name: '@dataparty/task-example', version: '0.0.1' })
+
+  const build = await service.compile(Path.join(__dirname,'/dataparty'), true)
+
+  debug('built', Object.keys(build))
+
+  const runner = new Dataparty.ServiceRunnerNode({
+    party, service,
+    sendFullErrors: false,
+    useNative: true
+  })
+  
+  const host = new Dataparty.ServiceHost({
+    runner,
+    trust_proxy: true,
+    wsEnabled: true,
+    listenUri: 'http://localhost:8080'
+  })
+
+  await party.start()
+  await runner.start()
+  await host.start()
+
+  console.log('started')
+  
+  //process.exit()
+}
+
+
+
+main().catch(err=>{
+  console.error(err)
+})

package.json

@@ -1,7 +1,7 @@
 {
   "name": "@dataparty/api",
   "private": false,
-  "version": "1.2.15",
+  "version": "1.2.16",
   "main": "dist/dataparty.js",
   "frontend": "dist/dataparty-browser.js",
   "backend": "dist/dataparty.js",
@@ -59,6 +59,7 @@
   "dependencies": {
     "@dataparty/bouncer-db": "1.0.1",
     "@dataparty/crypto": "datapartyjs/dataparty-crypto#parcel-build",
+    "@dataparty/tasker": "^0.0.2",
     "@hapi/joi": "^17.1.1",
     "@zeit/ncc": "^0.22.3",
     "ajv": "6.9.1",

src/service/iservice.js

@@ -21,7 +21,8 @@ module.exports = class IService {
       middleware: {
         pre: {},
         post: {}
-      }
+      },
+      tasks: {}
     }
 
     this.middleware_order = {
@@ -36,7 +37,8 @@ module.exports = class IService {
       middleware: {
         pre: {},
         post: {}
-      }
+      },
+      tasks: {}
     }
 
     this.compiled = {
@@ -55,7 +57,27 @@ module.exports = class IService {
       middleware_order: {
         pre: [],
         post: []
-      }
+      },
+      tasks: {}
+    }
+
+    this.compileSettings = {
+      // provide a custom cache path or disable caching
+      cache: false,
+      // externals to leave as requires of the build
+      externals: ['debug', '@dataparty/crypto', '@dataparty/tasker', '@hapi/joi', '@hapi/hoek'],
+      // directory outside of which never to emit assets
+      //filterAssetBase: process.cwd(), // default
+      minify: false, // default
+      sourceMap: true, // default
+      //sourceMapBasePrefix: '../', // default treats sources as output-relative
+      // when outputting a sourcemap, automatically include
+      // source-map-support in the output file (increases output by 32kB).
+      sourceMapRegister: false, // default
+      watch: false, // default
+      v8cache: false, // default
+      quiet: false, // default
+      debugLog: false // default
     }
    }
 
@@ -112,6 +134,18 @@ module.exports = class IService {
     this.constructors.middleware[type][name] = middleware
   }
 
+  addTask(task_path){
+
+    debug('addTask', task_path)
+
+    const TaskClass = require(task_path)
+
+    const name = TaskClass.Name
+
+    this.sources.tasks[name] = task_path
+    this.constructors.tasks[name] = TaskClass
+  }
+
 
   async compile(outputPath, writeFile=true){
 
@@ -131,11 +165,14 @@ module.exports = class IService {
       this.compileMiddleware('post'),
       this.compileList('documents'),
       this.compileList('endpoints'),
+      this.compileList('tasks'),
       this.compileSchemas()
     ])
 
     this.compiled.middleware_order = this.middleware_order
 
+    this.compiled.compileSettings = this.compileSettings
+
     if(writeFile){
       const buildOutput = outputPath+'/'+ this.compiled.package.name.replace('/', '-') +'.dataparty-service.json'
       fs.writeFileSync(buildOutput, JSON.stringify(this.compiled, null,2))
@@ -179,24 +216,7 @@ module.exports = class IService {
   }
 
   async compileFileTo(input, output){
-    const { code, map, assets } = await NCC(input, {
-      // provide a custom cache path or disable caching
-      cache: false,
-      // externals to leave as requires of the build
-      externals: ['debug', '@dataparty/crypto', '@hapi/joi', '@hapi/hoek'],
-      // directory outside of which never to emit assets
-      //filterAssetBase: process.cwd(), // default
-      minify: true, // default
-      sourceMap: true, // default
-      //sourceMapBasePrefix: '../', // default treats sources as output-relative
-      // when outputting a sourcemap, automatically include
-      // source-map-support in the output file (increases output by 32kB).
-      sourceMapRegister: false, // default
-      watch: false, // default
-      v8cache: false, // default
-      quiet: false, // default
-      debugLog: false // default
-    })
+    const { code, map, assets } = await NCC(input, this.compileSettings)
 
     debug('compileFileTo', input, '->', output)
     debug('\t','code length', Math.round(code.length/1024), 'KB')

src/service/itask.js

@@ -0,0 +1,38 @@
+const debug = require('debug')('dataparty.service.ITask')
+const tasker = require('@dataparty/tasker')
+
+
+module.exports = class ITask extends tasker.Task {
+
+  constructor(options){
+    super(options)
+  }
+
+  static get Name(){
+    throw new Error('not implemented')
+  }
+
+  static get Description(){
+    throw new Error('not implemented')
+  }
+
+  static get Config(){
+    throw new Error('not implemented')
+
+    /**
+     * 
+     * return {
+     *  background: true
+     *  autostart: true
+     * }
+     */
+  }
+
+  static get info(){
+    return {
+      Name: this.Name,
+      Description: this.Description,
+      Config: this.Config
+    }
+  }
+}

src/service/service-runner-node.js

@@ -8,6 +8,7 @@ const EndpointContext = require('./endpoint-context')
 const DeltaTime = require('../utils/delta-time')
 
 const Router = require('origin-router').Router
+const Runner = require('@dataparty/tasker').Runner
 
 class ServiceRunnerNode {
   constructor({service, party, sendFullErrors=false, useNative=true}){
@@ -18,11 +19,31 @@ class ServiceRunnerNode {
 
     this.middleware = { pre: {}, post: {} }
     this.endpoint = {}
+    this.tasks = {}
 
     this.router = new Router()
+    this.taskRunner = new Runner()
   }
 
   async start(){
+
+    debug('starting tasks')
+
+    const taskMap = Hoek.reach(this.service, 'compiled.tasks')
+    //const endpointsLoading = []
+    for(let name in taskMap){
+      debug('\t',name)
+      await this.loadTask(name)
+    }
+
+    debug('tasks ready:')
+    for(let name in this.tasks){
+      debug('\t', name)
+    }
+
+    await this.taskRunner.start()
+
+
     debug('starting endpoints')
 
     const eps = Hoek.reach(this.service, 'compiled.endpoints')
@@ -40,6 +61,56 @@ class ServiceRunnerNode {
     }
   }
 
+  async loadTask(name){
+    if(this.tasks[name]){
+      return
+    }
+
+    debug('loadTask', name)
+
+    let dt = new DeltaTime().start()
+    
+
+    "use strict"
+    let task=null
+
+    let TaskClass = null
+
+    if(!this.useNative){
+      const build = Hoek.reach(this.service, `compiled.tasks.${name}`)
+      TaskClass = eval(build.code/*, build.map*/)
+    }
+    else{
+      TaskClass = this.service.constructors.tasks[name]
+    }
+
+    task = new TaskClass({
+      context:{
+        party: this.party,
+        serviceRunner: this
+      }
+    })
+
+
+    debug('task info', TaskClass.info)
+
+    this.tasks[name] = task
+
+    if(TaskClass.Config.autostart){
+      this.taskRunner.addTask(task)
+    }
+
+
+    dt.end()
+    debug('loaded task',name,'in',dt.deltaMs,'ms')
+  }
+
+  runTask(name){
+    const task = this.tasks[name]
+
+    this.taskRunner.addTask(task)
+  }
+
   async loadEndpoint(name){
     if(this.endpoint[name]){
       return