node.js - Node Js/Typescript - AMQP potrošač

Po prvi put pokušavam raditi s node.js/typescript i imam malo problema s izradom konzumenta za rabbit red.

Kod :

let amqp = require('amqp');

let connection = amqp.createConnection({url: "amqp://" + RABBITMQ_USER + ":" + RABBITMQ_PASSWORD + "@" + RABBITMQ_HOST + ":" + RABBITMQ_PORT + RABBITMQ_VHOST});

connection.on('ready', function() {
    connection.exchange(RABBITMQ_WORKER_EXCHANGE, function (exchange) {
        connection.queue(RABBITMQ_QUEUE, function (queue) {
            queue.bind(exchange, function() {
                queue.publish(function (message) {
                    console.log('subscribed to queue');
                    let encoded_payload = unescape(message.data);
                    let payload = JSON.parse(encoded_payload);
                    console.log('Received a message:');
                    console.log(payload);
                })
            })
        })
    })
})

Čini se da se povezuje na AMQP poslužitelj i ne baca nikakve pogreške, ali samo stoji tamo i ništa ne konzumira. Nedostaje li mi neki korak?

Bila kakva pomoć bi bila vrlo cijenjena, hvala.

Odgovor #1

import * as Amqp from "amqp-ts";


var connection = new Amqp.Connection("amqp://localhost");
var exchange = connection.declareExchange("ExchangeName");
var queue = connection.declareQueue("QueueName");
queue.bind(exchange);
queue.activateConsumer((message) => {
    console.log("Message received: " + message.getContent());
});
 
// it is possible that the following message is not received because
// it can be sent before the queue, binding or consumer exist
var msg = new Amqp.Message("Test");
exchange.send(msg);
 
connection.completeConfiguration().then(() => {
    // the following message will be received because
    // everything you defined earlier for this connection now exists
    var msg2 = new Amqp.Message("Test2");
    exchange.send(msg2);
});

Odgovor #2

Evo mog rješenja koje radi na temelju amqp-ovog JS tutorijala. https://www.rabbitmq.com/tutorials/tutorial-three-javascript.html

Vjerojatno nije u skladu sa standardima TypeScripta, slobodno me ispravite ako postoji bolji način.

#!/usr/bin/env node

require('dotenv').config();
import amqp = require('amqplib/callback_api');
import db = require('./database');

amqp.connect({
    protocol: process.env.RABBITMQ_PROTOCOL,
    hostname: process.env.RABBITMQ_HOST,
    port: process.env.RABBITMQ_PORT,
    username: process.env.RABBITMQ_USER,
    password: process.env.RABBITMQ_PASSWORD,
    vhost: process.env.RABBITMQ_VHOST
}, function(err, conn) {
    conn.createChannel(function (err, ch) {
        // set exchange that is being used
        ch.assertExchange(process.env.RABBITMQ_WORKER_EXCHANGE, 'direct', {durable: true});
        // set queue that is being used
        ch.assertQueue(process.env.RABBITMQ_QUEUE, {durable: true}, function (err, q) {
            console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q.queue);
            // bind the queue to the exchange
            ch.bindQueue(q.queue, process.env.RABBITMQ_WORKER_EXCHANGE, '');
            // consume from the queue, one message at a time.
            ch.consume(q.queue, function (msg) {
                console.log("Message received: %s", msg.content.toString());
                //save message to db
                db.store(msg.content.toString()).then(function() {
                    //acknowledge receipt of message to amqp
                    console.log("Acknowledging message");
                    ch.ack(msg, true);
                });
            }, {noAck: false});
        });
    });
});

Povezani članci