Repository: influxdata/influxdb-client-js
Pull request #588 (2022-09-27 07:33:23)

feat(core): allow to pause/resume query result observers
This PR allows pausing query results by returning `false` in the `next` callback of `queryApi.queryLines` and `queryApi.queryRows` consumer. The consumer must define `useResume` callback that receives a function that resumes receiving data when called. This feature adds a low-level (node.js stream) style of lossless backpressure to query result processing via communication observer.

Method signatures changed, the `next` callback of `CommunicationObserver` and `FluxResultObserver` can newly return also a boolean value, `false` value will stop processing. All these changes are backward compatible, no change is required for the existing code.

## Checklist

<!-- Checkboxes below this note can be erased if not applicable to your Pull Request. -->

- [x] CHANGELOG.md updated
- [x] Rebased/mergeable
- [x] A test has been added if appropriate
- [x] `yarn test` completes successfully
- [x] Commit messages are [conventional](https://www.conventionalcommits.org/en/v1.0.0/)



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9a5288fc..ae8d1ddd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,7 @@
 ### Features
 
 1. [#571](https://github.com/influxdata/influxdb-client-js/pull/571): Regenerate APIs from swagger.
+1. [#588](https://github.com/influxdata/influxdb-client-js/pull/588): Allow to pause/resume flux query result communication observer.
 
 ### Bug Fixes
 
diff --git a/packages/core/src/impl/browser/FetchTransport.ts b/packages/core/src/impl/browser/FetchTransport.ts
index 7092e41a..93593d12 100644
--- a/packages/core/src/impl/browser/FetchTransport.ts
+++ b/packages/core/src/impl/browser/FetchTransport.ts
@@ -65,12 +65,19 @@ export default class FetchTransport implements Transport {
     const observer = completeCommunicationObserver(callbacks)
     let cancelled = false
     let signal = (options as any).signal
+    let pausePromise: Promise<void> | undefined
+    const resumeQuickly = () => {}
+    let resume = resumeQuickly
     if (callbacks && callbacks.useCancellable) {
       const controller = new AbortController()
       if (!signal) {
         signal = controller.signal
         options = {...(options as object), ...signal} as SendOptions
       }
+      // resume data reading so that it can exit on abort signal
+      signal.addEventListener('abort', () => {
+        resume()
+      })
       callbacks.useCancellable({
         cancel() {
           cancelled = true
@@ -126,8 +133,29 @@ export default class FetchTransport implements Transport {
             const reader = response.body.getReader()
             let chunk: ReadableStreamReadResult<Uint8Array>
             do {
+              if (pausePromise) {
+                await pausePromise
+              }
+              if (cancelled) {
+                break
+              }
               chunk = await reader.read()
-              observer.next(chunk.value)
+              if (observer.next(chunk.value) === false) {
+                const useResume = observer.useResume
+                if (!useResume) {
+                  const msg = 'Unable to pause, useResume is not configured!'
+                  await reader.cancel(msg)
+                  return Promise.reject(new Error(msg))
+                }
+                pausePromise = new Promise((resolve) => {
+                  resume = () => {
+                    resolve()
+                    pausePromise = undefined
+                    resume = resumeQuickly
+                  }
+                  useResume(resume)
+                })
+              }
             } while (!chunk.done)
           } else if (response.arrayBuffer) {
             const buffer = await response.arrayBuffer()
diff --git a/packages/core/src/impl/completeCommunicationObserver.ts b/packages/core/src/impl/completeCommunicationObserver.ts
index 483d3b78..3d795bc9 100644
--- a/packages/core/src/impl/completeCommunicationObserver.ts
+++ b/packages/core/src/impl/completeCommunicationObserver.ts
@@ -1,18 +1,24 @@
 import {CommunicationObserver, Headers} from '../results'
 
+type CompleteObserver = Omit<
+  Required<CommunicationObserver<any>>,
+  'useCancellable' | 'useResume'
+> &
+  Pick<CommunicationObserver<any>, 'useResume' | 'useCancellable'>
+
 export default function completeCommunicationObserver(
   callbacks: Partial<CommunicationObserver<any>> = {}
-): Omit<Required<CommunicationObserver<any>>, 'useCancellable'> {
+): CompleteObserver {
   let state = 0
-  const retVal = {
-    next: (data: any): void => {
+  const retVal: CompleteObserver = {
+    next: (data: any): void | boolean => {
       if (
         state === 0 &&
         callbacks.next &&
         data !== null &&
         data !== undefined
       ) {
-        callbacks.next(data)
+        return callbacks.next(data)
       }
     },
     error: (error: Error): void => {
@@ -35,5 +41,11 @@ export default function completeCommunicationObserver(
         callbacks.responseStarted(headers, statusCode)
     },
   }
+  if (callbacks.useCancellable) {
+    retVal.useCancellable = callbacks.useCancellable.bind(callbacks)
+  }
+  if (callbacks.useResume) {
+    retVal.useResume = callbacks.useResume.bind(callbacks)
+  }
   return retVal
 }
diff --git a/packages/core/src/impl/node/NodeHttpTransport.ts b/packages/core/src/impl/node/NodeHttpTransport.ts
index da9aab74..34054f78 100644
--- a/packages/core/src/impl/node/NodeHttpTransport.ts
+++ b/packages/core/src/impl/node/NodeHttpTransport.ts
@@ -26,8 +26,13 @@ const emptyBuffer = Buffer.allocUnsafe(0)
 
 class CancellableImpl implements Cancellable {
   private cancelled = false
+  public resume?: () => void
   cancel(): void {
     this.cancelled = true
+    if (this.resume) {
+      this.resume()
+      this.resume = undefined
+    }
   }
   isCancelled(): boolean {
     return this.cancelled
@@ -329,7 +334,23 @@ export class NodeHttpTransport implements Transport {
           if (cancellable.isCancelled()) {
             res.resume()
           } else {
-            listeners.next(data)
+            if (listeners.next(data) === false) {
+              // pause processing, the consumer signalizes that
+              // it is not able to receive more data
+              if (!listeners.useResume) {
+                listeners.error(
+                  new Error('Unable to pause, useResume is not configured!')
+                )
+                res.resume()
+                return
+              }
+              res.pause()
+              const resume = () => {
+                res.resume()
+              }
+              cancellable.resume = resume
+              listeners.useResume(resume)
+            }
           }
         })
         responseData.on('end', listeners.complete)
diff --git a/packages/core/src/results/CommunicationObserver.ts b/packages/core/src/results/CommunicationObserver.ts
index ff3ee14d..013bdf41 100644
--- a/packages/core/src/results/CommunicationObserver.ts
+++ b/packages/core/src/results/CommunicationObserver.ts
@@ -22,8 +22,10 @@ export interface CommunicationObserver<T> {
   /**
    * Data chunk received, can be called multiple times.
    * @param data - data
+   * @returns when `false` value is returned and {@link CommunicationObserver.useResume} is defined,
+   * future calls to `next` are paused until resume is called.
    */
-  next(data: T): void
+  next(data: T): void | boolean
   /**
    * Communication ended with an error.
    */
@@ -40,4 +42,11 @@ export interface CommunicationObserver<T> {
    * Setups cancelllable for this communication.
    */
   useCancellable?: (cancellable: Cancellable) => void
+  /**
+   * Setups a callback that resumes reading of next data, it is called whenever
+   * {@link CommunicationObserver.next} returns `false`.
+   *
+   * @param resume - a function that will resume reading of next data when called
+   */
+  useResume?: (resume: () => void) => void
 }
diff --git a/packages/core/src/results/FluxResultObserver.ts b/packages/core/src/results/FluxResultObserver.ts
index cf5badde..c0b3611c 100644
--- a/packages/core/src/results/FluxResultObserver.ts
+++ b/packages/core/src/results/FluxResultObserver.ts
@@ -7,8 +7,12 @@ import {FluxTableMetaData} from './FluxTableMetaData'
 export interface FluxResultObserver<T> {
   /**
    * Inform about a next record in a table.
+   * @param row - flux result
+   * @param tableMeta - actual table metata for the row supplied
+   * @returns when `false` value is returned and {@link FluxResultObserver.useResume} is defined,
+   * future calls to `next` are paused until resume is called.
    */
-  next(row: T, tableMeta: FluxTableMetaData): void
+  next(row: T, tableMeta: FluxTableMetaData): void | boolean
   /**
    * Signalizes processing error.
    */
@@ -21,4 +25,12 @@ export interface FluxResultObserver<T> {
    * Setups cancellable that can abort flux result processing.
    */
   useCancellable?: (cancellable: Cancellable) => void
+
+  /**
+   * Setups a callback that resumes reading of next data, it is called whenever
+   * {@link FluxResultObserver.next} returns `false`.
+   *
+   * @param resume - a function that will resume reading of next data when called
+   */
+  useResume?: (resume: () => void) => void
 }
diff --git a/packages/core/src/results/chunksToLines.ts b/packages/core/src/results/chunksToLines.ts
index dc6e52cf..4a6e07ed 100644
--- a/packages/core/src/results/chunksToLines.ts
+++ b/packages/core/src/results/chunksToLines.ts
@@ -17,13 +17,17 @@ export function chunksToLines(
   let previous: Uint8Array | undefined
   let finished = false
   let quoted = false
+  let paused = false
+  let resumeChunks: (() => void) | undefined
 
   function bufferReceived(chunk: Uint8Array): void {
     let index: number
     let start = 0
     if (previous) {
+      // inspect the whole remaining data upon empty chunk
+      // empty chunk signalizes to restart of receiving
+      index = chunk.length === 0 ? 0 : (previous as Uint8Array).length
       chunk = chunks.concat(previous, chunk)
-      index = (previous as Buffer).length
     } else {
       index = 0
     }
@@ -37,29 +41,50 @@ export function chunksToLines(
           if (finished) {
             return
           }
-          target.next(chunks.toUtf8String(chunk, start, end))
+          paused = target.next(chunks.toUtf8String(chunk, start, end)) === false
           start = index + 1
+          if (paused) {
+            break
+          }
         }
       } else if (c === 34 /* " */) {
         quoted = !quoted
       }
       index++
     }
-    if (start < index) {
-      previous = chunks.copy(chunk, start, index)
+    if (start < chunk.length) {
+      previous = chunks.copy(chunk, start, chunk.length)
     } else {
       previous = undefined
     }
+    if (paused) {
+      if (target.useResume) {
+        target.useResume(() => {
+          paused = false
+          bufferReceived(new Uint8Array(0))
+        })
+        return
+      }
+      retVal.error(new Error('Unable to pause, useResume is not configured!'))
+      paused = false // consume remaining data
+    }
+    if (resumeChunks) {
+      resumeChunks()
+      resumeChunks = undefined
+    }
   }
 
-  return {
-    next(chunk: Uint8Array): void {
-      if (finished) return
-      try {
-        bufferReceived(chunk)
-      } catch (e) {
-        this.error(e as Error)
+  const retVal: CommunicationObserver<Uint8Array> = {
+    next(chunk: Uint8Array): boolean {
+      if (!finished) {
+        try {
+          bufferReceived(chunk)
+          return !paused
+        } catch (e) {
+          this.error(e as Error)
+        }
       }
+      return true
     },
     error(error: Error): void {
       if (!finished) {
@@ -76,21 +101,27 @@ export function chunksToLines(
         target.complete()
       }
     },
-    useCancellable(cancellable: Cancellable): void {
-      if (target.useCancellable) {
-        // eslint-disable-next-line @typescript-eslint/no-this-alias
-        const self = this
+  }
+  if (target.useCancellable) {
+    retVal.useCancellable = (cancellable: Cancellable) => {
+      target.useCancellable &&
         target.useCancellable({
           cancel(): void {
             cancellable.cancel()
             previous = undefined // do not emit more lines
-            self.complete()
+            retVal.complete()
           },
           isCancelled(): boolean {
             return cancellable.isCancelled()
           },
         })
-      }
-    },
+    }
   }
+  if (target.useResume) {
+    retVal.useResume = (x: () => void) => {
+      resumeChunks = x
+    }
+  }
+
+  return retVal
 }
diff --git a/packages/core/src/results/linesToTables.ts b/packages/core/src/results/linesToTables.ts
index 6ff4acdf..e1c81c13 100644
--- a/packages/core/src/results/linesToTables.ts
+++ b/packages/core/src/results/linesToTables.ts
@@ -1,5 +1,4 @@
 import {CommunicationObserver} from './CommunicationObserver'
-import {Cancellable} from './Cancellable'
 import {LineSplitter} from './LineSplitter'
 import {FluxResultObserver} from './FluxResultObserver'
 import {
@@ -21,11 +20,11 @@ export function linesToTables(
   let expectMeta = true
   let firstColumnIndex = 0
   let lastMeta: FluxTableMetaData
-  return {
+  const retVal: CommunicationObserver<string> = {
     error(error: Error): void {
       consumer.error(error)
     },
-    next(line: string): void {
+    next(line: string): void | boolean {
       if (line === '') {
         expectMeta = true
         columns = undefined
@@ -67,15 +66,20 @@ export function linesToTables(
             }
           }
         } else {
-          consumer.next(values.slice(firstColumnIndex, size), lastMeta)
+          return consumer.next(values.slice(firstColumnIndex, size), lastMeta)
         }
       }
+      return true
     },
     complete(): void {
       consumer.complete()
     },
-    useCancellable(cancellable: Cancellable): void {
-      if (consumer.useCancellable) consumer.useCancellable(cancellable)
-    },
   }
+  if (consumer.useCancellable) {
+    retVal.useCancellable = consumer.useCancellable.bind(consumer)
+  }
+  if (consumer.useResume) {
+    retVal.useResume = consumer.useResume.bind(consumer)
+  }
+  return retVal
 }


Test patch:
diff --git a/packages/core/test/unit/impl/browser/FetchTransport.test.ts b/packages/core/test/unit/impl/browser/FetchTransport.test.ts
index d8ea33b6..edc2b911 100644
--- a/packages/core/test/unit/impl/browser/FetchTransport.test.ts
+++ b/packages/core/test/unit/impl/browser/FetchTransport.test.ts
@@ -1,9 +1,14 @@
 import FetchTransport from '../../../../src/impl/browser/FetchTransport'
 import {expect} from 'chai'
-import {removeFetchApi, emulateFetchApi} from './emulateBrowser'
+import {
+  removeFetchApi,
+  emulateFetchApi,
+  AbortController,
+} from './emulateBrowser'
 import sinon from 'sinon'
-import {SendOptions, Cancellable} from '../../../../src'
+import {SendOptions, Cancellable, CommunicationObserver} from '../../../../src'
 import {CollectedLogs, collectLogging} from '../../../util'
+import {waitForCondition} from '../../util/waitForCondition'
 
 describe('FetchTransport', () => {
   afterEach(() => {
@@ -346,7 +351,7 @@ describe('FetchTransport', () => {
       {
         url: 'customNext_cancelledWithSignal',
         body: [Buffer.from('a'), Buffer.from('b')],
-        signal: {aborted: true},
+        signal: new AbortController(true).signal,
         callbacks: ((): void => {
           const overriden = fakeCallbacks()
           overriden.useCancellable = sinon.spy((c: Cancellable): void => {
@@ -520,6 +525,108 @@ describe('FetchTransport', () => {
       expect(request?.credentials).is.deep.equal('my-val')
     })
   })
+  describe('send.backpressure', () => {
+    it(`it throws an error when paused and useResume is not set`, async () => {
+      emulateFetchApi({body: 'abc'.split('').map(Buffer.from)})
+      const observer: CommunicationObserver<Uint8Array> = {
+        next(_chunk: Uint8Array) {
+          // do not receive more than 1 chunk, but still
+          // there is no useResume callback!
+          return false
+        },
+        error() {},
+        complete(): void {},
+      }
+      const spy = sinon.spy(observer)
+      new FetchTransport({url: '/test'}).send(
+        '/test',
+        '',
+        {
+          method: 'GET',
+        },
+        spy
+      )
+      // wait for error being called
+      await waitForCondition(() => spy.error.callCount === 1)
+      expect(spy.next.callCount).equals(1)
+      expect(spy.error.getCall(0).args[0]?.message).contains(
+        'useResume is not configured!'
+      )
+    })
+    it(`is paused after the first chunk, then cancelled`, async () => {
+      let cancellable: Cancellable | undefined
+      let resume: () => void | undefined
+
+      emulateFetchApi({body: 'abc'.split('').map(Buffer.from)})
+      const observer: CommunicationObserver<Uint8Array> = {
+        next(_chunk: Uint8Array) {
+          return false // do not receive more than 1 chunk
+        },
+        error() {},
+        complete(): void {},
+        useCancellable(c: Cancellable) {
+          cancellable = c
+        },
+        useResume(r) {
+          resume = r
+        },
+      }
+      const spy = sinon.spy(observer)
+
+      new FetchTransport({url: '/test'}).send(
+        '/test',
+        '',
+        {
+          method: 'GET',
+        },
+        spy
+      )
+      // wait for resume being called
+      await waitForCondition(() => cancellable && resume)
+      expect(spy.next.callCount).equals(1)
+      cancellable?.cancel()
+    })
+    it(`is paused after the second chunk and then read fully`, async () => {
+      let resume: (() => void) | undefined
+      let chunkNumber = 0
+      const responseBody = 'abcd'
+
+      emulateFetchApi({body: responseBody.split('').map(Buffer.from)})
+      const observer: CommunicationObserver<Uint8Array> = {
+        next(_chunk: Uint8Array) {
+          return ++chunkNumber === 2 ? false : undefined // pause at 2nd chunk
+        },
+        error() {},
+        complete(): void {},
+        useResume(r) {
+          resume = r
+        },
+      }
+      const spy = sinon.spy(observer)
+
+      new FetchTransport({url: '/test'}).send(
+        '/test',
+        '',
+        {
+          method: 'GET',
+        },
+        spy
+      )
+      // wait for useResume being called
+      await waitForCondition(() => resume, 'resume callback is set')
+      expect(spy.next.callCount).equals(2)
+      expect(resume).is.not.null
+      if (resume) resume()
+      await waitForCondition(
+        () => spy.complete.callCount === 1,
+        'response is fully read'
+      )
+      expect(spy.next.callCount).equals(responseBody.length)
+      expect(
+        spy.next.args.reduce((acc, [body]) => acc + body.toString(), '')
+      ).equals(responseBody)
+    })
+  })
   describe('chunkCombiner', () => {
     const options = {url: 'http://test:8086'}
     const chunkCombiner = new FetchTransport(options).chunkCombiner
diff --git a/packages/core/test/unit/impl/browser/emulateBrowser.ts b/packages/core/test/unit/impl/browser/emulateBrowser.ts
index 490a2ae1..00b90e99 100644
--- a/packages/core/test/unit/impl/browser/emulateBrowser.ts
+++ b/packages/core/test/unit/impl/browser/emulateBrowser.ts
@@ -68,6 +68,9 @@ function createResponse({
               })
             }
           },
+          cancel(_msg = '') {
+            /* read cancelled with an optional message*/
+          },
         }
       },
     }
@@ -79,6 +82,23 @@ let beforeEmulation:
   | {fetch: any; abortController: any; textEncoder: any}
   | undefined
 
+export class AbortController {
+  private listeners: Array<() => void> = []
+  signal = {
+    aborted: false,
+    addEventListener: (type: string, listener: () => void) => {
+      this.listeners.push(listener)
+    },
+  }
+  constructor(aborted = false) {
+    this.signal.aborted = aborted
+  }
+  abort(): void {
+    this.signal.aborted = true
+    this.listeners.forEach((x) => x())
+  }
+}
+
 export function emulateFetchApi(
   spec: ResponseSpec,
   onRequest?: (options: any) => void
@@ -89,15 +109,6 @@ export function emulateFetchApi(
       ? Promise.reject(new Error(url))
       : Promise.resolve(createResponse(spec))
   }
-  class AbortController {
-    signal = {
-      aborted: false,
-    }
-    abort(): void {
-      this.signal.aborted = true
-    }
-  }
-
   class TextEncoder {
     encode(s: string): Uint8Array {
       return Buffer.from(s)
diff --git a/packages/core/test/unit/impl/node/NodeHttpTransport.test.ts b/packages/core/test/unit/impl/node/NodeHttpTransport.test.ts
index 9ab345b7..a34d5fba 100644
--- a/packages/core/test/unit/impl/node/NodeHttpTransport.test.ts
+++ b/packages/core/test/unit/impl/node/NodeHttpTransport.test.ts
@@ -3,7 +3,7 @@ import nock from 'nock' // WARN: nock must be imported before NodeHttpTransport,
 import NodeHttpTransport from '../../../../src/impl/node/NodeHttpTransport'
 import {ConnectionOptions} from '../../../../src/options'
 import {SendOptions} from '../../../../src/transport'
-import {Cancellable} from '../../../../src/results'
+import {Cancellable, CommunicationObserver} from '../../../../src/results'
 import * as http from 'http'
 import * as https from 'https'
 import sinon from 'sinon'
@@ -11,6 +11,8 @@ import {Readable} from 'stream'
 import zlib from 'zlib'
 import {CLIENT_LIB_VERSION} from '../../../../src/impl/version'
 import {CollectedLogs, collectLogging} from '../../../util'
+import {waitForCondition} from '../../util/waitForCondition'
+import {AddressInfo} from 'net'
 
 function sendTestData(
   connectionOptions: ConnectionOptions,
@@ -500,6 +502,156 @@ describe('NodeHttpTransport', () => {
       })
     })
   })
+  describe('send.backpressure', () => {
+    let server: http.Server
+    let url = ''
+    before(async () => {
+      await new Promise<void>((resolve) => {
+        server = http.createServer()
+        server.listen(() => {
+          const addr = server.address() as AddressInfo
+          url = `http://${addr.address}:${addr.port}`
+          resolve()
+        })
+      })
+    })
+    after(() => {
+      server.close()
+    })
+    afterEach(async () => {
+      server.removeAllListeners('request')
+    })
+
+    it(`it throws an error when paused and useResume is not set`, async () => {
+      server.on('request', async (_req, res) => {
+        res.setHeader('content-type', 'application/csv')
+        res.writeHead(200)
+        const writeUntilFull = () => {
+          while (res.write('.'));
+        }
+        writeUntilFull()
+        res.once('drain', () => {
+          res.write('.')
+          res.end()
+        })
+      })
+      const observer: CommunicationObserver<Uint8Array> = {
+        next(_chunk: Uint8Array) {
+          // do not receive more than 1 chunk, but still
+          // there is no useResume callback!
+          return false
+        },
+        error() {},
+        complete(): void {},
+      }
+      const spy = sinon.spy(observer)
+
+      new NodeHttpTransport({url, timeout: 10000}).send(
+        '/test',
+        '',
+        {
+          method: 'GET',
+        },
+        spy
+      )
+      // wait for error being called
+      await waitForCondition(() => spy.error.callCount === 1)
+      expect(spy.next.callCount).equals(1)
+      expect(spy.error.getCall(0).args[0]?.message).contains(
+        'useResume is not configured!'
+      )
+    })
+
+    it(`is paused after the first chunk, then cancelled`, async () => {
+      let cancellable: Cancellable | undefined
+      let resume: () => void | undefined
+
+      server.on('request', async (_req, res) => {
+        res.setHeader('content-type', 'application/csv')
+        res.writeHead(200)
+        const writeUntilFull = () => {
+          while (res.write('.'));
+        }
+        writeUntilFull()
+        res.once('drain', () => res.end())
+      })
+      const observer: CommunicationObserver<Uint8Array> = {
+        next(_chunk: Uint8Array) {
+          return false // do not receive more than 1 chunk
+        },
+        error() {},
+        complete(): void {},
+        useCancellable(c: Cancellable) {
+          cancellable = c
+        },
+        useResume(r) {
+          resume = r
+        },
+      }
+      const spy = sinon.spy(observer)
+
+      new NodeHttpTransport({url, timeout: 10000}).send(
+        '/test',
+        '',
+        {
+          method: 'GET',
+        },
+        spy
+      )
+      // wait for resume being called
+      await waitForCondition(() => cancellable && resume)
+      expect(spy.next.callCount).equals(1)
+      cancellable?.cancel()
+      await waitForCondition(() => spy.complete.callCount == 1)
+    })
+    it(`is paused after the second chunk and then read fully`, async () => {
+      let resume: (() => void) | undefined
+      let chunkNumber = 0
+
+      server.on('request', async (_req, res) => {
+        res.setHeader('content-type', 'application/csv')
+        res.writeHead(200)
+        const writeUntilFull = () => {
+          while (res.write('.'));
+        }
+        writeUntilFull()
+        res.once('drain', () => {
+          res.write('.')
+          res.end()
+        })
+      })
+      const observer: CommunicationObserver<Uint8Array> = {
+        next(_chunk: Uint8Array) {
+          return ++chunkNumber === 2 ? false : undefined // pause at 2nd chunk
+        },
+        error() {},
+        complete(): void {},
+        useResume(r) {
+          resume = r
+        },
+      }
+      const spy = sinon.spy(observer)
+
+      new NodeHttpTransport({url, timeout: 10000}).send(
+        '/test',
+        '',
+        {
+          method: 'GET',
+        },
+        spy
+      )
+      // wait for useResume being called
+      await waitForCondition(() => resume, 'resume callback is set')
+      expect(spy.next.callCount).equals(2)
+      expect(resume).is.not.null
+      if (resume) resume()
+      await waitForCondition(
+        () => spy.complete.callCount === 1,
+        'response is fully read'
+      )
+      expect(spy.next.callCount).is.greaterThan(2)
+    })
+  })
 
   describe('request', () => {
     beforeEach(() => {
diff --git a/packages/core/test/unit/results/chunksToLines.test.ts b/packages/core/test/unit/results/chunksToLines.test.ts
index e0e90bb2..c438b318 100644
--- a/packages/core/test/unit/results/chunksToLines.test.ts
+++ b/packages/core/test/unit/results/chunksToLines.test.ts
@@ -88,4 +88,82 @@ describe('chunksToLines', () => {
     expect((cancellable as any)?.isCancelled()).equals(true)
     expect(target.complete.callCount).equals(1)
   })
+  it('can pause and resume processing', () => {
+    let lines: string[] = []
+    let chunksResume = sinon.spy((): void => {})
+    const target = {
+      next(line: string): boolean {
+        lines.push(line)
+        const paused = line.includes('pause')
+        return !paused
+      },
+      error: sinon.fake(),
+      complete: sinon.fake(),
+      useResume: sinon.spy((_x: () => void): void => {}),
+    }
+    const subject = chunksToLines(target, nodeChunkCombiner)
+    let nextVal = subject.next(Buffer.from('a\npause\nb\nd', 'utf8'))
+    subject?.useResume?.((chunksResume = sinon.spy((): void => {})))
+    expect(nextVal).equals(false)
+    expect(lines).deep.equals(['a', 'pause'])
+    expect(target.error.callCount).equals(0)
+    expect(target.complete.callCount).equals(0)
+    expect(target.useResume.callCount).equals(1)
+    expect(chunksResume.callCount).equals(0)
+    target.useResume.args[0][0]() // resume
+    expect(lines).deep.equals(['a', 'pause', 'b'])
+    expect(target.error.callCount).equals(0)
+    expect(target.complete.callCount).equals(0)
+    expect(target.useResume.callCount).equals(1)
+    expect(chunksResume.callCount).equals(1)
+    lines = []
+    nextVal = subject.next(Buffer.from('pause\npause\nf', 'utf8'))
+    expect(nextVal).equals(false)
+    subject?.useResume?.((chunksResume = sinon.spy((): void => {})))
+    expect(lines).deep.equals(['dpause'])
+    expect(target.error.callCount).equals(0)
+    expect(target.complete.callCount).equals(0)
+    expect(target.useResume.callCount).equals(2)
+    expect(chunksResume.callCount).equals(0)
+    target.useResume.args[1][0]() // resume
+    expect(lines).deep.equals(['dpause', 'pause'])
+    expect(target.error.callCount).equals(0)
+    expect(target.complete.callCount).equals(0)
+    expect(target.useResume.callCount).equals(3)
+    expect(chunksResume.callCount).equals(0)
+    target.useResume.args[2][0]() // resume
+    expect(lines).deep.equals(['dpause', 'pause'])
+    expect(target.error.callCount).equals(0)
+    expect(target.complete.callCount).equals(0)
+    expect(target.useResume.callCount).equals(3)
+    expect(chunksResume.callCount).equals(1)
+    subject.complete()
+    expect(lines).deep.equals(['dpause', 'pause', 'f'])
+    expect(target.error.callCount).equals(0)
+    expect(target.complete.callCount).equals(1)
+    expect(target.useResume.callCount).equals(3)
+  })
+  it('requires useResume', () => {
+    const lines: string[] = []
+    const target = {
+      next(line: string): boolean {
+        lines.push(line)
+        const paused = line.includes('pause')
+        return !paused
+      },
+      error: sinon.fake(),
+      complete: sinon.fake(),
+    }
+    const subject = chunksToLines(target, nodeChunkCombiner)
+    const nextVal = subject.next(Buffer.from('a\npause\nb\nd', 'utf8'))
+    expect(nextVal).equals(true)
+    expect(lines).deep.equals(['a', 'pause'])
+    expect(target.error.callCount).equals(1)
+    expect(target.complete.callCount).equals(0)
+    subject.next(Buffer.from('whatever'))
+    subject.complete()
+    expect(lines).deep.equals(['a', 'pause'])
+    expect(target.error.callCount).equals(1)
+    expect(target.complete.callCount).equals(0)
+  })
 })
diff --git a/packages/core/test/unit/results/linesToTables.test.ts b/packages/core/test/unit/results/linesToTables.test.ts
index f150b501..c1a65c27 100644
--- a/packages/core/test/unit/results/linesToTables.test.ts
+++ b/packages/core/test/unit/results/linesToTables.test.ts
@@ -6,6 +6,7 @@ import {
   createTextDecoderCombiner,
   linesToTables,
 } from '../../../src/results'
+import sinon from 'sinon'
 
 describe('linesToTables', () => {
   const chunkCombiner = createTextDecoderCombiner()
@@ -39,4 +40,40 @@ describe('linesToTables', () => {
     input.error(new Error())
     expect(target.failed).equals(1)
   })
+  it('can pause and resume processing', () => {
+    const rows: string[][] = []
+    let chunksResume = sinon.spy((): void => {})
+    const target = {
+      next(row: string[]): boolean {
+        rows.push(row)
+        const paused = row[0].includes('pause')
+        return !paused
+      },
+      error: sinon.fake(),
+      complete: sinon.fake(),
+      useResume: sinon.spy((_x: () => void): void => {}),
+    }
+    const subject = linesToTables(target)
+    let nextVal = subject.next('a,b\n')
+    expect(nextVal).equals(true)
+    nextVal = subject.next('pause,1')
+    expect(nextVal).equals(false)
+    subject?.useResume?.((chunksResume = sinon.spy((): void => {})))
+    expect(nextVal).equals(false)
+    expect(rows).deep.equals([['pause', '1']])
+    expect(target.error.callCount).equals(0)
+    expect(target.complete.callCount).equals(0)
+    expect(target.useResume.callCount).equals(1)
+    expect(chunksResume.callCount).equals(0)
+    target.useResume.args[0][0]() // resume
+    expect(chunksResume.callCount).equals(1)
+    subject.next('ok,2')
+    subject.complete()
+    expect(rows).deep.equals([
+      ['pause', '1'],
+      ['ok', '2'],
+    ])
+    expect(target.error.callCount).equals(0)
+    expect(target.complete.callCount).equals(1)
+  })
 })
diff --git a/packages/core/test/unit/util/waitForCondition.ts b/packages/core/test/unit/util/waitForCondition.ts
index 6270d9b9..fa6697da 100644
--- a/packages/core/test/unit/util/waitForCondition.ts
+++ b/packages/core/test/unit/util/waitForCondition.ts
@@ -20,6 +20,5 @@ export async function waitForCondition(
     }
     if (condition()) return
   }
-  // eslint-disable-next-line no-console
-  console.error(`WARN:waitForCondition: ${message}`)
+  return Promise.reject(`WARN:waitForCondition: ${message}`)
 }