dowjones/dj-dna-streams-javascript

JavaScript library for the DNA Streams API.

dj-dna-streams-javascript's Issues

No way to shut down the Listener

Occasionally processes need to be restarted, but the Listener class appears to expose no way of cleanly shutting down. Forcibly shutting down introduces a risk of losing messages that are still being processed by the client, but have been ack-ed by the listener immediately after entering the onMessage function.

Having better control of ack-ing would help, but having say a shutdown method on the Listener that could be called after an appropriate signal has been received would be ideal.

Recieved messages automatically ack-ed

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, onMessage then acks
• leave ack-ing up to the onMessageCallback function (the ack() 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?