few questions

[panosru]

Hello, using this example I do some tests to node-amqp package.

var amqp = require('amqp');
var connection = amqp.createConnection({ url : 'amqp://guest:guest@localhost:5672' });
var queue_id = 'queue.node-amqp-rnd';

connection.addListener('ready', function () {
  console.log('Connection Ready');
  this.queue(queue_id, { durable : true }, function (queue) {
    console.log('Queue created');
    connection.exchange(queue_id, { type : 'direct', durable : true, autoDelete : false }, function (exchange) {
      console.log('Exchange created');
      queue.bind(exchange.name, queue.name);
      console.log('Queue {' + queue.name + '} bind to Exchange {' + exchange.name + '}');

      // Subscribe to queue after 30 seconds
      setTimeout(function () {
        queue.subscribe(function (message, headers, deliveryInfo) {
          console.log('Got message: ');
          console.log(message.data.toString());
        });
      }, 30000);

      // Add message to queue (with exchange.publish)
      setTimeout(function () {
        console.log('Adding test message to queue');
        exchange.publish(
          queue_id,
          'Some test content',
          { deliveryMode : 2, contentType : 'text/plain' }
        );
      }, 2000);

      // Add message to queue (with connection.publish)
      setTimeout(function () {
        console.log('Adding test message to queue');
        connection.publish(
          queue_id,
          'Some more content goes here'
        );
      }, 3000);
    });
  });
});

So runing the above I get this output:

 #> node amqp.js 
Connection Ready
Queue created
Exchange created
Queue {queue.node-amqp-rnd} bind to Exchange {queue.node-amqp-rnd}
Adding test message to queue
Adding test message to queue
Got message: 
Some test content
Got message: 
Some more content goes here

I got used to use php-amqplib and I have few questions for node-amqp.

1) What's the difference between connection.publish and exchange.publish besides that fact that you can pass options to exchange.publish?

2) How to get manually commit a message?

3) How do you perform a rollback?

4) How do you remove a message from queue?

5) How do you get list of messages from a queue?

Sorry for many questions I just couldn't find the methods in the library even after checking out the source code of amqp.js and since those functionalities exists in PHP ampq lib I though it's best to ask to find out how to do it or if they are not implemented.

Thanks in advance!

[zeisss]

Hi panosru,

I'm not postwait, but maybe I can help :)

1) connection.publish publishes to the default exchange (normally the nameless one you see in the management plugin), but this can be configured. The first parameter you provide here is the routing key, which is equal the queue name when using the default exchange.

exchange.publish is more explicit and makes it possible to provide a routing key on an explicit exchange.

2) I don't exactly understand. What do you mean by commit?

3) I don't understand this neither. Afaik there is no concept of a rollback in AMQP. Consumers can request an acknowledgement mode when subscribing to a queue. This has the effect that on connection loss, any unacknowledged message gets requeued. Can you detail you question a bit more?

4) Subscribe to a queue and provide a message handler. The message handler will be invoked for every message deliverey to the consumer. This propably differs from php-amqplib, since node.js is aync by design.

5) You can't. You can only subscribe and republish the messages, but this destroys the order of messages in the queue. If you are only interested in the length of a queue, you have several options:

• the management plugin provides a status api
• call connection.queue with the passive:true option to read the state of the queue.

[postwait]

I don't think transaction support is implemented in node-amqp.... I'd need to look. So, IIRC all message sent are done so non-transactionally: no need to commit, not possibility of rollback.

[panosru]

@ZeissS

Thanks a lot for your reply!

1) Ok I got that ;) I'll stick with exchange.publish :)

2) 3) By commit I mean in case a transaction does not fail to perform a commit that would actually remove the messages from queue. By rollback I mean, in case of transaction failuter to "rollback" to previous state before the start of transaction but since @postwait confirmed that transaction support is not yet implemented so is not possible to rollback. PHP AMQPChannel RabbitMQ Broker Semantics :)

4) Would you mind to provide a simple example in order to get the point? Messages in PHP also work async, of course PHP itself is not async by design.

5) If you don't set auto delete for messages and for any reason you need to know not only how many messages are in the queue but you need to iterate through them is not possible? Would be great if you could have an array of objects where each object would be a message object.

One more question I forgot to ask, is it possible to get a message from queue? For instance:

var amqp = require('amqp');

var connection = amqp.createConnection({ host: 'dev.rabbitmq.com' });

// Wait for connection to become established.
connection.on('ready', function () {
  // Use the default 'amq.topic' exchange
  connection.queue('my-queue', function(q){
      // Get message by message id (Sync)
      var message = q.get('message-id-here');

      // Get message by message id (Async)
      q.get('message-id-here', function (err, message) {
           if (err) { 
               // Handle error in case message does not exist 
            } else {
              // Do anything with message
           }
      });
  });
});

@postwait Thanks for clarifying that transactions are not yet supported, are there any plans to support them?

Thanks a lot to everyone :)

[zeisss]

@panossru Regarding 4+5) Afaik (I may be wrong), AMQP does not allow arbitrary (by message-id/all) access to the messages nor deleting specific messages. I guess you could write a plugin to allow that, but I don't know much of the plugin world.