Multithreading implementation

Author: tshemsedinov

JavaScript/3-multithread/actors/mailer.js

@@ -0,0 +1,29 @@
+'use strict';
+
+const ActorSystem = require('../system');
+const nodemailer = require('nodemailer');
+const auth = require('../config');
+
+const FROM = 'nodeua.com@gmail.com';
+
+ActorSystem.register(class Mailer {
+  constructor() {
+    console.log('Start actor: Mailer');
+    this.transport = nodemailer.createTransport({
+      service: 'gmail', auth
+    });
+  }
+
+  message({ to, subject, message }) {
+    const mail = { from: FROM, to, subject, text: message };
+    this.transport.sendMail(mail, (error, data) => {
+      if (error) console.log(error);
+      else console.log(`Email sent: ${data.response}`);
+    });
+  }
+
+  exit() {
+    this.transport.close();
+    console.log('Stop actor: Mailer');
+  }
+});

JavaScript/3-multithread/actors/monitoring.js

@@ -0,0 +1,40 @@
+'use strict';
+
+const ActorSystem = require('../system');
+const http = require('http');
+
+const URL = 'http://localhost:8000/';
+const INTERVAL = 2000;
+
+ActorSystem.register(class Monitoring {
+  constructor() {
+    console.log('Start actor: Monitoring');
+    this.prevSuccess = true;
+    this.timer = setInterval(() => {
+      this.attempt(URL);
+    }, INTERVAL);
+  }
+
+  attempt(url) {
+    http.get(url, res => {
+      const success = res.statusCode === 200;
+      this.notify({ url, success, status: res.statusCode });
+    }).on('error', error => {
+      this.notify({ url, success: false, status: error.message });
+    });
+  }
+
+  notify({ url, success, status }) {
+    if (this.prevSuccess !== success) {
+      this.prevSuccess = success;
+      ActorSystem.send('Renderer', { url, success, status });
+    }
+  }
+
+  message() {}
+
+  exit() {
+    clearInterval(this.timer);
+    console.log('Stop actor: Monitoring');
+  }
+});

JavaScript/3-multithread/actors/renderer.js

@@ -0,0 +1,23 @@
+'use strict';
+
+const ActorSystem = require('../system');
+
+ActorSystem.register(class Renderer {
+  constructor() {
+    console.log('Start actor: Renderer');
+  }
+
+  message({ url, success, status }) {
+    const to = 'nodeua.com@gmail.com';
+    const msg = success ? 'is available again' : 'is not available';
+    const date = new Date().toUTCString();
+    const reason = (success ? 'Status code: ' : 'Error code: ') + status;
+    const message = `Resource ${url} ${msg} (${date})\n${reason}`;
+    const subject = 'Server Monitoring';
+    ActorSystem.send('Mailer', { to, subject, message });
+  }
+
+  exit() {
+    console.log('Stop actor: Renderer');
+  }
+});

JavaScript/3-multithread/actors/root.js

@@ -0,0 +1,21 @@
+'use strict';
+
+const ActorSystem = require('../system');
+
+ActorSystem.register(class Root {
+  constructor() {
+    console.log('Start actor: Root');
+    ActorSystem.start('Monitoring');
+    ActorSystem.start('Renderer');
+    ActorSystem.start('Mailer', 3);
+  }
+
+  message() {}
+
+  exit() {
+    ActorSystem.stop('Monitoring');
+    ActorSystem.stop('Renderer');
+    ActorSystem.stop('Mailer');
+    console.log('Stop actor: Root');
+  }
+});

JavaScript/3-multithread/config.js

@@ -0,0 +1,6 @@
+'use strict';
+
+module.exports = {
+  user: 'account-name@gmail.com',
+  pass: 'your-password',
+};

JavaScript/3-multithread/main.js

@@ -0,0 +1,21 @@
+'use strict';
+
+const ActorSystem = require('./system.js');
+
+const EXIT_NORMAL = 1000;
+const EXIT_ABNORMAL = 5000;
+
+ActorSystem.start('Root');
+
+process.on('SIGINT', () => {
+  console.log('');
+  ActorSystem.stop('Root');
+  setTimeout(() => {
+    console.log('Graceful shutdown');
+    process.exit(0);
+  }, EXIT_NORMAL);
+  setTimeout(() => {
+    console.log('Abnormal termination');
+    process.exit(1);
+  }, EXIT_ABNORMAL);
+});

JavaScript/3-multithread/master.js

@@ -0,0 +1,85 @@
+'use strict';
+
+const threads = require('worker_threads');
+const { Worker } = threads;
+
+const actors = new Map();
+
+class MasterSystem {
+  static start(name, count = 1) {
+    if (!actors.has(name)) {
+      const ready = [];
+      const instances = [];
+      const queue = [];
+      actors.set(name, { ready, instances, queue });
+    }
+    const { ready, instances } = actors.get(name);
+    for (let i = 0; i < count; i++) {
+      const actor = new Worker('./system.js');
+      console.log({ actor });
+      MasterSystem.subscribe(actor);
+      ready.push(actor);
+      instances.push(actor);
+      actor.postMessage({ command: 'start', name });
+    }
+  }
+
+  static stop(name) {
+    const record = actors.get(name);
+    if (record) {
+      const { instances } = record;
+      for (const actor of instances) {
+        actor.postMessage({ command: 'stop' });
+      }
+    }
+  }
+
+  static send(name, data) {
+    const record = actors.get(name);
+    if (record) {
+      const { ready, queue } = record;
+      const actor = ready.shift();
+      if (!actor) {
+        queue.push(data);
+        return;
+      }
+      actor.postMessage({ command: 'message', data });
+    }
+  }
+
+  static subscribe(actor) {
+    actor.on('message', message => {
+      const { command, name } = message;
+      if (command === 'message') {
+        const { data } = message;
+        MasterSystem.send(name, data);
+        return;
+      }
+      if (command === 'start') {
+        const { count } = message;
+        MasterSystem.start(name, count);
+        return;
+      }
+      if (command === 'stop') {
+        MasterSystem.stop(name);
+        return;
+      }
+      if (command === 'ready') {
+        const { id } = message;
+        const record = actors.get(name);
+        if (record) {
+          const { ready, instances, queue } = record;
+          for (const actor of instances) {
+            if (actor.id === id) ready.push(actor);
+          }
+          if (queue.length > 0) {
+            const next = queue.shift();
+            MasterSystem.send(name, next);
+          }
+        }
+      }
+    });
+  }
+}
+
+module.exports = MasterSystem;

JavaScript/3-multithread/package-lock.json

@@ -0,0 +1,19 @@
+{
+  "name": "server-monitoring",
+  "version": "1.0.0",
+  "description": "Actor Model example application",
+  "main": "main.js",
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/HowProgrammingWorks/ActorModel.git"
+  },
+  "author": "Timur Shemsedinov <timur.shemsedinov@gmail.com>",
+  "license": "MIT",
+  "bugs": {
+    "url": "https://github.com/HowProgrammingWorks/ActorModel/issues"
+  },
+  "homepage": "https://github.com/HowProgrammingWorks/ActorModel#readme",
+  "dependencies": {
+    "nodemailer": "^6.1.1"
+  }
+}

JavaScript/3-multithread/package.json

@@ -0,0 +1,6 @@
+'use strict';
+
+const threads = require('worker_threads');
+const { isMainThread } = threads;
+
+module.exports = isMainThread ? require('./master') : require('./worker');

JavaScript/3-multithread/system.js

@@ -0,0 +1,56 @@
+'use strict';
+
+const threads = require('worker_threads');
+const master = threads.parentPort;
+
+class ActorSystem {
+  static register(actor) {
+    ActorSystem.actor = actor;
+  }
+
+  static start(name, count = 1) {
+    master.postMessage({ command: 'start', name, count });
+  }
+
+  static stop(name) {
+    master.postMessage({ command: 'stop', name });
+  }
+
+  static send(name, data) {
+    master.postMessage({ command: 'message', name, data });
+  }
+}
+
+ActorSystem.actor = null;
+ActorSystem.instance = null;
+
+//process.on('SIGINT', () => {});
+
+master.on('message', message => {
+  const { command } = message;
+  if (command === 'start') {
+    const { name } = message;
+    require(`./actors/${name.toLowerCase()}.js`);
+    const ActorClass = ActorSystem.actor;
+    ActorSystem.instance = new ActorClass();
+    return;
+  }
+  if (command === 'stop') {
+    const { instance } = ActorSystem;
+    if (instance) instance.exit();
+    process.exit(0);
+    return;
+  }
+  if (command === 'message') {
+    const { instance } = ActorSystem;
+    if (instance) {
+      const { data } = message;
+      const { name } = ActorSystem.actor;
+      instance.message(data);
+      const msg = { command: 'ready', name, id: threads.threadId };
+      master.postMessage(msg);
+    }
+  }
+});
+
+module.exports = ActorSystem;