node.js imqplib sendToQueue zu RabbitMQ hängt

Question

Ich habe eine Funktion für eine Nachricht an RabbitMQ Warteschlangen wie folgt:

var amqp = require('amqplib/callback_api'); 

var _queueURL = 'amqp://127.0.0.1'; 
var _toBlahBlahQueueName = 'blahblah'; 

var self = module.exports = { 
    queueMessage: function (msgObj, callback) { 
    try { 
     amqp.connect(_queueURL, function (err, connection) { 
     if (err) { 
      callback(err); 
     } 

     connection.createChannel(function (err, channel) { 
      if (err) { 
      callback(err); 
      } 

      channel.assertQueue(_toBlahBlahQueueName, { durable: true }, function (err, _ok) { 
      if (err) { 
       callback(err); 
      } 

      var msg = new Buffer(JSON.stringify(msgObj)); 

      channel.sendToQueue(_toBlahBlahQueueName, msg, { persistent: true }, function (err, ok) { 
       if (err) { 
       console.log(err); 
       callback(err); 
       } 

       console.log('published', ok); 

       channel.connection.close(); 
       callback(null, { message: 'queued' }); 
      }); 
      }); 
     }); 
     }); 
    } 
    catch (e) { 
     console.log(e.stack); 
     callback(e); 
    } 
    } 
}; 

Ich bin die Funktion queueMessage mit Nachrichten aufrufen, die über 250K lang sind.

Der sendToQueue-Aufruf hängt jedes Mal. Es sitzt nur da, ohne einen Fehler zu melden. Die Nachricht scheint jedoch in die Warteschlange eingereiht zu werden!

Das Serverprotokoll hat die Fehlermeldung: Client unerwartet geschlossen TCP-Verbindung

Vielen Dank für jede Hilfe!

Answer 1

amqplib unterstützt keinen Rückruf für sendToQueue oder publish.

the documentation zeigt dies keine Option ist:

Kanal # sendToQueue Versprechen und Rückrufe

sendToQueue (Queue, Inhalt, [Optionen])

eine einzelne Nachricht mit dem Inhalt senden als Puffer für die spezifische Queue namens Bypass-Routing gegeben. Die Optionen und der Rückgabewert sind exakt die gleichen wie für publish.

Um dies zu umgehen, müssen Sie sendToQueue aufrufen, als wäre es eine synchrone Nachricht.

Wenn Sie die App sofort beenden möchten, müssen Sie einige Millisekunden warten. Wenn Sie dies nicht tun, wird die Nachricht nicht gesendet. Hier

ist ein Beispiel, wie Sie Ihren Code ändern könnte auf diese Weise funktionieren:

channel.sendToQueue(_toBlahBlahQueueName, msg, { persistent: true }); 
setTimeout(function() { 
    channel.connection.close(); 
    callback(null, { message: 'queued' }); 
}, 500);