chore: working on chapter 13

Author: lmammino

13-messaging-and-integration-patterns/04-pubsub-amqp/README.md

@@ -12,7 +12,7 @@ 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:3
+docker run -it -p 5672:5672 --hostname my-rabbit rabbitmq:4
 ```
 
 This example requires you to install some third-party dependencies from npm.

13-messaging-and-integration-patterns/06-task-distribution-zmq/README.md

@@ -1,4 +1,4 @@
-# 05-pubsub-redis-streams
+# 06-task-distribution-zmq
 
 This sample demonstrates how to run parallel tasks using a messaging pipeline
 and ZeroMQ

13-messaging-and-integration-patterns/06-task-distribution-zmq/package.json

@@ -1,5 +1,5 @@
 {
-  "name": "05-pubsub-redis-streams",
+  "name": "06-task-distribution-zmq",
   "version": "1.0.0",
   "description": "This sample demonstrates how to run parallel tasks using a messaging pipeline and ZeroMQ",
   "type": "module",

13-messaging-and-integration-patterns/07-task-distribution-amqp/README.md

@@ -0,0 +1,59 @@
+# 07-task-distribution-amqp
+
+This sample demonstrates how to distribute tasks to a set of remote workers
+using RabbitMQ and AMQP.
+
+## 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 the various components, run in different terminals the following
+commands:
+
+```bash
+node worker.js # runs a worker that will process tasks
+node worker.js # runs a second worker that will process tasks (you can run as many as you want)
+node collector.js # runs a collector that will collect results from the workers
+node producer.js 4 f8e966d1e207d02c44511a58dccff2f5429e9a3b # runs a producer that will send tasks to the workers
+```
+
+> [!TIP] If you want to test other hashes, you can generate them with the
+> following code:
+>
+> ```js
+> import { createHash } from "node:crypto";
+> console.log(createHash("sha1").update("your-string-here").digest("hex"));
+> ```

13-messaging-and-integration-patterns/07-task-distribution-amqp/collector.js

@@ -0,0 +1,10 @@
+import amqp from 'amqplib' // v0.10.8
+
+const connection = await amqp.connect('amqp://localhost')
+const channel = await connection.createChannel()
+const { queue } = await channel.assertQueue('results_queue')
+
+channel.consume(queue, async msg => {
+  console.log(`Message from worker: ${msg.content.toString()}`)
+  await channel.ack(msg)
+})

13-messaging-and-integration-patterns/07-task-distribution-amqp/generateTasks.js

@@ -0,0 +1,27 @@
+export function* generateTasks(searchHash, alphabet, maxWordLength, batchSize) {
+  const alphabetLength = BigInt(alphabet.length)
+  const maxWordLengthBigInt = BigInt(maxWordLength)
+  let nVariations = 0n
+  for (let n = 1n; n <= maxWordLengthBigInt; n++) {
+    nVariations += alphabetLength ** n
+  }
+  console.log(
+    `Finding the hashsum source string over ${nVariations} possible variations`
+  )
+
+  let batchStart = 1n
+  while (batchStart <= nVariations) {
+    const expectedBatchSize = batchStart + BigInt(batchSize) - 1n
+    const batchEnd =
+      expectedBatchSize > nVariations ? nVariations : expectedBatchSize
+    yield JSON.stringify({
+      searchHash,
+      alphabet: alphabet,
+      // convert BigInt to string for JSON serialization
+      batchStart: batchStart.toString(),
+      batchEnd: batchEnd.toString(),
+    })
+
+    batchStart = batchEnd + 1n
+  }
+}

13-messaging-and-integration-patterns/07-task-distribution-amqp/package.json

@@ -0,0 +1,18 @@
+{
+  "name": "07-task-distribution-amqp",
+  "version": "1.0.0",
+  "description": "This sample demonstrates how to distribute tasks to a set of remote workers using RabbitMQ and AMQP",
+  "type": "module",
+  "scripts": {},
+  "engines": {
+    "node": ">=24"
+  },
+  "engineStrict": true,
+  "keywords": [],
+  "author": "Luciano Mammino and Mario Casciaro",
+  "license": "MIT",
+  "dependencies": {
+    "indexed-string-variation": "^2.1.0",
+    "amqplib": "^0.10.8"
+  }
+}

13-messaging-and-integration-patterns/07-task-distribution-amqp/pnpm-lock.yaml

@@ -0,0 +1,29 @@
+import { createHash } from 'node:crypto'
+import isv from 'indexed-string-variation' // v2.0.1
+
+export function processTask(task) {
+  const strings = isv({
+    alphabet: task.alphabet,
+    from: BigInt(task.batchStart),
+    to: BigInt(task.batchEnd),
+  })
+
+  let first
+  let last
+  for (const string of strings) {
+    if (!first) {
+      first = string
+    }
+
+    const digest = createHash('sha1').update(string).digest('hex')
+
+    if (digest === task.searchHash) {
+      console.log(`>> Found: ${string} => ${digest}`)
+      return string
+    }
+    last = string
+  }
+  console.log(
+    `Processed ${first}..${last} (${task.batchStart}..${task.batchEnd})`
+  )
+}

13-messaging-and-integration-patterns/07-task-distribution-amqp/processTask.js

@@ -0,0 +1,21 @@
+import amqp from 'amqplib' // v0.10.8
+import { generateTasks } from './generateTasks.js'
+
+const ALPHABET = 'abcdefghijklmnopqrstuvwxyz'
+const BATCH_SIZE = 10000
+
+const [, , maxLength, searchHash] = process.argv
+
+const connection = await amqp.connect('amqp://localhost')
+const channel = await connection.createConfirmChannel()
+await channel.assertQueue('tasks_queue')
+
+const generatorObj = generateTasks(searchHash, ALPHABET, maxLength, BATCH_SIZE)
+for (const task of generatorObj) {
+  console.log(`Sending task: ${task}`)
+  await channel.sendToQueue('tasks_queue', Buffer.from(task))
+}
+
+await channel.waitForConfirms()
+channel.close()
+connection.close()