chore: working on chapter 13

Author: lmammino

13-messaging-and-integration-patterns/09-correlation-id/README.md

@@ -0,0 +1,37 @@
+# 09-correlation-id
+
+This sample demonstrates how to implement a request/reply pattern on top of a
+simple point-to-point one-way asynchronous channel.
+
+## Dependencies
+
+This example requires you to install some third-party dependencies from npm.
+
+If you have `pnpm` installed, you can do that with:
+
+```bash
+pnpm install
+```
+
+Alternatively, if you prefer to use another package manager, make sure to delete
+the `pnpm-lock.yaml` file before using it.
+
+If you want to use `npm`, you can run:
+
+```bash
+npm install
+```
+
+If you want to use `yarn`, you can run:
+
+```bash
+yarn install
+```
+
+## Run
+
+To run this example, use the following command:
+
+```bash
+node requestor.js
+```

13-messaging-and-integration-patterns/09-correlation-id/createReplyChannel.js

@@ -0,0 +1,16 @@
+export function createReplyChannel(channel) {
+  return function registerHandler(handler) {
+    channel.on('message', async message => {
+      if (message.type !== 'request') {
+        return
+      }
+
+      const replyData = await handler(message.data)
+      channel.send({
+        type: 'response',
+        data: replyData,
+        inReplyTo: message.id,
+      })
+    })
+  }
+}

13-messaging-and-integration-patterns/09-correlation-id/createRequestChannel.js

@@ -0,0 +1,38 @@
+import { nanoid } from 'nanoid' // v5.1.5
+
+export function createRequestChannel(channel) {
+  const correlationMap = new Map()
+
+  function sendRequest(data) {
+    console.log('Sending request', data)
+    return new Promise((resolve, reject) => {
+      const correlationId = nanoid()
+
+      const replyTimeout = setTimeout(() => {
+        correlationMap.delete(correlationId)
+        reject(new Error('Request timeout'))
+      }, 10000)
+
+      correlationMap.set(correlationId, replyData => {
+        correlationMap.delete(correlationId)
+        clearTimeout(replyTimeout)
+        resolve(replyData)
+      })
+
+      channel.send({
+        type: 'request',
+        data,
+        id: correlationId,
+      })
+    })
+  }
+
+  channel.on('message', message => {
+    const replyCb = correlationMap.get(message.inReplyTo)
+    if (replyCb) {
+      replyCb(message.data)
+    }
+  })
+
+  return sendRequest
+}

13-messaging-and-integration-patterns/09-correlation-id/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "09-correlation-id",
+  "version": "1.0.0",
+  "description": "This sample demonstrates how to implement a request/reply pattern on top of a simple point-to-point one-way asynchronous channel.",
+  "type": "module",
+  "scripts": {},
+  "engines": {
+    "node": ">=24"
+  },
+  "engineStrict": true,
+  "keywords": [],
+  "author": "Luciano Mammino and Mario Casciaro",
+  "license": "MIT",
+  "dependencies": {
+    "nanoid": "^5.1.5"
+  }
+}

13-messaging-and-integration-patterns/09-correlation-id/pnpm-lock.yaml

@@ -0,0 +1,13 @@
+import { createReplyChannel } from './createReplyChannel.js'
+
+const registerReplyHandler = createReplyChannel(process)
+
+registerReplyHandler(req => {
+  return new Promise(resolve => {
+    setTimeout(() => {
+      resolve({ sum: req.a + req.b })
+    }, req.delay)
+  })
+})
+
+process.send('ready')

13-messaging-and-integration-patterns/09-correlation-id/replier.js

@@ -0,0 +1,23 @@
+import { fork } from 'node:child_process'
+import { once } from 'node:events'
+import { join } from 'node:path'
+import { createRequestChannel } from './createRequestChannel.js'
+
+const channel = fork(join(import.meta.dirname, 'replier.js'))
+const request = createRequestChannel(channel)
+
+try {
+  const [message] = await once(channel, 'message')
+  console.log(`Child process initialized: ${message}`)
+  const p1 = request({ a: 1, b: 2, delay: 500 }).then(res => {
+    console.log(`Reply: 1 + 2 = ${res.sum}`)
+  })
+
+  const p2 = request({ a: 6, b: 1, delay: 100 }).then(res => {
+    console.log(`Reply: 6 + 1 = ${res.sum}`)
+  })
+
+  await Promise.all([p1, p2])
+} finally {
+  channel.disconnect()
+}