🐇 Publish/Subscribe (#2)

🐇 Publish/Subscribe

Author: lucasgianine

README.md

@@ -4,7 +4,7 @@ O intuito desse repositório é testar e aprender a usar o sistema de mensageria
 Cada etapa (exceto a primeira) será separada por Pull Requests para que possa ser mais fácil de identificar as etapas da versão final.
 
 ## ([#0](https://github.com/lucasgianine/message-queuing/commit/c86a7cd0750668b64d3e57371d61874107304e26)) Hello world!
-O conceito dessa etapa é apresentar o básico da mensageria, teremos um <i>Producer</i> que irá enviar uma mensagem, a <i>queue</i>, ou <i>fila</i> que irá fazer o processo onde tranformará a mensagem em um buffer para que, finalmente, seja entregue ao <i>Consumer</i>, que imprimirá a mensagem.
+O conceito dessa etapa é apresentar o básico da mensageria, teremos um <i>Producer</i> que irá enviar uma mensagem, a <i>queue</i>, ou <i>fila</i> que irá fazer o processo onde armazenará a mensagem em um buffer para que, finalmente, seja entregue ao <i>Consumer</i>, que imprimirá a mensagem.
 
 ```mermaid
 flowchart LR
@@ -31,7 +31,7 @@ Utilize esses comandos para teste:
   # -> [x] Sent: Hello World!
 ```
 
-## (#1) Work Queues
+## ([#1](https://github.com/lucasgianine/message-queuing/pull/1)) Work Queues
 Vamos trabalhar em criar Work Queues (ou Task Queues) para distribuir tarefas demoradas entre vários workers, ou seja, quando uma tarefa exije muitos recursos, todo fluxo espera que ela seja concluída para que a mensagem seja exibida, a ideia do Work Queues é que agendemos a tarefa para que ela seja feita mais tarde.
 
 ```mermaid
@@ -105,3 +105,71 @@ Pra corrigir esse feito, usamos `prefetch` com o valor `1` para que o Rabbit ent
 ```typescript
   channel.prefetch(1)
 ```
+
+## ([#2](https://github.com/lucasgianine/message-queuing/pull/2)) Publish/Subscribe
+Dessa vez iremos entregar uma mensagem para vários consumidores, criaremos um registro simples com dois programas, onde um emitirá mensagens de registro e outro que vai receber e imprimir, no nosso programa, cada cópia em execução do receptor receberá as mensagens, onde o receptor poderá se comunicar com os dois queues ao mesmo tempo.
+
+Toda ideia do Rabbit é que, na verdade o <i>producer</i> nunca envie mensagem diretamente para fila (pois na realidade é que o <i>producer</i> nem sabe se a mensagem chegará até lá), mas ao invés disso ele envie mensagens para uma `exchange`, pois ela sabe exatamente o que fazer com a mensagem que recebeu para empurrá-lá para uma <i>queue</i>.
+
+```mermaid
+flowchart LR
+  P["Producer"]
+  X{"Exchange"}
+  Q1["Queue 1"]
+  Q2["Queue 2"]
+
+  P --> X --> Q1
+  X --> Q2
+```
+
+Há alguns tipos de exchanges, mas vamos trabalhar em cima do `fanout`: Ela transmite todas as mensagens que recebe para todas as filas que ela tem conhecimento.
+```typescript
+  channel.assertExchange('logs', 'fanout', { durable: false })
+```
+
+#### Filas temporárias
+Dar o nome para uma fila é importante para compartilharmos ela entre os <i>produces</i> e <i>consumers</i>, mas no caso dessa aplicação de logs, não precisamos criar uma fila permanente, deixaremos o nome da fila vazio para que o próprio servidor possa dar um nome aleatório, já que nesse momento isso não é prioritário visto que a fila, depois de ser consumida, deverá ser apagada automaticamente.
+```typescript
+  channel.assertQueue('', {
+    exclusive: true
+  })
+
+  // Exemplo de retorno: amq.gen-JzTY20BRgKO-HjmUJj0wLg
+```
+
+Depois de todo o processo de criar <i>exchange</i> e as filas temporárias agora vamos fazer nossa <i>exchange</i> enviar mensagem para a <i>fila</i>.
+```mermaid
+flowchart LR
+  P["Producer"]
+  X{"Exchange"}
+  Q1["Queue 1"]
+  Q2["Queue 2"]
+
+  P --> X -- Binding --> Q1
+  X -- Binding --> Q2
+```
+
+Chamamos de `binding` o relacionamento entre exchange (troca) e uma queue (fila).
+```typescript
+  channel.bindQueue(queue_name, 'logs', '')
+```
+
+Utilize esses comandos para teste:
+```bash
+  # shell 1
+  npm run receive_logs
+
+  # -> Será criado um arquivo .log na pasta src/logs
+  # -> No arquivo aparecerá a <mensagem> escrita no próximo shell
+```
+
+```bash
+  # shell 2
+  npm run emit_logs <mensagem>
+
+  # -> [x] Sent: <mensagem>
+```
+
+## Referência
+- [RabbitMQ](https://www.rabbitmq.com/)
+- [Documentação do RabbitMQ](https://www.rabbitmq.com/tutorials)

package.json

@@ -6,7 +6,9 @@
   "scripts": {
     "test": "echo \"Error: no test specified\" && exit 1",
     "producer": "node src/new_task.js",
-    "consumer": "node src/worker.js"
+    "consumer": "node src/worker.js",
+    "emit_logs": "node src/logs/emit_logs.js",
+    "receive_logs": "node src/logs/receive_logs.js > src/logs/logs_emitidas.log"
   },
   "keywords": [],
   "author": "",

src/logs/emit_logs.js

@@ -0,0 +1,30 @@
+const amqp = require('amqplib/callback_api')
+
+amqp.connect('amqp://localhost', (err, conn) => {
+  if (err) throw err
+
+  conn.createChannel((err, channel) => {
+    if (err) throw err
+
+    const exchange = 'logs'
+    const message = process.argv.slice(2).join(' ') ?? 'Hello World!'
+
+    channel.assertExchange(exchange, 'fanout', {
+      durable: false
+    })
+
+    /*
+      A string vazia como segundo parâmetro significa que não queremos
+      enviar a mensagem para nenhuma fila específica.
+      Queremos apenas publicá-la em nossa troca de 'logs'.
+    */
+    channel.publish(exchange, '', Buffer.from(message))
+
+    console.log(` [x] Sent: ${message}`)
+  })
+
+  setTimeout(() => {
+    conn.close()
+    process.exit(0)
+  }, 500)
+})

src/logs/receive_logs.js

@@ -0,0 +1,27 @@
+const amqp = require('amqplib/callback_api')
+
+amqp.connect('amqp://localhost', (err, conn) => {
+  if (err) throw err
+
+  conn.createChannel((err, channel) => {
+    if (err) throw err
+
+    const exchange = 'logs'
+
+    channel.assertExchange(exchange, 'fanout', {
+      durable: false
+    })
+
+    channel.assertQueue('', {
+      exclusive: true
+    }, (err, q) => {
+      if (err) throw err
+
+      channel.bindQueue(q.queue, exchange, '')
+
+      channel.consume(q.queue, (message) => {
+        if (message.content) console.log(` [x] ${message.content.toString()}`)
+      }, { noAck: true })
+    })
+  })
+})