Started writing tests for supervision

lib/actor.js

@@ -6,16 +6,13 @@ const Queue = require('denque');
 const assert = require('assert');
 const freeze = require('deep-freeze-node');
 const { Subject } = require('rxjs');
-const { stop } = require('./functions');
-
-const defaultSupervisionPolicy = (child, supervisionCtx) =>
-  stop(child);
+const { stop, defaultSupervisionPolicy } = require('./functions');
 
 class Actor {
   constructor (parent, name, system, f, { shutdownAfter, whenChildCrashes } = {}) {
     this.parent = parent;
     if (!name) {
-      name = `anonymous-${parent.children.size}`;
+      name = `anonymous-${(Math.random() * Number.MAX_SAFE_INTEGER) | 0}`;
     }
     if (name && parent.children.has(name)) {
       throw new Error(`child actor of name ${name} already exists`);
@@ -35,6 +32,7 @@ class Actor {
     this.immediate = undefined;
     this.parent.childSpawned(this);
     this.whenChildCrashes = whenChildCrashes || defaultSupervisionPolicy;
+
     if (shutdownAfter) {
       if (typeof (shutdownAfter) !== 'number') {
         throw new Error('Shutdown should be specified as a number in milliseconds');
@@ -44,16 +42,17 @@ class Actor {
     }
   }
 
-  static serializeErr (err) {
-    return JSON.stringify(err, Object.getOwnPropertyNames(err));
-  }
-
   setTimeout () {
     if (this.shutdownPeriod) {
       this.timeout = setTimeout(() => this.stop(), this.shutdownPeriod);
     }
   }
 
+  reset () {
+    this.state = undefined;
+    [...this.children.entries()].forEach(x => x.reset());
+  }
+
   clearTimeout () {
     clearTimeout(this.timeout);
   }
@@ -120,6 +119,7 @@ class Actor {
   }
 
   get state$ () {
+    console.warn('nact deprecation notice: state$ is deprecated');
     return this.subject.asObservable();
   }
 
@@ -144,10 +144,28 @@ class Actor {
     }
   }
 
-  signalFault (error) {
-    const serializedErr = Actor.serializeErr(error);
-    console.error(serializedErr);
-    this.stop();
+  signalFault (msg, sender, error) {
+    if (this.parent.createSupervisionContext) {
+      const ctx = this.parent.createSupervisionContext(this, msg, sender, error);
+      this.parent.whenChildCrashes(this.reference, msg, error, ctx);
+    } else {
+      this.stop();
+    }
+  }
+
+  resume () {
+    this.processNext(this.state);
+  }
+
+  createSupervisionContext (child, msg, sender, error) {
+    const ctx = this.createContext(sender);
+    const stop = () => child.stop();
+    const resume = () => child.resume();
+    const stopAll = () => { [...this.children.entries()].forEach(x => x.stop()); };
+    const reset = () => child.reset();
+    const resetAll = () => { [...this.children.entries()].forEach(x => x.reset()); };
+    const escalate = () => this.signalFault(msg, sender, error);
+    return { ...ctx, stop, stopAll, reset, resetAll, escalate, resume };
   }
 
   createContext (sender) {
@@ -168,12 +186,12 @@ class Actor {
         let ctx = this.createContext(sender);
         let next = this.f.call(ctx, freeze(this.state), message, ctx);
         if (next && next.then && next.catch) {
-          next.then(result => this.processNext(result)).catch(err => this.signalFault(err));
+          next.then(result => this.processNext(result)).catch(err => this.signalFault(message, sender, err));
         } else {
           this.processNext(next);
         }
       } catch (e) {
-        this.signalFault(e);
+        this.signalFault(message, sender, e);
       }
     });
   }

lib/functions.js

@@ -1,8 +1,20 @@
+const serialize = (err) =>
+  JSON.stringify(err, Object.getOwnPropertyNames(err));
 
 const stop = (actor) => {
   return actor.stop();
 };
 
+const defaultSupervisionPolicy = (child, msg, err, ctx) => {
+  let serializedMsg = msg;
+  try { serializedMsg = JSON.stringify(msg); } catch (e) {}
+  let serializedErr = err;
+  try { serializedErr = serialize(err); } catch (e) {}
+  let path = child.path.toString();
+  console.error(`${path}: The following error was raised when processing ${serializedMsg}:\n ${serializedErr} \nTerminating faulted actor`);
+  ctx.stop();
+};
+
 const query = (actor, msg, timeout, ...args) => {
   if (!timeout) {
     throw new Error('A timeout is required to be specified');
@@ -22,5 +34,6 @@ module.exports = {
   stop,
   query,
   dispatch,
-  state$
+  state$,
+  defaultSupervisionPolicy
 };

lib/paths.js

@@ -23,6 +23,10 @@ class LocalPath {
   static root () {
     return new LocalPath([]);
   }
+
+  toString () {
+    return this.localParts.join('/');
+  }
 }
 
 module.exports = { LocalPath };

lib/references.js

@@ -56,6 +56,10 @@ class ActorReference {
     return actorOrFallback(this).stop();
   }
 
+  reset () {
+    return actorOrFallback(this).reset();
+  }
+
   get state$ () {
     return actorOrFallback(this).state$;
   }

lib/system.js

@@ -1,7 +1,7 @@
 const { ActorSystemReference } = require('./references');
 const { LocalPath } = require('./paths');
 const assert = require('assert');
-const { stop } = require('./functions');
+const { stop, defaultSupervisionPolicy } = require('./functions');
 
 class ActorSystem {
   constructor (extensions) {
@@ -11,6 +11,7 @@ class ActorSystem {
     this.path = LocalPath.root();
     this.stopped = false;
     this.system = this.reference;
+    this.whenChildCrashes = defaultSupervisionPolicy;
     assert(extensions instanceof Array);
     extensions.forEach(extension => extension(this));
   }
@@ -29,6 +30,7 @@ class ActorSystem {
     [...this.children.values()].map(stop);
     this.stopped = true;
   }
+  reset () {}
 
   assertNotStopped () { assert(!this.stopped); return true; }
 }

package.json

@@ -5,6 +5,7 @@
   "main": "lib/index.js",
   "scripts": {
     "test": "yarn run lint && nyc mocha ./test/*.js",
+    "test-watch": "yarn run lint && nyc mocha --watch ./test/*.js",
     "interactive-cover": "nyc --reporter=html mocha ./test/*.js && open ./coverage/index.html",
     "lint": "semistandard",
     "coverage": "nyc report --reporter=text-lcov | coveralls",

test/actor.js

@@ -395,6 +395,22 @@ describe('Actor', function () {
     });
   });
 
+  describe('#whenChildCrashes', function () {
+    let system;
+    beforeEach(() => { system = start(); });
+    afterEach(() => stop(system));
+
+    const createSupervisor = (parent, name, whenChildCrashes) => spawn(parent, (state = true, msg, ctx) => state, name, { whenChildCrashes });
+
+    it('should be able to continue processing messages without loss of state', async function () {
+
+    });
+
+    it('should be able to be restarted', async function () {
+
+    });
+  });
+
   describe('#state$', function () {
     let system;
     beforeEach(() => { system = start(); });