Add data stores for stream states

Author: starwarfan

source/stream_service/controllers/typeController.js

@@ -40,17 +40,14 @@ class TypeController extends EventEmitter {
    */
   constructor(rpc) {
     super();
-    log.debug('rpc:', rpc);
     this.makeRPC = rpc.makeRPC.bind(rpc);
-    log.debug('this.makeRPC:', this.makeRPC);
     this.selfId = rpc.selfId;
     this.clusterId = rpc.clusterId;
   }
 
   // Return Promise<{agent: string, node: string}>
   async getWorkerNode(purpose, domain, taskId, preference) {
     log.debug('getWorkerNode:', purpose, domain, taskId, preference);
-    log.debug('this:', this);
     const args =  [purpose, taskId, preference, 30 * 1000];
     return this.makeRPC(this.clusterId, 'schedule', args)
       .then((workerAgent) => {
@@ -70,49 +67,5 @@ class TypeController extends EventEmitter {
   }
 }
 
-class LocalState {
-  constructor(type) {
-    this.type = type ? type + '.' : '';
-    this.tsc = false;
-    this.state = new Map();
-  }
-  async create(key, value) {
-    key = type + key;
-    if (this.state.has(key)) {
-      throw new Error('Duplicate key');
-    }
-    this.state.set(key, value);
-    return value;
-  }
-  async read(key, value) {
-    key = type + key;
-    return this.state.get(key, value);
-  }
-  // condition: {op: string, path: string, value: any}
-  async update(key, value, condition) {
-    key = type + key;
-    if (!this.state.has(key)) {
-      throw new Error(`${key} not found`);
-    }
-    const prev = this.state.get(key);
-    if (condition) {
-      if (condition.op === 'eq') {
-        if (_.get(prev, condition.path) !== condition.value) {
-          throw new Error('Update condition not match');
-        }
-      } else if (condition.op === 'exists') {
-        if (!_.get(prev, condition.path)) {
-          throw new Error('Update condition not match');
-        }
-      }
-    }
-    this.state.set(key, value);
-  }
-  createTransaction() {
-    throw new Error('Transaction not supported');
-  }
-}
-
-
 exports.TypeController = TypeController;
-exports.LocalState = LocalState;
+// exports.LocalState = LocalState;

source/stream_service/dist.json

@@ -11,12 +11,13 @@
             "format.proto",
             "service.proto",
             "service.toml",
-            "rtcController.js",
             "streamingController.js",
             "audioController.js",
             "videoController.js",
             "analyticsController.js",
             "domainHandler.js",
+            "stateStores.js",
+            "stateTypes.js",
             "controllers.json",
             "../agent/conference/rpcRequest.js",
             "../common/amqpClient.js",
@@ -30,7 +31,10 @@
             "../../scripts/detectOS.sh"
         ],
         "folders": {
-            "controllers": ["controllers/*"]
+            "controllers": ["controllers/*"],
+            "data_access": [
+                "../data_access/*"
+            ]
         }
     },
     "debug": {

source/stream_service/domainHandler.js

@@ -4,6 +4,7 @@
 
 'use strict';
 
+const {randomUUID} = require('crypto');
 const log = require('./logger').logger.getLogger('DomainHandler');
 const {Publication, Subscription, Processor} = require('./session');
 
@@ -37,7 +38,7 @@ class DomainHandler {
       throw new Error('Duplicate join ID');
     }
     // Build join response
-    const ret = {
+    const response = {
       permission: {},
       room: {
         id: this.domain,
@@ -46,13 +47,17 @@ class DomainHandler {
       },
     };
     for (const [id, ppt] of this.participants) {
-      ret.room.participants.push(ppt);
+      response.room.participants.push(ppt);
     }
     for (const [id, st] of this.streams) {
-      ret.room.streams.push(st.toSignalingFormat());
+      response.room.streams.push(st.toSignalingFormat());
     }
+    req.notifying = true;
     this.participants.set(req.id, req);
-    return ret;
+    return {
+      participant: req,
+      response,
+    };
   }
 
   // Leave from portal
@@ -143,6 +148,11 @@ class DomainHandler {
       }
     }
   }
+  // Add processor request
+  async addProcessor(req) {
+    req.id = req.id || randomUUID().replace(/-/g, '');
+    return req;
+  }
 }
 
 class RemoteDomainHandler {
@@ -154,6 +164,7 @@ class RemoteDomainHandler {
     'streamControl',
     'subscriptionControl',
     'onStatus',
+    'addProcessor',
   ];
 
   constructor(locality, rpcChannel) {

source/stream_service/package.json

@@ -1,11 +1,12 @@
 {
   "name": "OWT-Server-Stream-Service",
-  "version":"5.0.0",
+  "version": "5.0.0",
   "dependencies": {
     "amqplib": "^0.7.0",
     "log4js": "^6.4.0",
     "lodash": "^4.17.11",
     "protobufjs": "^6.11.2",
+    "mongodb": "^4.10.0",
     "toml": "^3.0.0"
   },
   "devDependencies": {