ConfirmChannel#publish / ConfirmChannel#sendToQueue should return a promise ?

[pdelanauze]

Right now , (v0.2.1) ConfirmChannel#publish / ConfirmChannel#sendToQueue dont seem to return a promise , they only accept a callback ? Not 100% sure about this , but our current testing is returning true even if no callback is supplied ... If our testing is right , would it be worthwhile to return a promise ?

[michaelklishin]

I assume it returns true or false to indicate whether all confirms were received successfully.

[pdelanauze]

Indeed it does , but would be awesome to have it consistent with other promise-like interfaces .. e.g.

ch.publish(...).then(function(ok){}, function(err){})

[michaelklishin]

@pdelanauze channels that don't use publisher confirms don't have a "success" value to return (publishing is then fire-and-forget). Channels that do use publisher confirms need to return a value. Other clients use a separate method for this, though, but it is uncommon to return a 2-tuple in JavaScript libraries.

[michaelklishin]

Channel#waitForConfirms or similar is what most other clients have. The method is supposed to be used with batches (publish N, wait once). amqp.node can perhaps return a future/promise, in which case we may need another name.

I'm not sure if the #publish return value needs to change.

[pdelanauze]

Yep, got it .
Agreed that they might need another name if it returns a promise .. Otherwise might be a slight pain for people migrating to the new version , have to add .then handlers otherwise might cause non resolved promises

[squaremo]

sendToQueue and publish both return a boolean indicating whether or not you can keep writing, by analogy with Node.JS's streams.

[squaremo]

I don't like the #waitForConfirms-style APIs for a couple of reasons:

1. It makes publishing "choppy" (send a fixed number of things, wait for an extended period)
2. If one of the confirmations is in fact a nack, you usually don't get told which message it was (not a problem with the style, but a problem with the implementations, to be fair)

On the other hand, I could probably do a bit better than callbacks in amqplib's confirm channels, at least by providing extra bits of library.

[pdelanauze]

Would be interesting to have promises around ConfirmChannels, it's how we expected it to work , but the callbacks are definitely fine for now

[squaremo]

Would be interesting to have promises around ConfirmChannels, it's how we expected it to work , but the callbacks are definitely fine for now

OK, good to know. One difficulty with returning promises is that then there's no way to communicate flow control (as with streams). Ironically that's one way a waitForConfirms API is better; maybe using a promise there is possible:

ch.sendToQueue(...);
var await = ch.awaitConfirms();
await.then(function(..) {});

The point being that since it's a promise, you can do this after each publish with no especial penalty.

[squaremo]

There's a first pass at waitForConfirms in master branch now (and an example of using it in the examples directory).

[squaremo]

@pdelanauze Did you have a chance to try waitForConfirms() out?

[ScOut3R]

I'm trying to use a ConfirmChannel with sendToQueue but the callback isn't called. Is that normal?

ch.sendToQueue(q, new Buffer(msg), { persistent: true }, function (error, ok) {
          if (error) {
            console.log('Message dropped!');
          } else {
            console.log('Message OK');
          }
        });

I've checked the waitForConfirms example and you are using a callback there with #publish. I assume the callback is working there.

[squaremo]

@ScOut3R I just tried your example, and it appears to work OK.

I note that the example doesn't show how you obtain ch. Did you create the channel with Connection#createConfirmChannel()? If you did, would you post a minimal test case that fails as another issue -- thanks!

[squaremo]

NB #103 demonstrates a use for waitForConfirms()

[gustaf-advisa]

Got here by having the same assumption as @pdelanauze by assuming a promise interface for #publish().

How can I write an application that safely does:

setupAmqp().then(function() {
    channel.publish(ex, 'a.b', msg);
}).then(function() {
    connection.close(); // I'm maybe running this, maybe I'm just terminating the program abruptly here for whatever reason
}).then(function() {
    process.exit();
});

...without publish returning a promise? Should I have know implementation details, such as:

• Internally, amqp.node is synchronous and blocking, so publish()+process.exit() is safe (which is what I frightfully derive from the discussion above)
• The AMQP protocol ensures messages a published before I do connection.close() because it handshakes the disconnection or similar?

Is any of the above even true? And why am I forced to know this?

To me, publish() returning a promise has nothing to do with having channel.waitForConfirms().
The first means that the message I'm about to send hasn't been fired and forgotten, but actually confirmed. Or this promise fails for some valid reason, like there was no queue receiving a mandatory message.
The second means that all outgoing messages are confirmed (w success or failure correspondingly).

Any other implementation than a promisified publish() inflicts protocol/implementation details/knowledge upon me as a user of the library to feel safe that the message has been taken care of.

[squaremo]

Should I have know implementation details, such as:

Internally, amqp.node is synchronous and blocking, so publish()+process.exit() is safe (which is what I frightfully derive from the discussion above)
The AMQP protocol ensures messages a published before I do connection.close() because it handshakes the disconnection or similar?

Is any of the above even true? And why am I forced to know this?

No, neither of those are true.

Any other implementation than a promisified publish() inflicts protocol/implementation details/knowledge upon me as a user of the library to feel safe that the message has been taken care of.

Yes, you have to know about the protocol you're using. Is that really to your detriment? It's like having to know that you need to commit a database transaction before you exit your program. Furthermore, returning a promise from publish doesn't remove that requirement -- you would still have to know to synchronise on that promise (i.e., what it means in terms of the protocol) in specific circumstances.

I appreciate that you would like to be shielded from awkward details. I've explained above, and in #62, why it works the way it works. Summary: the return value from publish, which is inherently an asynchronous operation, indicates the writability of the channel, and is crucial for respecting flow control.

[gustaf-advisa]

@squaremo i do get the details, and unfortunately I only know the concepts of the protocol, not its details, and this seems like an implementation detail. This is why I brought up these two issues, which, since both are wrong, show how (unnecessary) hard it is to use publish(). The problem is an API that is almost entirely promise based, but one specific function is not, and it's one of those you really want to know and distinguish the results of, whether it succeeded or failed.

For example, what happens with mandatory messages that are being, according to the documentation, "sent back"? Can I detect this from the return value of publish, i.e. for a particular message? How do I handle different errors for different messages? What happens if there's an error in the message I'm trying to send, or the routing-key is invalid? In an API that is officially "promise based", I expect to get a rejected promise back on failures. The only viable solution I can see from the current interface is that publish may throw if something fails. This is not consistent with the other functions and makes a big difference when calling these functions from callbacks:

void result = null;
setTimeout(function() {
    result = client.publish("ex", {invalid: "routing key"}, data);
}, 500);

If publish does not return a promise, which people obviously expect (I know the documentation says otherwise), the above code will throw an error to the event loop if publish fails. Or should I read errors elsewhere?

Also, how do I deal with errors for specific messages? How do I know the message has actually been handled by the server?

If you want to keep the current API, which is reasonable, I would like to see something like a confirmedPublish() which returns a promise only resolving when the message is safe on the server.

Or maybe I am completely missing something :)

[squaremo]

Or maybe I am completely missing something :)

Not really, just some details of the AMQP protocol. I agree with your sentiment that APIs should be as unsurprising as possible. Let me fill in the details and see if I can persuade you that the API is the least surprising it can be, given AMQP's delightful quirks.

The problem is an API that is almost entirely promise based, but one specific function is not

Actually it's not just publish: none of ack, nack, and reject expect an answer from the server, and thus don't return promises.

In addition, it's only publish on a ConfirmChannel that gets a response from the server, and it isn't serialised with other responses (e.g., the confirmation can come back after some other operations have happened, oh and it may be batched with other confirmations).

So there's a consistent rule: RPCs return promises (or accept callbacks in the callback-oriented API), non-RPCs don't.

For example, what happens with mandatory messages that are being, according to the documentation, "sent back"?

They get returned in a set of "return" frames, at some indefinite time after the publish (and possibly other operations). The returned message is a copy of the original, but is treated the same as a new delivery, so is not in general able to be correlated (by the library) with the original message.

That all might be OK, except that in general there is no success indication. A confirmation does mean that if the message was going to be returned, it will have been returned already, so you could in principle always use a confirm channel, wait for the confirmation for each message, and make sure there was no returned message in the meantime. It would be dreadfully slow, especially if you're using persistent messages.

Can I detect this from the return value of publish, i.e. for a particular message?

No, publish does not get a synchronous response from the server (not even in the case of confirmation); the return value in the API indicates whether that write buffer has room left for more messages.

By the way, this write buffer thing may seem like an implementation detail, but it's pretty important. It's quite easy, if you're not paying attention to flow control, to bring Node.JS to its knees, jamming e.g. the contents of a large file into AMQP messages.

How do I handle different errors for different messages?

There's really only one error that you can make with publish, and that's to address it to an exchange that doesn't exist. In this case, the server closes the channel, which is what triggers the 'error' event. Unless you use confirmations and synchronise on every single message, there's no way to pin it to a specific message (although you can look at the error and see which was the erroneous exchange).

In this sense, it's consistent with all other operations, which also close the channel when there's an error. It's just that those are expecting an answer, so it's natural to supply the error in lieu of the answer.

Returns (as in undeliverable messages) are not considered errors (in the protocol, and thus in the library).

The channel can also be closed by the server if, say, an operator decides they don't like you. So it might not even happen in response to something you've done.

If you want to keep the current API, which is reasonable, I would like to see something like a confirmedPublish() which returns a promise only resolving when the message is safe on the server.

There's not really a "safe on the server". For instance, a message can go straight to a consumer, not resting in a queue (let alone being written to disk), and be confirmed.

All that aside, yes one probably could write a publishSynchronously that waited for each confirmation to come back (you'd lose the flow control information, but since you're synchronising on every message reaching the server that wouldn't matter). If you want to accomplish the same thing without any library changes though, surely it's easy to write

channel.publish(...); channel.waitForConfims().then(...);

and perhaps treat message returns as errors.

[squaremo]

I'm closing this, since I think I've met the brief of the original issue (with waitForConfirms). I don't mind if anyone wants to add more to the discussion though.

[matt-cook]

After reading the last post, I'm still unclear. Why can't sendToQueue simply return a promise that resolves with the boolean value?

Is this unsafe?

var sendIt = new Promise((fulfill, reject) => {
        myConfirmChannel.sendToQueue(myQueue, myBuffer, {}, (err, res) => {
            if(err){
              reject(err);
            }else{
               fulfill(res);
            }
        });
});

It's counterintuitive that the method supports a callback but can't return a promise.

Edit: to answer my own question: because the return value is immediate and the callback returns later. two values, different times. Got it. I'm just not using the immediate value in this case.

[cyrilchapon]

@matt-cook and others, expanding on @matt-cook answer. I'm currently using a similar implementation

  async send (data, options) {
    this._checkChannel()

    const sendAsPromise = (this.amqp.channel instanceof ConfirmChannel)
      ? new Promise((resolve, reject) => {
        this.amqp.channel.sendToQueue(this.name, data, options,
          (err, ok) => err ? reject(err) : resolve(ok)
        })
      : this.amqp.channel.sendToQueue(this.name, data, options)

    return sendAsPromise
  }

Is that safe ?

More clearly : Can I consider "the callback being called with no err" (on a ConfirmChannel) as safe as "the promise resolve with true" (on a simple Channel) ?