const sub = amqp.subscriber('rpc_queue');
const pub = amqp.publisher({ contentType: 'text/plain' });
const transformer = AmqpClient.transformer((message, callback) => {
    const n = Number(message.content);
    callback(null, fibonacci(n).toString());
});
sub.pipe(transformer).pipe(pub);

publisher(queue, options) publisher(exchange, routingKey, options)

// without options and callback
     publish('foo', 'bar', 'message')
     // without options
     publish('foo', 'bar', 'message', (err) => {})
     // without callback
     publish('foo', 'bar', 'message', {expirationMs: 15})
     // with options and callback
     publish('foo', 'bar', 'message', {expirationMs: 15}, (err) => {})

subscriber(queue, options) Readable, stream subscribe on messages from queue subscriber(exchange, routingKey, options) Readable, create exclusive queue and bind it to the exchange If incoming message has replyTo property

NB RabbitMQ v3.3.0 changes the meaning of prefetch (basic.qos) to apply per-consumer, rather than per-channel. It will apply to consumers started after the method is called. See rabbitmq-prefetch.

Use the global flag to get the per-channel behaviour. To keep life interesting, using the global flag with an RabbitMQ older than v3.3.0 will bring down the whole connection.

Close AMQP connection

Assert a queue into existence. This operation is idempotent given identical arguments; however, it will bork the channel if the queue already exists but has different properties (values supplied in the arguments field may or may not count for borking purposes).

Check whether a queue exists. This will bork the channel if the named queue doesn't exist; if it does exist, you go through to the next round!

Delete the queue named. Naming a queue that doesn't exist will result in the server closing the channel, to teach you a lesson (except in RabbitMQ version 3.2.0 and after1).

Note the obverse semantics of the options: if both are true, the queue will be deleted only if it has no consumers and no messages. You should leave out the options altogether if you want to delete the queue unconditionally.

The server reply contains a single field, messageCount, with the number of messages deleted or dead-lettered along with the queue.

Remove all undelivered messages from the queue named. Note that this won't remove messages that have been delivered but not yet acknowledged; they will remain, and may be requeued under some circumstances (e.g., if the channel to which they were delivered closes without acknowledging them).

The server reply contains a single field, messageCount, containing the number of messages purged from the queue.

Assert a routing path from an exchange to a queue: the exchange named by source will relay messages to the queue named, according to the type of the exchange and the pattern given. Pattern is a list of words, delimited by dots. The words can be anything, but usually they specify some features connected to the message.

There are two important special cases for binding keys:

• * (star) can substitute for exactly one word,
• # (hash) can substitute for zero or more words.

A few valid routing key examples: "stock.usd.nyse", "quick.orange.rabbit", "*.*.rabbit".

Remove a routing path between the queue named and the exchange named as source with the pattern and arguments given.

Assert an exchange into existence. As with queues, if the exchange exists already and has properties different to those supplied, the channel will 'splode; fields in the arguments object may or may not be 'splodey, depending on the type of exchange. Unlike queues, you must supply a name, and it can't be the empty string. You must also supply an exchange type, which determines how messages will be routed through the exchange.

Check that an exchange exists. If it doesn't exist, the channel will be closed with an error. If it does exist, happy days.

Delete exchange, if the exchange does not exist, a channel error is raised (RabbitMQ version 3.2.0 and after will not raise an error1).

Bind an exchange to another exchange. The exchange named by destination will receive messages from the exchange named by source, according to the type of the source and the pattern given. For example, a direct exchange will relay messages that have a routing key equal to the pattern.

NB Exchange to exchange binding is a RabbitMQ extension.

Remove a binding from an exchange to another exchange. A binding with the exact source exchange, destination exchange, routing key pattern, and extension args will be removed. If no such binding exists, it's – you guessed it – a channel error, except in RabbitMQ >= version 3.2.0, for which it succeeds trivially 1.

Ask a queue for a message, as an RPC. This will be resolved with either false, if there is no message to be had (the queue has no messages ready), or a message.

Acknowledge all outstanding messages on the channel. This is a "safe" operation, in that it won't result in an error even if there are no such messages.

Reject all messages outstanding on this channel. If requeue is truthy, or omitted, the server will try to re-enqueue the messages.

Requeue unacknowledged messages on this channel. The server will reply (with an empty object) once all messages are requeued.