Recieved messages automatically ack-ed #20
Description
Hi,
With the way the Listener class is configured, upon receiving a message it immediately performs an ack() on the message before the message handler has been called:
const onMessage = (msg) => {
msg.ack();
return onMessageCallback(msg);
};
This is effectively telling the pubsub queue that the message has been successfully handled/processed, and that the topic can move on.
However most use cases of message processing involve an asynchronous activity (e.g. writing to another queue or database). By having the ack() performed immediately, this means that a failure during processing the item (e.g. database down, node process crashed) would effectively mean the message could have been lost, as the library has already performed the ack() and the pubsub queue has moved on.
By immediately ack-ing, it also removes any flow control from the client. The Listener is saying the client is ready for more, but the client could have an ever growing number of async message processing jobs still in-flight.
In these scenarios, I would have thought the ack() call is surely more appropriate after the message processing has completed, but the Listener class does not provide such a way to do this.
My suggestions for possible implementations are:
- async or callback message handlers that upon successful completion,
onMessagethen acks - leave ack-ing up to the
onMessageCallbackfunction (theack()method is available as the msg object is passed through as the param)
With such a setup, it removes the risk of the topic moving on without the client successfully performing its activity on the message.
A simple implementation with a callback approach could look something like:
const onMessage = (msg) => {
onMessageCallback(msg, err => {
if (!err) {
msg.ack();
} else { // some failure handling
console.err(err);
msg.nack();
}
});
};
The API could be updated in a way to make it backward compatible (e.g. additional param indicating it is async/wants to control message ack-ing).
Does this seem reasonable?