chore: working on chapter 13

Author: lmammino

13-messaging-and-integration-patterns/10-return-address/README.md

@@ -0,0 +1,49 @@
+# 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
+
+As pre-requisite to this sample, you first need to
+[install RabbitMQ](http://www.rabbitmq.com/download.html)
+
+If you have docker installed, you can run an ephemeral instance of RabbitMQ with
+the following command:
+
+```bash
+docker run -it -p 5672:5672 --hostname my-rabbit rabbitmq:4
+```
+
+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 commands (in different terminals):
+
+```bash
+node replier.js
+node requestor.js
+node requestor.js # you can run as many requestors as you want
+```

13-messaging-and-integration-patterns/10-return-address/amqpReply.js

@@ -0,0 +1,27 @@
+import amqp from 'amqplib' // v0.10.8
+
+export class AmqpReply {
+  constructor(requestsQueueName) {
+    this.requestsQueueName = requestsQueueName
+  }
+
+  async initialize() {
+    const connection = await amqp.connect('amqp://localhost')
+    this.channel = await connection.createChannel()
+    const { queue } = await this.channel.assertQueue(this.requestsQueueName)
+    this.queue = queue
+  }
+
+  handleRequests(handler) {
+    this.channel.consume(this.queue, async msg => {
+      const content = JSON.parse(msg.content.toString())
+      const replyData = await handler(content)
+      this.channel.sendToQueue(
+        msg.properties.replyTo,
+        Buffer.from(JSON.stringify(replyData)),
+        { correlationId: msg.properties.correlationId }
+      )
+      this.channel.ack(msg)
+    })
+  }
+}

13-messaging-and-integration-patterns/10-return-address/amqpRequest.js

@@ -0,0 +1,53 @@
+import amqp from 'amqplib' // v0.10.8
+import { nanoid } from 'nanoid' // v5.1.5
+
+export class AmqpRequest {
+  constructor() {
+    this.correlationMap = new Map()
+  }
+
+  async initialize() {
+    this.connection = await amqp.connect('amqp://localhost')
+    this.channel = await this.connection.createChannel()
+    const { queue } = await this.channel.assertQueue('', { exclusive: true })
+    this.replyQueue = queue
+
+    this.channel.consume(
+      this.replyQueue,
+      msg => {
+        const correlationId = msg.properties.correlationId
+        const handler = this.correlationMap.get(correlationId)
+        if (handler) {
+          handler(JSON.parse(msg.content.toString()))
+        }
+      },
+      { noAck: true }
+    )
+  }
+
+  send(queue, message) {
+    return new Promise((resolve, reject) => {
+      const id = nanoid()
+      const replyTimeout = setTimeout(() => {
+        this.correlationMap.delete(id)
+        reject(new Error('Request timeout'))
+      }, 10000)
+
+      this.correlationMap.set(id, replyData => {
+        this.correlationMap.delete(id)
+        clearTimeout(replyTimeout)
+        resolve(replyData)
+      })
+
+      this.channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)), {
+        correlationId: id,
+        replyTo: this.replyQueue,
+      })
+    })
+  }
+
+  destroy() {
+    this.channel.close()
+    this.connection.close()
+  }
+}

13-messaging-and-integration-patterns/10-return-address/package.json

@@ -0,0 +1,18 @@
+{
+  "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",
+    "amqplib": "^0.10.8"
+  }
+}

13-messaging-and-integration-patterns/10-return-address/pnpm-lock.yaml

@@ -0,0 +1,9 @@
+import { AmqpReply } from './amqpReply.js'
+
+const reply = new AmqpReply('requests_queue')
+await reply.initialize()
+
+reply.handleRequests(req => {
+  console.log('Request received', req)
+  return { sum: req.a + req.b }
+})

13-messaging-and-integration-patterns/10-return-address/replier.js

@@ -0,0 +1,19 @@
+import { setTimeout } from 'node:timers/promises'
+import { AmqpRequest } from './amqpRequest.js'
+
+const request = new AmqpRequest()
+await request.initialize()
+
+async function sendRandomRequest() {
+  const a = Math.round(Math.random() * 100)
+  const b = Math.round(Math.random() * 100)
+  const reply = await request.send('requests_queue', { a, b })
+  console.log(`${a} + ${b} = ${reply.sum}`)
+}
+
+for (let i = 0; i < 20; i++) {
+  await sendRandomRequest()
+  await setTimeout(1000)
+}
+
+request.destroy()