fix: fix the majordomo example

Author: aminya

examples/majordomo/README.md

@@ -6,12 +6,12 @@ not send or listen to heartbeats.
 
 ## Running this example
 
-To run this example, install the example project depedencies and run the
-majordomo example script with `yarn`:
+To run this example, install the example project dependencies and run the
+majordomo example script with `pnpm`:
 
 ```
-> yarn install
-> yarn majordomo
+> pnpm install
+> pnpm run majordomo
 ```
 
 ## Expected behaviour
@@ -20,7 +20,13 @@ The example will start a broker and some workers, then do some requests. The
 output will be similar to this:
 
 ```
+
 starting broker on tcp://127.0.0.1:5555
+starting worker on tcp://127.0.0.1:5555
+starting worker on tcp://127.0.0.1:5555
+starting worker on tcp://127.0.0.1:5555
+---------- Started -----------
+requesting 'cola' from 'soda'
 requesting 'oolong' from 'tea'
 requesting 'sencha' from 'tea'
 requesting 'earl grey, with milk' from 'tea'
@@ -29,33 +35,33 @@ requesting 'cappuccino' from 'coffee'
 requesting 'latte, with soy milk' from 'coffee'
 requesting 'espresso' from 'coffee'
 requesting 'irish coffee' from 'coffee'
-registered worker 00800041af for 'coffee'
-dispatching 'coffee' 00800041ab req -> 00800041af
-registered worker 00800041b0 for 'tea'
-dispatching 'tea' 00800041a7 req -> 00800041b0
-registered worker 00800041b1 for 'tea'
-dispatching 'tea' 00800041a8 req -> 00800041b1
-dispatching 'tea' 00800041a7 <- rep 00800041b0
-dispatching 'tea' 00800041a9 req -> 00800041b0
-received 'oolong' from 'tea'
-dispatching 'coffee' 00800041ab <- rep 00800041af
-dispatching 'coffee' 00800041ac req -> 00800041af
-received 'cappuccino' from 'coffee'
-dispatching 'tea' 00800041a9 <- rep 00800041b0
-dispatching 'tea' 00800041aa req -> 00800041b0
-received 'earl grey, with milk' from 'tea'
-dispatching 'tea' 00800041a8 <- rep 00800041b1
+registered worker 00800041a7 for 'tea'
+registered worker 00800041a8 for 'coffee'
+registered worker 00800041a9 for 'tea'
+dispatching 'tea' 00800041ab req -> 00800041a7
+dispatching 'tea' 00800041ac req -> 00800041a9
+dispatching 'coffee' 00800041af req -> 00800041a8
+dispatching 'tea' 00800041ac <- rep 00800041a9
+dispatching 'tea' 00800041ad req -> 00800041a9
 received 'sencha' from 'tea'
-dispatching 'tea' 00800041aa <- rep 00800041b0
-received 'jasmine' from 'tea'
-dispatching 'coffee' 00800041ac <- rep 00800041af
-dispatching 'coffee' 00800041ad req -> 00800041af
+dispatching 'tea' 00800041ad <- rep 00800041a9
+dispatching 'tea' 00800041ae req -> 00800041a9
+received 'earl grey, with milk' from 'tea'
+dispatching 'coffee' 00800041af <- rep 00800041a8
+dispatching 'coffee' 00800041b0 req -> 00800041a8
+received 'cappuccino' from 'coffee'
+dispatching 'coffee' 00800041b0 <- rep 00800041a8
+dispatching 'coffee' 00800041b1 req -> 00800041a8
 received 'latte, with soy milk' from 'coffee'
-dispatching 'coffee' 00800041ad <- rep 00800041af
-dispatching 'coffee' 00800041ae req -> 00800041af
+dispatching 'coffee' 00800041b1 <- rep 00800041a8
+dispatching 'coffee' 00800041b2 req -> 00800041a8
 received 'espresso' from 'coffee'
-dispatching 'coffee' 00800041ae <- rep 00800041af
+dispatching 'tea' 00800041ae <- rep 00800041a9
+received 'jasmine' from 'tea'
+dispatching 'coffee' 00800041b2 <- rep 00800041a8
 received 'irish coffee' from 'coffee'
+dispatching 'tea' 00800041ab <- rep 00800041a7
+received 'oolong' from 'tea'
 timeout expired waiting for 'soda'
-deregistered worker 00800041b1 for 'tea'
+---------- Stopping -----------
 ```

examples/majordomo/broker.ts

@@ -17,22 +17,18 @@ export class Broker {
     console.log(`starting broker on ${this.address}`)
     await this.socket.bind(this.address)
 
-    const loop = async () => {
-      for await (const [sender, _blank, header, ...rest] of this.socket) {
-        switch (header.toString()) {
-          case Header.Client:
-            this.handleClient(sender, ...rest)
-            break
-          case Header.Worker:
-            this.handleWorker(sender, ...rest)
-            break
-          default:
-            console.error(`invalid message header: ${header}`)
-        }
+    for await (const [sender, _blank, header, ...rest] of this.socket) {
+      switch (header.toString()) {
+        case Header.Client:
+          await this.handleClient(sender as Buffer, ...(rest as Buffer[]))
+          break
+        case Header.Worker:
+          await this.handleWorker(sender as Buffer, ...(rest as Buffer[]))
+          break
+        default:
+          console.error(`invalid message header: ${header}`)
       }
     }
-
-    return loop()
   }
 
   async stop() {
@@ -43,33 +39,30 @@ export class Broker {
 
   handleClient(client: Buffer, service?: Buffer, ...req: Buffer[]) {
     if (service) {
-      this.dispatchRequest(client, service, ...req)
+      return this.dispatchRequest(client, service, ...req)
     }
   }
 
   handleWorker(worker: Buffer, type?: Buffer, ...rest: Buffer[]) {
     switch (type?.toString()) {
       case Message.Ready: {
         const [service] = rest
-        this.register(worker, service)
-        break
+        return this.register(worker, service)
       }
 
       case Message.Reply: {
         const [client, _blank, ...rep] = rest
-        this.dispatchReply(worker, client, ...rep).catch(err => {
+        return this.dispatchReply(worker, client, ...rep).catch(err => {
           console.error(err)
         })
-        break
       }
 
       case Message.Heartbeat:
         /* Heartbeats not implemented yet. */
         break
 
       case Message.Disconnect:
-        this.deregister(worker)
-        break
+        return this.deregister(worker)
 
       default:
         console.error(`invalid worker message type: ${type}`)
@@ -78,11 +71,11 @@ export class Broker {
 
   register(worker: Buffer, service: Buffer) {
     this.setWorkerService(worker, service)
-    this.getService(service).register(worker)
+    return this.getService(service).register(worker)
   }
 
   dispatchRequest(client: Buffer, service: Buffer, ...req: Buffer[]) {
-    this.getService(service).dispatchRequest(client, ...req)
+    return this.getService(service).dispatchRequest(client, ...req)
   }
 
   dispatchReply(worker: Buffer, client: Buffer, ...rep: Buffer[]) {
@@ -92,7 +85,7 @@ export class Broker {
 
   deregister(worker: Buffer) {
     const service = this.getWorkerService(worker)
-    this.getService(service).deregister(worker)
+    return this.getService(service).deregister(worker)
   }
 
   getService(name: Buffer): Service {

examples/majordomo/index.ts

@@ -12,7 +12,7 @@ async function sleep(msec: number) {
 class TeaWorker extends Worker {
   service = "tea"
 
-  async process(...msgs: Buffer[]): Promise<Buffer[]> {
+  override async process(...msgs: Buffer[]): Promise<Buffer[]> {
     await sleep(Math.random() * 500)
     return msgs
   }
@@ -21,7 +21,7 @@ class TeaWorker extends Worker {
 class CoffeeWorker extends Worker {
   service = "coffee"
 
-  async process(...msgs: Buffer[]): Promise<Buffer[]> {
+  override async process(...msgs: Buffer[]): Promise<Buffer[]> {
     await sleep(Math.random() * 200)
     return msgs
   }
@@ -51,10 +51,14 @@ async function request(
 }
 
 async function main() {
-  for (const worker of workers) {
-    await worker.start()
-  }
-  await broker.start()
+  const started = Promise.all([
+    // start the broker
+    broker.start(),
+    // start the workers
+    ...workers.map(worker => worker.start()),
+  ])
+
+  console.log("---------- Started -----------")
 
   /* Requests are issued in parallel. */
   await Promise.all([
@@ -69,10 +73,16 @@ async function main() {
     request("coffee", "irish coffee"),
   ])
 
-  for (const worker of workers) {
-    await worker.stop()
-  }
-  await broker.stop()
+  console.log("---------- Stopping -----------")
+
+  await Promise.all([
+    // stop the broker
+    broker.stop(),
+    // stop the workers
+    ...workers.map(worker => worker.stop()),
+  ])
+  // await outstanding promises
+  await started
 }
 
 main().catch(err => {

examples/majordomo/worker.ts

@@ -13,34 +13,25 @@ export class Worker {
   }
 
   async start() {
+    console.log(`starting worker on ${this.address}`)
     await this.socket.send([null, Header.Worker, Message.Ready, this.service])
 
-    const loop = async () => {
-      for await (const [
-        _blank1,
-        _header,
-        _type,
-        client,
-        _blank2,
-        ...req
-      ] of this.socket) {
-        const rep = await this.process(...req)
-        try {
-          await this.socket.send([
-            null,
-            Header.Worker,
-            Message.Reply,
-            client,
-            null,
-            ...rep,
-          ])
-        } catch (err) {
-          console.error(`unable to send reply for ${this.address}`)
-        }
+    for await (const [_blank1, _header, _type, client, _blank2, ...req] of this
+      .socket) {
+      const rep = await this.process(...req)
+      try {
+        await this.socket.send([
+          null,
+          Header.Worker,
+          Message.Reply,
+          client,
+          null,
+          ...rep,
+        ])
+      } catch (err) {
+        console.error(`unable to send reply for ${this.address}`)
       }
     }
-
-    return loop()
   }
 
   async stop() {
@@ -55,6 +46,9 @@ export class Worker {
     }
   }
 
+  /**
+   * @virtual
+   */
   async process(...req: Buffer[]): Promise<Buffer[]> {
     return req
   }

examples/package.json

@@ -8,8 +8,8 @@
     "ts-node": ">= 0"
   },
   "scripts": {
-    "majordomo": "ts-node majordomo",
-    "queue": "ts-node queue",
-    "threaded-worker": "ts-node threaded-worker"
+    "majordomo": "ts-node ./majordomo/index.ts",
+    "queue": "ts-node ./queue/index.ts",
+    "threaded-worker": "ts-node ./threaded-worker/index.ts"
   }
 }