(번역) RabbitMQ 및 Node.js를 사용한 비동기 마이크로 서비스

require('dotenv').config()

const amqp = require('amqplib')

// RabbitMQ connection string
const messageQueueConnectionString = process.env.CLOUDAMQP_URL

async function setup() {
  console.log('Setting up RabbitMQ Exchanges/Queues')
  // connect to RabbitMQ Instance
  let connection = await amqp.connect(messageQueueConnectionString)

  // create a channel
  let channel = await connection.createChannel()

  // create exchange
  await channel.assertExchange('processing', 'direct', { durable: true })

  // create queues
  await channel.assertQueue('processing.requests', { durable: true })
  await channel.assertQueue('processing.results', { durable: true })

  // bind queues
  await channel.bindQueue('processing.requests', 'processing', 'request')
  await channel.bindQueue('processing.results', 'processing', 'result')

  console.log('Setup DONE')
  process.exit()
}

setup()

const path = require('path')
require('dotenv').config({ path: path.resolve(process.cwd(), '../.env') })

const express = require('express')
const app = express()
const http = require('http')
const bodyParser = require('body-parser')
const amqp = require('amqplib')

// Middleware
app.use(bodyParser.json())

// simulate request ids
let lastRequestId = 1

// RabbitMQ connection string
const messageQueueConnectionString = process.env.CLOUDAMQP_URL

// handle the request
app.post('/api/v1/processData', async function(req, res) {
  // save request id and increment
  let requestId = lastRequestId
  lastRequestId++

  // connect to Rabbit MQ and create a channel
  let connection = await amqp.connect(messageQueueConnectionString)
  let channel = await connection.createConfirmChannel()

  // publish the data to Rabbit MQ
  let requestData = req.body.data
  console.log('Published a request message, requestId:', requestId)
  await publishToChannel(channel, {
    routingKey: 'request',
    exchangeName: 'processing',
    data: { requestId, requestData },
  })

  // send the request id in the response
  res.send({ requestId })
})

// utility function to publish messages to a channel
function publishToChannel(channel, { routingKey, exchangeName, data }) {
  return new Promise((resolve, reject) => {
    channel.publish(
      exchangeName,
      routingKey,
      Buffer.from(JSON.stringify(data), 'utf-8'),
      { persistent: true },
      function(err, ok) {
        if (err) {
          return reject(err)
        }

        resolve()
      }
    )
  })
}

async function listenForResults() {
  // connect to Rabbit MQ
  let connection = await amqp.connect(messageQueueConnectionString)

  // create a channel and prefetch 1 message at a time
  let channel = await connection.createChannel()
  await channel.prefetch(1)

  // start consuming messages
  await consume({ connection, channel })
}

// consume messages from RabbitMQ
function consume({ connection, channel, resultsChannel }) {
  return new Promise((resolve, reject) => {
    channel.consume('processing.results', async function(msg) {
      // parse message
      let msgBody = msg.content.toString()
      let data = JSON.parse(msgBody)
      let requestId = data.requestId
      let processingResults = data.processingResults
      console.log(
        'Received a result message, requestId:',
        requestId,
        'processingResults:',
        processingResults
      )

      // acknowledge message as received
      await channel.ack(msg)
    })

    // handle connection closed
    connection.on('close', err => {
      return reject(err)
    })

    // handle errors
    connection.on('error', err => {
      return reject(err)
    })
  })
}

// Start the server
const PORT = 3000
server = http.createServer(app)
server.listen(PORT, 'localhost', function(err) {
  if (err) {
    console.error(err)
  } else {
    console.info('Listening on port %s.', PORT)
  }
})

// listen for results on RabbitMQ
listenForResults()

const path = require('path')
require('dotenv').config({ path: path.resolve(process.cwd(), '../.env') })

const amqp = require('amqplib')

// RabbitMQ connection string
const messageQueueConnectionString = process.env.CLOUDAMQP_URL

async function listenForMessages() {
  // connect to Rabbit MQ
  let connection = await amqp.connect(messageQueueConnectionString)

  // create a channel and prefetch 1 message at a time
  let channel = await connection.createChannel()
  await channel.prefetch(1)

  // create a second channel to send back the results
  let resultsChannel = await connection.createConfirmChannel()

  // start consuming messages
  await consume({ connection, channel, resultsChannel })
}

// utility function to publish messages to a channel
function publishToChannel(channel, { routingKey, exchangeName, data }) {
  return new Promise((resolve, reject) => {
    channel.publish(
      exchangeName,
      routingKey,
      Buffer.from(JSON.stringify(data), 'utf-8'),
      { persistent: true },
      function(err, ok) {
        if (err) {
          return reject(err)
        }

        resolve()
      }
    )
  })
}

// consume messages from RabbitMQ
function consume({ connection, channel, resultsChannel }) {
  return new Promise((resolve, reject) => {
    channel.consume('processing.requests', async function(msg) {
      // parse message
      let msgBody = msg.content.toString()
      let data = JSON.parse(msgBody)
      let requestId = data.requestId
      let requestData = data.requestData
      console.log('Received a request message, requestId:', requestId)

      // process data
      let processingResults = await processMessage(requestData)

      // publish results to channel
      await publishToChannel(resultsChannel, {
        exchangeName: 'processing',
        routingKey: 'result',
        data: { requestId, processingResults },
      })
      console.log('Published results for requestId:', requestId)

      // acknowledge message as processed successfully
      await channel.ack(msg)
    })

    // handle connection closed
    connection.on('close', err => {
      return reject(err)
    })

    // handle errors
    connection.on('error', err => {
      return reject(err)
    })
  })
}

// simulate data processing that takes 5 seconds
function processMessage(requestData) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve(requestData + '-processed')
    }, 5000)
  })
}

listenForMessages()

$ *cd web-service
$ node web-service.js
*Listening on port 3000.

*// From a different terminal
*$ *curl --header "Content-Type: application/json" \
  --request POST \
  --data '{"data":"my-data"}' \
  [http://localhost:3000/api/v1/processData](http://localhost:3000/api/v1/processData)*

{"requestId":1}

$ *curl --header "Content-Type: application/json" \
  --request POST \
  --data '{"data":"more-data"}' \
  [http://localhost:3000/api/v1/processData](http://localhost:3000/api/v1/processData)*

{"requestId":2}

$ cd web-service
$ node web-service.js
Received a result message, requestId: 1 processingResults: my-data-processed
Received a result message, requestId: 2processingResults: more-data-processed