chore: working on chapter 13

Author: lmammino

13-messaging-and-integration-patterns/08-task-distribution-redis-streams/README.md

@@ -0,0 +1,60 @@
+# 08-task-distribution-redis-streams
+
+This sample demonstrates how to distribute tasks to a set of remote workers
+using Redis Streams.
+
+## Dependencies
+
+As a pre-requisite to this sample, you first need to install
+[Redis](http://redis.io/download) and have it running locally on its default
+port.
+
+If you have docker installed you can easily run an ephemeral redis instance
+locally with:
+
+```bash
+docker run -it -p 6379:6379 redis redis-server --appendonly yes
+```
+
+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 workerA # runs a worker that will process tasks
+node worker.js workerB # runs a second worker that will process tasks (you can run as many as you want but make sure to use a different name for each worker)
+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/08-task-distribution-redis-streams/collector.js

@@ -0,0 +1,20 @@
+import Redis from 'ioredis' // v5.6.1
+
+const redisClient = new Redis()
+
+let lastRecordId = '$'
+while (true) {
+  const data = await redisClient.xread(
+    'BLOCK',
+    '0',
+    'STREAMS',
+    'results_stream',
+    lastRecordId
+  )
+  for (const [, logs] of data) {
+    for (const [recordId, [, message]] of logs) {
+      console.log(`Message from worker: ${message}`)
+      lastRecordId = recordId
+    }
+  }
+}

13-messaging-and-integration-patterns/08-task-distribution-redis-streams/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/08-task-distribution-redis-streams/package.json

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

13-messaging-and-integration-patterns/08-task-distribution-redis-streams/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/08-task-distribution-redis-streams/processTask.js

@@ -0,0 +1,17 @@
+import Redis from 'ioredis' // v5.6.1
+import { generateTasks } from './generateTasks.js'
+
+const ALPHABET = 'abcdefghijklmnopqrstuvwxyz'
+const BATCH_SIZE = 10000
+
+const [, , maxLength, searchHash] = process.argv
+
+const redisClient = new Redis()
+
+const generatorObj = generateTasks(searchHash, ALPHABET, maxLength, BATCH_SIZE)
+for (const task of generatorObj) {
+  console.log(`Sending task: ${task}`)
+  await redisClient.xadd('tasks_stream', '*', 'task', task)
+}
+
+redisClient.disconnect()

13-messaging-and-integration-patterns/08-task-distribution-redis-streams/producer.js

@@ -0,0 +1,49 @@
+import Redis from 'ioredis' // v5.6.1
+import { processTask } from './processTask.js'
+
+const redisClient = new Redis()
+const [, , consumerName] = process.argv
+
+await redisClient
+  .xgroup('CREATE', 'tasks_stream', 'workers_group', '$', 'MKSTREAM')
+  .catch(() => console.log('Consumer group already exists'))
+
+const [[, records]] = await redisClient.xreadgroup(
+  'GROUP',
+  'workers_group',
+  consumerName,
+  'STREAMS',
+  'tasks_stream',
+  '0'
+)
+for (const [recordId, [, rawTask]] of records) {
+  await processAndAck(recordId, rawTask)
+}
+
+while (true) {
+  const [[, records]] = await redisClient.xreadgroup(
+    'GROUP',
+    'workers_group',
+    consumerName,
+    'BLOCK',
+    '0',
+    'COUNT',
+    '1',
+    'STREAMS',
+    'tasks_stream',
+    '>'
+  )
+  for (const [recordId, [, rawTask]] of records) {
+    await processAndAck(recordId, rawTask)
+  }
+}
+
+async function processAndAck(recordId, rawTask) {
+  const found = processTask(JSON.parse(rawTask))
+  if (found) {
+    console.log(`Found! => ${found}`)
+    await redisClient.xadd('results_stream', '*', 'result', `Found: ${found}`)
+  }
+
+  await redisClient.xack('tasks_stream', 'workers_group', recordId)
+}