changesreader /added fields & body support

README.md

@@ -666,7 +666,10 @@ You may supply a number of options when you start to listen to the changes feed:
 | includeDocs | Whether to include document bodies or not | false | e.g. true |
 | wait | For `get`/`start` mode, automatically pause the changes reader after each request. When the the user calls `resume()`, the changes reader will resume.  | false | e.g. true |
 | fastChanges | Adds a seq_interval parameter to fetch changes more quickly | false           | true                             |   |
-| selector | Filters the changes feed with the supplied Mango selector | {"name":"fred}           | null                             |   |
+| selector | Filters the changes feed with the supplied Mango selector |  null          | { "name": "fred" } |   |
+| fields | Filters the changes feed with the supplied fields list | null           | [ 'id', 'seq' ]                             |   |
+| qs | Additional query string parameters | null           | { "filter": "_selector" }                            |   |
+| body | Additional body parameters | null           | { "fields": [ "id", "seq" ] }                            |   |
 | timeout | The number of milliseconds a changes feed request waits for data| 60000         | 10000     
 
 The events it emits are as follows:s

lib/changesreader.js

@@ -101,7 +101,9 @@ class ChangesReader {
     this.stopOnEmptyChanges = false // whether to stop polling if we get an empty set of changes back
     this.continue = true // whether to poll again
     this.qs = {} // extra querystring parameters
+    this.body = {} // extra body parameters
     this.selector = null
+    this.fields = null
     this.paused = false
   }
 
@@ -165,11 +167,17 @@ class ChangesReader {
           if (self.fastChanges) {
             req.qs.seq_interval = self.batchSize
           }
-          if (self.selector) {
+          if (self.selector || self.fields) {
             req.qs.filter = '_selector'
-            req.body.selector = self.selector
+            if (self.selector) {
+              req.body.selector = self.selector
+            }
+            if (self.fields) {
+              req.body.fields = self.fields
+            }
           }
           Object.assign(req.qs, opts.qs)
+          Object.assign(req.body, opts.body)
 
           // make HTTP request to get up to batchSize changes from the feed
           try {

lib/nano.d.ts

@@ -257,18 +257,22 @@ declare namespace nano {
     wait?: boolean;
     /** additional query string parameters */
     qs?: object;
+    /** additional body parameters */
+    body?: object;
     /** a MangoSelector defining the slice of the changes feed to return */
     selector?: MangoSelector;
+    /** a MangoSelector defining changes feed fields to return */
+    fields?: string[];
   }
 
   /** ChangesReader functions */
   interface ChangesReaderScope {
     /** fetch changes forever */
-    start(opts: ChangesReaderOptions): EventEmitter;
+    start(opts?: ChangesReaderOptions): EventEmitter;
     /** fetch changes and stop when an empty batch is received */
-    get(opts: ChangesReaderOptions): EventEmitter;
+    get(opts?: ChangesReaderOptions): EventEmitter;
     /** spool the change in one long feed, instead of batches */
-    spool(opts: ChangesReaderOptions): EventEmitter;
+    spool(opts?: ChangesReaderOptions): EventEmitter;
     /** stop consuming the changes feed */
     stop(): void;
     /** pause consuming the changes feed */

test/document.changesreader.test.js

@@ -41,6 +41,58 @@ test('should be able to follow changes feed - db.changesReader.start', async ()
   })
 })
 
+test('should be able to filter changes feed via fields param - db.changesReader.start', async () => {
+  const changeURL = `/${DBNAME}/_changes`
+  const fields = ['id', 'seq']
+  const changes = [
+    { seq: null, id: '1' }
+  ]
+  nock(COUCH_URL)
+    .post(changeURL, { fields })
+    .query({ filter: '_selector', feed: 'longpoll', timeout: 60000, since: 'now', limit: 100, include_docs: false })
+    .reply(200, { results: changes, last_seq: '1-0', pending: 0 })
+    .post(changeURL)
+    .delay(2000)
+    .reply(500)
+  const db = nano.db.use(DBNAME)
+  const cr = db.changesReader.start({
+    fields
+  })
+  return new Promise((resolve, reject) => {
+    cr.on('change', function (change) {
+      expect(change).toStrictEqual(changes[0])
+      db.changesReader.stop()
+      resolve()
+    })
+  })
+})
+test('should be able to filter changes feed via body and qs params - db.changesReader.start', async () => {
+  const changeURL = `/${DBNAME}/_changes`
+  const fields = ['id', 'seq']
+  const changes = [
+    { seq: null, id: '1' }
+  ]
+  nock(COUCH_URL)
+    .post(changeURL, { fields })
+    .query({ filter: '_selector', feed: 'longpoll', timeout: 60000, since: 'now', limit: 100, include_docs: false })
+    .reply(200, { results: changes, last_seq: '1-0', pending: 0 })
+    .post(changeURL)
+    .delay(2000)
+    .reply(500)
+  const db = nano.db.use(DBNAME)
+  const cr = db.changesReader.start({
+    qs: { filter: '_selector' },
+    body: { fields }
+  })
+  return new Promise((resolve, reject) => {
+    cr.on('change', function (change) {
+      expect(change).toStrictEqual(changes[0])
+      db.changesReader.stop()
+      resolve()
+    })
+  })
+})
+
 test('should respect the fastChanges flag - db.changesReader.start', async () => {
   const changeURL = `/${DBNAME}/_changes`
   nock(COUCH_URL)