added more protections for errors in callback

Listener.js

@@ -40,10 +40,10 @@ class Listener {
    * Leave as null or undefined if you
    * want to use the default.
    */
-  listen(onMessageCallback, subscription) {
+  listen(onMessageCallback, subscription, userErrorHandling = false) {
     return this.apiService.getStreamingCredentials().then((credentials) => {
       this.initialize(credentials);
-      this.readyListener(onMessageCallback, subscription);
+      this.readyListener(onMessageCallback, subscription, userErrorHandling);
       return true;
     }).catch((err) => {
       if (err.message) {
@@ -55,7 +55,7 @@ class Listener {
     });
   }
 
-  readyListener(onMessageCallback, subscriptionId) {
+  readyListener(onMessageCallback, subscriptionId, userErrorHandling) {
     const sub = subscriptionId || this.defaultSubscriptionId;
 
     if (!sub || sub.length <= 0) {
@@ -66,11 +66,31 @@ class Listener {
 
     console.log(`Listening to subscription: ${subscriptionFullName}`);
 
-    const onMessage = (msg) => {
-      onMessageCallback(msg);
-      msg.ack();
+    const onMessageTryCatch = (msg) => {
+      try {
+        onMessageCallback(msg);
+        msg.ack();
+      } catch (err) {
+        console.error(`Error from callback: ${err}\n`);
+        msg.nack();
+        throw err;
+      }
     };
 
+    const onMessageUserHandling = (msg) => {
+      onMessageCallback(msg, err => {
+        if (err) {
+          console.error(`Error from callback: ${err}`);
+          msg.nack();
+          throw err;
+        } else {
+          msg.ack();
+        }
+      });
+    }
+
+    const onMessage = (userErrorHandling) ? onMessageUserHandling : onMessageTryCatch;
+
     const pubsubSubscription = this.pubsubClient.subscription(subscriptionFullName);
 
     this.apiService.getAccountInfo().then(accountInfo =>

README.md

@@ -135,3 +135,24 @@ The following is some very basic code. Use it to listen to a DNA subscription. I
   const listener = new Listener();
   listener.listen(onMessageCallback);
 ~~~~
+
+###### Error Handling
+
+If your callback fails, the message will be nack'd and the listener will rethrow the error. If you wish to write your own error handling for callbacks then set the `userErrorHandling` parameter to true. This allows you to use an error handler callback to force the callback handler to nack messages. The following is a very basic example illustrating how this may work.
+
+~~~~
+  var Listener = require('dj-dna-streaming-javascript').Listener;
+
+  var onMessageCallback = function((msg, handleErr) {
+     let err = null;
+     try {
+      console.log('One incoming message:' + JSON.stringify(msg.data));
+      } catch (e) {
+        err = e
+      };
+    handleErr(err)
+  };
+
+  const listener = new Listener();
+  listener.listen(onMessageCallback, my_subscription_id, true);
+~~~~

demo_error_handling/index.js

@@ -0,0 +1,22 @@
+const Listener = require('../Listener');
+
+const quietDemo = process.env.QUIET_DEMO;
+
+const onMessageCallback = (msg, handleErr) => {
+  let err = null;
+  try {
+    let message = `${msg.data}`;
+    if (quietDemo === true.toString()) {
+      message = `${message.substring(0, 50)} ...`;
+    }
+    console.log(`Received message: ${message}\n`);
+  } catch (e) {
+   err = e
+  };
+
+  handleErr(err)
+};
+
+const listener = new Listener();
+
+listener.listen(onMessageCallback, null, true);