communicating embedded worker factories

communicating-embedded-worker-factories.js

// 25 june 2022
// communicating embedded worker factories

// 29 june 2022
// - api finally works
// - fix Self parsing function (concat(";\n"))

// 4 July 2022
// - added "X forwarding to Y" example:
// - anti-pattern: client creates X and Y, attaches Y.send in an X.receive callback;
// - pattern: X should create, monitor, and close Y, and any other child workers.

// Yet one more step in the quest for making embedded workers testable
// from the console.

// Demonstrates how to supply one or more worker script functions as capabilities
// to a factory that knits the scripts together.

// Once we get this working, we'll move on (in no particular order) to:
//  - testing worker functions independently ~ prior to parsing and embedding
//      https://gist.github.com/dfkaye/4f368f446cb1efc06fb0c764f0e8a4f6
//  - message queueing
//      [ pause and resume gist ]
//      https://gist.github.com/dfkaye/ddb4a391c2f6a58bc8f039bbda3cb7ee
//  - creating/monitoring with the actor model
//      [ TODO...]
//  - crash and restart
//      [...very barebones...]
//      https://gist.github.com/dfkaye/a19ba42ef9172a032b7cc104087b8fc2
//  - workers as microservices
//      [ TODO... re Nygard's post on state services vs. entity services]
//  - chunked responses or streaming (sort of)
//      [...much to add...]
//      https://gist.github.com/dfkaye/458210c93b0b9ffc5de9a9a2f1257dcd

// ...continued from "parse body of a function given constraints"
// at https://gist.github.com/dfkaye/c50ff265f4411145151fff2f44c2e1bf

// We'll use Self to parse body of a function with `function (self) {}`
// or `(self) => { }` format.

function Self(fn) {
  var fs = String(fn).trim();

  var parseable = typeof fn == "function"
    && fn.length === 1
    && fs.match(/(self)[^\)]*\)[^\{]*\{/);

  if (!parseable) {
    return;
  }

  return fs /* .substring(fs.indexOf("self") + 4) */
    .replace(/(self)?[^\)]*\)[^\{]*\{/, '')
    .replace(/\s*\}\s*$/, '')
    .split("\n")
    .filter(s => /\S/.test(s))
    .join("\n")
    .concat(";\n");
}

// Accept any number of functions restricted to the Self format, parse the
// body of each to an array passed to a blob, create the blob url, create
// the worker, and return an api for a client to communicate
// with the worker, including a "close" (i.e., terminate) message.

function Factory(...fns) {
  if (!(this instanceof Factory)) {
    return new Factory(...fns);
  }

  if (!fns.length) {
    throw new Error("Factory requires at least one function argument.");
  }

  var scripts = [];

  fns.forEach(function (fn) {
    var scriptBody = Self(fn);

    if (!scriptBody) {
      throw new Error("Factory could not parse script body with Self");
    }

    scripts.push(scriptBody);
  });

  var blob = new Blob(scripts, { type: "text/javascript" });
  var address = URL.createObjectURL( blob );
  var worker = new Worker( address );
  var sender = URL.createObjectURL( new Blob([this]) );

  // sender
  // address
  // action
  // value

  return {
    close(message) {
      console.log("close", sender, address, message);

      worker.postMessage({ action: "close", value: message });
      //worker.terminate();
    },
    send(message) {
      console.log("send", sender, address, message);

      Object.assign(message, { sender, address });

      worker.postMessage(message);
    },
    receive(fn) {
      if (typeof fn != "function") {
        return new TypeError("invalid receive function");
      }

      console.warn("assigning receive fn", sender, address, fn);

      worker.addEventListener("message", fn);
    }
  }
}

/* test it out */

// Each function to be added represents a capability or behavior.

var onmessage = function (self) {
  self.onmessage = function (request) {
    var data = Object(request.data);

    console.log("onmessage:", data);

    var response = Object.assign({}, data);

    self.postMessage(response);
  };
}

var test = function (self) {
  // The test IIFE exercises the onmessage method directly.
  (function (data) {

    var sender = "sender";

    // 29 June 2022: Great finding!
    // Use the URL from the location, not the location object,
    // which doesn't pass the structured cloning algorithm!
    var address = self.location.toString();

    Object.assign(data, { sender, address });

    self.onmessage({ data });

  })({ action: "test", value: "init" });
}

var x = Factory(onmessage, test);

x.send("bad request");
x.receive("bad response");
x.send({ action: "log", value: "test log x" });
x.receive(function (response) {
  var data = Object(response.data);
  var name = "X receiving " + Date.now();

  console.group(name);
    Object.entries(data).forEach(e => {
      var [k, v] = e;
      console.log(k, ":", v);
    });
  console.groupEnd(name);
});
x.send({ action: "receive", value: "test receive from x" });


/*
  4 July 2022
  Now let's define another one (y) and have x and y talk to each other.
*/

var y = Factory(onmessage, test);

y.send({ action: "log", value: "Hello from Y" });
y.receive(function (response) {
  var data = Object(response.data);
  var name = "Y receiving " + Date.now();

  console.group(name);
    Object.entries(data).forEach(e => {
      var [k, v] = e;
      console.log(k, ":", v);
    });
  console.groupEnd(name);
});
y.send({ action: "receive", value: "test receive from Y" });

// 4 July 2022
// We'll need to de-couple y from x in future tests.
// X really should be the actor that creates Y using the Factory, then
// sends messages to Y, listens to Y, closes Y, etc.

x.receive(function(response) {
  var data = Object.assign({}, response.data);

  data.sender = "x";
  data.address = "y";
  data.subject = "forwarding test";

  y.send(data);
});

x.send({ action: "forward", value: "from x to y, maybe" });

/*
send blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991 bad request
send blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991 Object { action: "log", value: "test log x" }
receive blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991 function (response)
send blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991 Object { action: "receive", value: "test receive from x" }
send blob:null/e78f01d3-6fae-462e-99ff-b42b62732347 blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d Object { action: "log", value: "Hello from Y" }
receive blob:null/e78f01d3-6fae-462e-99ff-b42b62732347 blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d function (response)
send blob:null/e78f01d3-6fae-462e-99ff-b42b62732347 blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d Object { action: "receive", value: "test receive from Y" }
receive blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991 function (response)
send blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991 Object { action: "forward", value: "from x to y, maybe" }
undefined
onmessage: Object { action: "test", value: "init", sender: "sender", address: "blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991" }
onmessage: String { "bad request" }
onmessage: Object { action: "test", value: "init", sender: "sender", address: "blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d" }
onmessage: Object { action: "log", value: "test log x", sender: "blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd", address: "blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991" }
onmessage: Object { action: "log", value: "Hello from Y", sender: "blob:null/e78f01d3-6fae-462e-99ff-b42b62732347", address: "blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d" }
onmessage: Object { action: "receive", value: "test receive from x", sender: "blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd", address: "blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991" }
onmessage: Object { action: "receive", value: "test receive from Y", sender: "blob:null/e78f01d3-6fae-462e-99ff-b42b62732347", address: "blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d" }
onmessage: Object { action: "forward", value: "from x to y, maybe", sender: "blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd", address: "blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991" }
X receiving 1657082852347
  action : test
  value : init
  sender : sender
  address : blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991
send blob:null/e78f01d3-6fae-462e-99ff-b42b62732347 blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d Object { action: "test", value: "init", sender: "x", address: "y", subject: "forwarding test" }
onmessage: Object { action: "test", value: "init", sender: "blob:null/e78f01d3-6fae-462e-99ff-b42b62732347", address: "blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d", subject: "forwarding test" }
Y receiving 1657082852350
  action : test
  value : init
  sender : sender
  address : blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d
X receiving 1657082852351
  0 : b
  1 : a
  2 : d
  3 :
  4 : r
  5 : e
  6 : q
  7 : u
  8 : e
  9 : s
  10 : t
send blob:null/e78f01d3-6fae-462e-99ff-b42b62732347 blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d Object { 0: "b", 1: "a", 2: "d", 3: " ", 4: "r", 5: "e", 6: "q", 7: "u", 8: "e", 9: "s", … }
onmessage: Object { 0: "b", 1: "a", 2: "d", 3: " ", 4: "r", 5: "e", 6: "q", 7: "u", 8: "e", 9: "s", … }
Y receiving 1657082852355
  action : log
  value : Hello from Y
  sender : blob:null/e78f01d3-6fae-462e-99ff-b42b62732347
  address : blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d
X receiving 1657082852356
  action : log
  value : test log x
  sender : blob:null/39b29b45-e4fc-4d28-96ea-fe3cbc6ce7fd
  address : blob:null/b9d7fd81-cddf-4f98-8aac-e891d40a4991
send blob:null/e78f01d3-6fae-462e-99ff-b42b62732347 blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d Object { action: "log", value: "test log x", sender: "x", address: "y", subject: "forwarding test" }
onmessage: Object { action: "log", value: "test log x", sender: "blob:null/e78f01d3-6fae-462e-99ff-b42b62732347", address: "blob:null/893fabab-1bb4-4ea9-96d7-c1e31a62331d", subject: "forwarding test" }
 */