Add support for worker threads

Add support for passing request to a worker thread using a simple
wrapper function.

```
micri(handler);
```

becomes

```
micri(withWorker(handler));
```

.github/workflows/nodejs.yml

@@ -9,7 +9,7 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [10.x, 12.x, 13.x]
+        node-version: [12.x, 13.x]
 
     steps:
     - uses: actions/checkout@v1

README.md

@@ -11,7 +11,7 @@ _**Micri** — Asynchronous HTTP microservices_
 
 * **Easy**: Designed for usage with `async` and `await` ([more](https://zeit.co/blog/async-and-await))
 * **Fast**: Ultra-high performance (even JSON parsing is opt-in)
-* **Micri**: The whole project is ~300 lines of code
+* **Micri**: The whole project is ~500 lines of code
 * **Agile**: Super easy deployment and containerization
 * **Simple**: Oriented for single purpose modules (function)
 * **Standard**: Just HTTP!
@@ -22,11 +22,11 @@ _**Micri** — Asynchronous HTTP microservices_
 ## Usage
 
 ```js
-const micri = require('micri')
+const { serve } = require('micri')
 
 const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
 
-const server = micri(async (req, res) => {
+const server = serve(async (req, res) => {
   await sleep(500)
   return 'Hello world'
 })
@@ -99,6 +99,10 @@ function may return a truthy value if the handler function should take care of
 this request, or it may return a falsy value if the handler should not take
 this request.
 
+Multiple predicates can be combined by using `Router.everyPredicate(...)` that
+takes predicate functions as arguments. The function returns true if every
+predicate function given as an argument returns true.
+
 The order of the route arguments marks the priority order of the routes.
 Therefore if two routes would match to a request the one that was passed earlier
 in the arguments list to the `router()` function will handle the request.
@@ -117,6 +121,48 @@ micri(router(
 	.listen(3000);
 ```
 
+
+### Worker Threads
+
+Micri supports offloading computationally heavy request handlers to worker
+threads seamlessly. The offloading is configured per handler by wrapping the
+handler function with `withWorker()`. It works directly at the top-level or per
+route when using the router. See [worker-threads](examples/worker-threads) for
+a couple of examples how to use it.
+
+```js
+micri(withWorker(() => doSomethingCPUHeavy))
+```
+
+Offloading requests to a worker may improve the responsiveness of a busy API
+significantly, as it removes almost all blocking from the main thread. In the
+following examples we first try to find prime numbers and finally return one
+as a response. In both cases we do two concurrent HTTP `GET` requests using
+`curl`.
+
+Finding prime numbers using the main thread:
+
+```
+~% time curl 127.0.0.1:3000/main
+299993curl 127.0.0.1:3000/main  0.01s user 0.00s system 0% cpu 8.791 total
+~% time curl 127.0.0.1:3000/main
+299993curl 127.0.0.1:3000/main  0.00s user 0.00s system 0% cpu 16.547 total
+```
+
+Notice that the second curl needs to wait until the first request finishes.
+
+Finding prime numbers using a worker thread:
+
+```
+~% time curl 127.0.0.1:3000/worker
+299993curl 127.0.0.1:3000/worker  0.00s user 0.00s system 0% cpu 9.025 total
+~% time curl 127.0.0.1:3000/worker
+299993curl 127.0.0.1:3000/worker  0.00s user 0.00s system 0% cpu 9.026 total
+```
+
+Note how both concurrently executed requests took the same time to finish.
+
+
 ## API
 
 ##### `buffer(req, { limit = '1mb', encoding = 'utf8' })`

examples/worker-threads/.gitignore

@@ -0,0 +1,3 @@
+!node_modules
+/micri
+/lib

examples/worker-threads/README.md

@@ -0,0 +1,22 @@
+Worker Threads
+==============
+
+Two concurrent requests running a CPU intensive request method in the
+main thread:
+
+```
+~% time curl 127.0.0.1:3000/main
+299993curl 127.0.0.1:3000/main  0.01s user 0.00s system 0% cpu 8.791 total
+~% time curl 127.0.0.1:3000/main
+299993curl 127.0.0.1:3000/main  0.00s user 0.00s system 0% cpu 16.547 total
+```
+
+Two concurrent requests running a CPU intensive request method in worker
+threads:
+
+```
+~% time curl 127.0.0.1:3000/worker
+299993curl 127.0.0.1:3000/worker  0.00s user 0.00s system 0% cpu 9.025 total
+~% time curl 127.0.0.1:3000/worker
+299993curl 127.0.0.1:3000/worker  0.00s user 0.00s system 0% cpu 9.026 total
+```

examples/worker-threads/package.json

@@ -0,0 +1,13 @@
+{
+  "name": "router",
+  "description": "Router example",
+  "main": "lib/index.js",
+  "scripts": {
+    "build": "rm -rf ./lib ./micri && mkdir ./micri && cp ../../dist/* ./micri && tsc",
+    "start": "node lib/"
+  },
+  "devDependencies": {
+    "@types/node": "12.7.11",
+    "typescript": "3.6.3"
+  }
+}

examples/worker-threads/src/index.ts

@@ -0,0 +1,21 @@
+import path from 'path'
+import { parse } from 'url';
+import micri, {
+	IncomingMessage,
+	ServerResponse,
+	Router,
+	send,
+	withWorker
+} from 'micri';
+import prime  from './prime';
+
+const { router, on, otherwise } = Router;
+
+const parsePath = (req: IncomingMessage): string => parse(req.url || '/').path || '/';
+
+micri(router(
+	on.get((req: IncomingMessage) => parsePath(req) === '/main', prime),
+	on.get((req: IncomingMessage) => parsePath(req) === '/worker', withWorker(path.join(__dirname, './prime.js'))),
+	on.get((req: IncomingMessage) => parsePath(req) === '/stream', withWorker(path.join(__dirname, './stream.js'))),
+	otherwise((_req: IncomingMessage, res: ServerResponse) => send(res, 400, 'Method Not Accepted'))))
+	.listen(3000);

examples/worker-threads/src/prime.ts

@@ -0,0 +1,19 @@
+export default function fn() {
+	const limit = 3e5;
+	let r;
+
+	for (let n = 2; n <= limit; n++) {
+		let isPrime = true;
+		for(let factor = 2; factor < n; factor++) {
+			if(n % factor == 0) {
+				isPrime = false;
+				break;
+			}
+		}
+		if(isPrime) {
+			r = n;
+		}
+	}
+
+	return r;
+}

examples/worker-threads/src/stream.ts

@@ -0,0 +1,23 @@
+import { IncomingMessage, ServerResponse } from "http";
+
+function getNumber() {
+	return new Promise((resolve) => {
+		setTimeout(() => {
+			const x = Math.round(Math.random() * 100);
+
+			resolve(x);
+		}, 300);
+	});
+}
+
+export default async function wrkStream(_req: IncomingMessage, res: ServerResponse) {
+	res.writeHead(200, {
+		'X-Custom-Header': "I'm a header"
+	});
+
+	for (let i = 0; i < 30; i++) {
+		const x = await getNumber();
+		res.write(`${x}\n`);
+	}
+	res.end();
+}

examples/worker-threads/tsconfig.json

@@ -0,0 +1,23 @@
+{
+  "compilerOptions": {
+    "declaration": true,
+    "esModuleInterop": true,
+    "lib": ["esnext"],
+    "module": "CommonJS",
+    "moduleResolution": "node",
+    "resolveJsonModule": true,
+    "strictNullChecks": true,
+    "noImplicitAny": true,
+    "noUnusedLocals": true,
+    "noUnusedParameters": true,
+    "outDir": "./lib",
+    "strict": true,
+    "target": "ES2018",
+    "inlineSourceMap": true,
+    "types": ["node"],
+    "typeRoots": [
+      "./node_modules/@types"
+    ]
+  },
+  "include": ["./src"]
+}

examples/worker-threads/yarn.lock

@@ -0,0 +1,13 @@
+# THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY.
+# yarn lockfile v1
+
+
+"@types/[email protected]":
+  version "12.7.11"
+  resolved "https://registry.yarnpkg.com/@types/node/-/node-12.7.11.tgz#be879b52031cfb5d295b047f5462d8ef1a716446"
+  integrity sha512-Otxmr2rrZLKRYIybtdG/sgeO+tHY20GxeDjcGmUnmmlCWyEnv2a2x1ZXBo3BTec4OiTXMQCiazB8NMBf0iRlFw==
+
+[email protected]:
+  version "3.6.3"
+  resolved "https://registry.yarnpkg.com/typescript/-/typescript-3.6.3.tgz#fea942fabb20f7e1ca7164ff626f1a9f3f70b4da"
+  integrity sha512-N7bceJL1CtRQ2RiG0AQME13ksR7DiuQh/QehubYcghzv20tnh+MQnQIuJddTmsbqYj+dztchykemz0zFzlvdQw==

lib/worker-wrapper.js

@@ -0,0 +1,54 @@
+const { serve } = require('./index');
+const { request } = require('http');
+const { parentPort, workerData } = require('worker_threads');
+
+module.exports = async function wrap(fn) {
+	try {
+		const buf = await new Promise((resolve) => {
+			parentPort.on('message', (body) => resolve(body));
+		});
+
+		const server = serve(fn);
+		server.listen(() => {
+			const port = server.address().port;
+			const intReq = request(
+				`http://127.0.0.1${workerData.req.url}`,
+				{
+					hostname: '127.0.0.1',
+					port,
+					method: workerData.req.method,
+					headers: {
+						...workerData.req.headers,
+						'Content-Length': buf.length,
+					},
+				},
+				(res) => {
+					const head = {
+						statusCode: res.statusCode || 200,
+						statusMessage: res.statusMessage,
+						headers: res.headers,
+					};
+
+					parentPort.postMessage(head);
+
+					res.on('data', (chunk) => {
+						parentPort.postMessage(chunk);
+					});
+					res.on('end', () => {
+						process.exit(0);
+					});
+				}
+			);
+			intReq.on('error', (err) => {
+				console.error(err);
+				process.exit(1);
+			});
+
+			intReq.write(Buffer.from(buf));
+			intReq.end();
+		});
+	} catch (err) {
+		console.error(err);
+		process.exit(1);
+	}
+};

package.json

@@ -7,15 +7,15 @@
     "dist"
   ],
   "scripts": {
-    "build": "rm -rf ./dist && ncc build -m ./src/index.ts",
+    "build": "rm -rf ./dist && ncc build -m -s ./src/index.ts && cp ./lib/worker-wrapper.js ./dist/",
     "test": "npm run lint && NODE_ENV=test jest",
     "lint": "eslint --ext .ts ./src",
     "git-pre-commit": "eslint --ext .ts ./src",
     "prepublish": "npm run build",
-    "prettier": "prettier --write './{src,test}/**/*.ts'"
+    "prettier": "prettier --write './{src,test}/**/*.ts' && prettier --write './lib/*.js'"
   },
   "engines": {
-    "node": ">= 8.0.0"
+    "node": ">= 12.0.0"
   },
   "repository": "OlliV/micri",
   "keywords": [

src/index.ts

@@ -1,10 +1,12 @@
 // Utilities
 import { serve } from './serve';
 import * as Router from './router';
+import withWorker from './with-worker';
 
 export default serve;
 export * from './body';
 export * from './serve';
 export * from './types';
 export { Router };
+export { withWorker };
 export { MicriError } from './errors';