refactor: more readable code for consuming message stream

License: MIT
Signed-off-by: Alan Shaw <alan.shaw@protocol.ai>

Author: Alan Shaw

src/pubsub/subscribe.js

@@ -36,32 +36,38 @@ module.exports = configure(({ fetch, apiAddr, apiPath, headers }) => {
       throw err
     }
 
-    // eslint-disable-next-line no-console
-    const onError = options.onError || (err => console.error(err))
+    readMessages(ndjson(toIterable(res.body)), {
+      onMessage: handler,
+      onEnd: () => subsTracker.unsubscribe(topic, handler),
+      onError: options.onError
+    })
+  }
+})
+
+async function readMessages (msgStream, { onMessage, onEnd, onError }) {
+  // eslint-disable-next-line no-console
+  onError = onError || (err => console.error(err))
 
-    ;(async () => {
+  try {
+    for await (const msg of msgStream) {
       try {
-        for await (const msg of ndjson(toIterable(res.body))) {
-          try {
-            handler({
-              from: bs58.encode(Buffer.from(msg.from, 'base64')).toString(),
-              data: Buffer.from(msg.data, 'base64'),
-              seqno: Buffer.from(msg.seqno, 'base64'),
-              topicIDs: msg.topicIDs
-            })
-          } catch (err) {
-            onError(explain(err, 'Failed to parse pubsub message'), false) // Not fatal
-          }
-        }
+        onMessage({
+          from: bs58.encode(Buffer.from(msg.from, 'base64')).toString(),
+          data: Buffer.from(msg.data, 'base64'),
+          seqno: Buffer.from(msg.seqno, 'base64'),
+          topicIDs: msg.topicIDs
+        })
       } catch (err) {
-        // FIXME: In testing with Chrome, err.type is undefined (should not be!)
-        // Temporarily use the name property instead.
-        if (err.type !== 'aborted' && err.name !== 'AbortError') {
-          onError(err, true) // Fatal
-        }
-      } finally {
-        subsTracker.unsubscribe(topic, handler)
+        onError(explain(err, 'Failed to parse pubsub message'), false) // Not fatal
       }
-    })()
+    }
+  } catch (err) {
+    // FIXME: In testing with Chrome, err.type is undefined (should not be!)
+    // Temporarily use the name property instead.
+    if (err.type !== 'aborted' && err.name !== 'AbortError') {
+      onError(err, true) // Fatal
+    }
+  } finally {
+    onEnd()
   }
-})
+}