handleCommand.ts

const handleCommand = async ({
  channel,
  from,
  func,
  to,
}: {
  from: string;
  to: string;
  func: (args: any) => any;
  channel: amqp.Channel;
}) => {
  await channel.assertQueue(from, { durable: true });
  channel.consume(
    from,
    async (msg) => {
      channel.ack(msg);
      const payload = JSON.parse(msg.content.toString());
      try {
        const result = await func(payload);
        channel.sendToQueue(to, Buffer.from(JSON.stringify(result)), {
          correlationId: msg.properties.correlationId,
          persistent: true,
        });
      } catch (e) {
        channel.sendToQueue(to, Buffer.from(JSON.stringify({ error: { name: e.name, message: e.message } })), {
          correlationId: msg.properties.correlationId,
          persistent: true,
        });
      }
    },
    { noAck: false },
  );
};