serverless/test/fixtures/programmatic/function-rabbit-mq/core.js

'use strict'

// NOTE: `amqplib` is bundled into the deployment package
// eslint-disable-next-line import/no-unresolved
const amqp = require('amqplib')

function consumer(event, context, callback) {
  const functionName = 'consumer'
  const messages = []

  Object.keys(event.rmqMessagesByQueue).forEach((queueKey) => {
    const queue = event.rmqMessagesByQueue[queueKey]
    queue.forEach((message) => {
      messages.push(Buffer.from(message.data, 'base64').toString())
    })
  })
  // eslint-disable-next-line no-console
  console.log(functionName, JSON.stringify(messages))

  return callback(null, event)
}

async function producer() {
  const connectOptions = {
    protocol: 'amqps',
    hostname: process.env.RABBITMQ_HOST,
    port: 5671,
    username: process.env.RABBITMQ_USERNAME,
    password: process.env.RABBITMQ_PASSWORD,
  }

  const connection = await amqp.connect(connectOptions)
  const channel = await connection.createChannel()
  const queueName = process.env.QUEUE_NAME
  await channel.assertQueue(queueName)
  await channel.sendToQueue(
    queueName,
    Buffer.from('Hello from RabbitMQ Integration test!'),
  )

  return {
    statusCode: 200,
  }
}

module.exports = { producer, consumer }