Abort HTTP connection when stopping changesReader (#310)

Author: insidewhy

README.md

@@ -650,7 +650,7 @@ reliable, resumable changes feed follower, then you need the `changesReader`.
 
 There are three ways to start listening to the changes feed:
 
-1. `changesReader.start()` - to listen to changes indefinitely by repeated "long poll" requests. This mode continues to poll for changes forever.
+1. `changesReader.start()` - to listen to changes indefinitely by repeated "long poll" requests. This mode continues to poll for changes until `changesReader.stop()` is called, at which point any active long poll will be canceled.
 2. `changesReader.get()` - to listen to changes until the end of the changes feed is reached, by repeated "long poll" requests. Once a response with zero changes is received, the 'end' event will indicate the end of the changes and polling will stop.
 3. `changesReader.spool()` - listen to changes in one long HTTP request. (as opposed to repeated round trips) - spool is faster but less reliable.
 

lib/changesreader.js

@@ -1,4 +1,5 @@
 const EventEmitter = require('events').EventEmitter
+const AbortController = global.AbortController || require('node-abort-controller').AbortController
 const stream = require('stream')
 const EVENT_BATCH = 'batch'
 const EVENT_CHANGE = 'change'
@@ -103,6 +104,7 @@ class ChangesReader {
     this.qs = {} // extra querystring parameters
     this.selector = null
     this.paused = false
+    this.abortController = null
   }
 
   pause () {
@@ -116,6 +118,9 @@ class ChangesReader {
   // prevent another poll happening
   stop () {
     this.continue = false
+    if (this.abortController) {
+      this.abortController.abort()
+    }
   }
 
   // sleep, promise style
@@ -148,11 +153,13 @@ class ChangesReader {
     const work = async () => {
       do {
         if (!self.paused) {
+          self.abortController = new AbortController()
           // formulate changes feed longpoll HTTP request
           const req = {
             method: 'post',
             db: self.db,
             path: '_changes',
+            signal: self.abortController.signal,
             qs: {
               feed: 'longpoll',
               timeout: self.timeout,
@@ -174,6 +181,7 @@ class ChangesReader {
           // make HTTP request to get up to batchSize changes from the feed
           try {
             const data = await self.request(req)
+            self.abortController = null
             delay = 0
 
             // update the since state

lib/nano.js

@@ -129,6 +129,16 @@ module.exports = exports = function dbScope (cfg) {
       statusCode: statusCode
     }, response.headers)
     if (!response.status) {
+      if (axios.isCancel(response)) {
+        if (resolve) {
+          resolve('canceled')
+        }
+        if (callback) {
+          callback(null, 'canceled', responseHeaders)
+        }
+        return
+      }
+
       log({ err: 'socket', body: body, headers: responseHeaders })
       if (reject) {
         reject(new Error(`error happened in your connection. Reason: ${response.message}`))
@@ -267,6 +277,10 @@ module.exports = exports = function dbScope (cfg) {
     // https://github.com/mikeal/request#requestjar
     const isJar = opts.jar || cfg.jar || (cfg.requestDefaults && cfg.requestDefaults.jar)
 
+    if (opts.signal) {
+      req.signal = opts.signal
+    }
+
     if (isJar) {
       req.jar = cookieJar
       req.withCredentials = true

package-lock.json

@@ -20,6 +20,7 @@
     "http-cookie-agent": "^1.0.5",
     "@types/tough-cookie": "^4.0.0",
     "axios": "^0.26.1",
+    "node-abort-controller": "^3.0.1",
     "qs": "^6.10.3",
     "tough-cookie": "^4.0.0"
   },
@@ -48,4 +49,4 @@
       "jest"
     ]
   }
-}
+}

package.json

@@ -505,3 +505,29 @@ test('should survive malformed JSON - db.changesReader.start', async () => {
     })
   })
 }, 10000)
+
+test('should cancel HTTP connection as soon as stop is called', async () => {
+  const changeURL = `/${DBNAME}/_changes`
+  nock(COUCH_URL)
+    .post(changeURL)
+    .query({ feed: 'longpoll', timeout: 60000, since: 'now', limit: 100, include_docs: false })
+    .reply(200, { results: [], last_seq: '1-0', pending: 0 })
+    .post(changeURL)
+    .query({ feed: 'longpoll', timeout: 60000, since: '1-0', limit: 100, include_docs: false })
+    .delay(60000)
+    .reply(500)
+  const db = nano.db.use(DBNAME)
+  const cr = db.changesReader.start()
+  await new Promise((resolve, reject) => {
+    cr.on('seq', function (seq) {
+      setTimeout(function () {
+        // give the next http connection a chance to be established
+        db.changesReader.stop()
+      }, 200)
+    })
+
+    cr.on('end', function () {
+      resolve()
+    })
+  })
+})