Change examples and tests to pipeline-based async iteration

This avoids recommending a pattern that exposes a footgun - pipe() needs
to have independent error-handling to avoid possible triggering an
unhandled error.

Author: danmactough

README.md

@@ -65,21 +65,26 @@ feedparser.on('readable', function () {
 You can also consume feeds using async iteration:
 
 ```js
-
 var FeedParser = require('feedparser');
 var fetch = require('node-fetch');
+var pipeline = require('stream/promises').pipeline;
 
 async function main() {
-  var res = await fetch('http://somefeedurl.xml');
+  var res = await fetch('http://someurl.site/rss.xml');
   if (res.status !== 200) throw new Error('Bad status code');
 
   var feedparser = new FeedParser([options]);
-  res.body.pipe(feedparser);
 
   try {
-    for await (var item of feedparser) {
-      console.log(item.title);
-    }
+    await pipeline(
+      res.body,
+      feedparser,
+      async function (feedparserIterable) {
+        for await (var item of feedparserIterable) {
+          console.log(item.title);
+        }
+      }
+    )
   } catch (err) {
     console.error(err);
   }

test/async-iterator.js

@@ -1,28 +1,31 @@
 var FeedParser = require('..');
+var PassThrough = require('stream').PassThrough;
+// We're using this form so we can run tests on older Node versions that don't have stream.promises.pipeline
+var pipeline = require('util').promisify(require('stream').pipeline);
 
 describe('async iterator usage', function () {
   it('should work as an async iterator', async function () {
     var feedparser = new FeedParser();
     var feed = __dirname + '/feeds/rss2sample.xml';
     var items = [];
 
-    fs.createReadStream(feed).pipe(feedparser);
-
-    for await (const item of feedparser) {
-      items.push(item);
-    }
+    await pipeline(fs.createReadStream(feed), feedparser, async function (fpIterable) {
+      for await (const item of fpIterable) {
+        items.push(item);
+      }
+    });
 
     assert.equal(items.length, 4);
   });
 
   it('should surface errors via try/catch', async function () {
     var feedparser = new FeedParser();
     var feed = __dirname + '/feeds/notafeed.html';
-    fs.createReadStream(feed).pipe(feedparser);
-
     var caught = null;
     try {
-      for await (const item of feedparser) {} // eslint-disable-line no-empty, no-unused-vars
+      await pipeline(fs.createReadStream(feed), feedparser, async function (fpIterable) {
+        for await (const item of fpIterable) {} // eslint-disable-line no-empty, no-unused-vars
+      });
     } catch (err) {
       caught = err;
     }
@@ -31,32 +34,66 @@ describe('async iterator usage', function () {
     assert.equal(caught.message, 'Not a feed');
   });
 
+  it('should catch errors after a delayed iteration start', async function () {
+    var feedparser = new FeedParser();
+    var source = new PassThrough();
+    var items = [];
+    var caught = null;
+    var uncaught = null;
+    function onUncaught(err) {
+      uncaught = err;
+    }
+    process.prependOnceListener('uncaughtException', onUncaught);
+
+    source.end('not a feed');
+
+    await new Promise(setImmediate);
+
+    try {
+      await pipeline(source, feedparser, async function (fpIterable) {
+        for await (const item of fpIterable) {
+          items.push(item.title);
+        }
+      });
+    } catch (err) {
+      caught = err;
+    } finally {
+      process.removeListener('uncaughtException', onUncaught);
+      assert.equal(uncaught, null);
+      assert.ok(caught instanceof Error);
+      assert.equal(caught.message, 'Not a feed');
+      assert.equal(items.length, 0);
+    }
+  });
+
   describe('resume_saxerror behavior', function () {
     var feed = __dirname + '/feeds/saxerror.xml';
 
     it('should continue iterating past SAX errors by default (resume_saxerror: true)', async function () {
       var feedparser = new FeedParser({ strict: true });
-      fs.createReadStream(feed).pipe(feedparser);
       var items = [];
 
-      for await (const item of feedparser) {
-        items.push(item.title);
-      }
+      await pipeline(fs.createReadStream(feed), feedparser, async function (fpIterable) {
+        for await (const item of fpIterable) {
+          items.push(item.title);
+        }
+      });
 
       assert.equal(items.length, 3);
       assert.deepEqual(items, ['Good Item', 'Bad Item', 'Item After Error']);
     });
 
     it('should throw on SAX errors when (resume_saxerror: false)', async function () {
       var feedparser = new FeedParser({ strict: true, resume_saxerror: false });
-      fs.createReadStream(feed).pipe(feedparser);
       var items = [];
 
       var caught = null;
       try {
-        for await (const item of feedparser) {
-          items.push(item.title);
-        }
+        await pipeline(fs.createReadStream(feed), feedparser, async function (fpIterable) {
+          for await (const item of fpIterable) {
+            items.push(item.title);
+          }
+        });
       } catch (err) {
         caught = err;
       }