onSuccess function for bulk helper (#2199)

* Bulk helper onSuccess callback

For https://github.com/elastic/elasticsearch-js/issues/2090

Includes refactor of the tryBulk result processing code, to make
iterating over bulk response data easier to understand.

* Add onSuccess tests for each datasource type

* Cleanup, additional comments

* Add documentation for onSuccess callback

* Update changelog

* Drop link to 8.14 release notes.

Page not yet published, breaking docs build.

test/unit/helpers/bulk.test.ts

@@ -514,7 +514,7 @@ test('bulk index', t => {
 
     t.test('Server error', async t => {
       const MockConnection = connection.buildMockConnection({
-        onRequest (params) {
+        onRequest (_params) {
           return {
             statusCode: 500,
             body: { somothing: 'went wrong' }
@@ -530,12 +530,12 @@ test('bulk index', t => {
         datasource: dataset.slice(),
         flushBytes: 1,
         concurrency: 1,
-        onDocument (doc) {
+        onDocument (_doc) {
           return {
             index: { _index: 'test' }
           }
         },
-        onDrop (doc) {
+        onDrop (_doc) {
           t.fail('This should never be called')
         }
       })
@@ -550,7 +550,7 @@ test('bulk index', t => {
 
     t.test('Server error (high flush size, to trigger the finish error)', async t => {
       const MockConnection = connection.buildMockConnection({
-        onRequest (params) {
+        onRequest (_params) {
           return {
             statusCode: 500,
             body: { somothing: 'went wrong' }
@@ -566,12 +566,12 @@ test('bulk index', t => {
         datasource: dataset.slice(),
         flushBytes: 5000000,
         concurrency: 1,
-        onDocument (doc) {
+        onDocument (_doc) {
           return {
             index: { _index: 'test' }
           }
         },
-        onDrop (doc) {
+        onDrop (_doc) {
           t.fail('This should never be called')
         }
       })
@@ -625,12 +625,12 @@ test('bulk index', t => {
         flushBytes: 1,
         concurrency: 1,
         wait: 10,
-        onDocument (doc) {
+        onDocument (_doc) {
           return {
             index: { _index: 'test' }
           }
         },
-        onDrop (doc) {
+        onDrop (_doc) {
           b.abort()
         }
       })
@@ -651,7 +651,7 @@ test('bulk index', t => {
     t.test('Invalid operation', t => {
       t.plan(2)
       const MockConnection = connection.buildMockConnection({
-        onRequest (params) {
+        onRequest (_params) {
           return { body: { errors: false, items: [{}] } }
         }
       })
@@ -666,7 +666,7 @@ test('bulk index', t => {
           flushBytes: 1,
           concurrency: 1,
           // @ts-expect-error
-          onDocument (doc) {
+          onDocument (_doc) {
             return {
               foo: { _index: 'test' }
             }
@@ -678,6 +678,43 @@ test('bulk index', t => {
         })
     })
 
+    t.test('should call onSuccess callback for each indexed document', async t => {
+      const MockConnection = connection.buildMockConnection({
+        onRequest (params) {
+          // @ts-expect-error
+          let [action] = params.body.split('\n')
+          action = JSON.parse(action)
+          return { body: { errors: false, items: [action] } }
+        }
+      })
+
+      const client = new Client({
+        node: 'http://localhost:9200',
+        Connection: MockConnection
+      })
+
+      let count = 0
+      await client.helpers.bulk<Document>({
+        datasource: dataset.slice(),
+        flushBytes: 1,
+        concurrency: 1,
+        onDocument (_doc) {
+          return {
+            index: { _index: 'test' }
+          }
+        },
+        onSuccess ({ result, document }) {
+          t.same(result, { index: { _index: 'test' }})
+          t.same(document, dataset[count++])
+        },
+        onDrop (_doc) {
+          t.fail('This should never be called')
+        }
+      })
+      t.equal(count, 3)
+      t.end()
+    })
+
     t.end()
   })
 
@@ -731,6 +768,44 @@ test('bulk index', t => {
       })
     })
 
+    t.test('onSuccess is called for each indexed document', async t => {
+      const MockConnection = connection.buildMockConnection({
+        onRequest (params) {
+          // @ts-expect-error
+          let [action] = params.body.split('\n')
+          action = JSON.parse(action)
+          return { body: { errors: false, items: [action] } }
+        }
+      })
+
+      const client = new Client({
+        node: 'http://localhost:9200',
+        Connection: MockConnection
+      })
+      const stream = createReadStream(join(__dirname, '..', '..', 'fixtures', 'small-dataset.ndjson'), 'utf8')
+
+      let count = 0
+      await client.helpers.bulk<Document>({
+        datasource: stream.pipe(split()),
+        flushBytes: 1,
+        concurrency: 1,
+        onDocument (_doc) {
+          return {
+            index: { _index: 'test' }
+          }
+        },
+        onSuccess ({ result, document }) {
+          t.same(result, { index: { _index: 'test' }})
+          t.same(document, dataset[count++])
+        },
+        onDrop (_doc) {
+          t.fail('This should never be called')
+        }
+      })
+      t.equal(count, 3)
+      t.end()
+    })
+
     t.end()
   })
 
@@ -785,6 +860,50 @@ test('bulk index', t => {
         aborted: false
       })
     })
+
+    t.test('onSuccess is called for each indexed document', async t => {
+      const MockConnection = connection.buildMockConnection({
+        onRequest (params) {
+          // @ts-expect-error
+          let [action] = params.body.split('\n')
+          action = JSON.parse(action)
+          return { body: { errors: false, items: [action] } }
+        }
+      })
+
+      const client = new Client({
+        node: 'http://localhost:9200',
+        Connection: MockConnection
+      })
+
+      async function * generator () {
+        const data = dataset.slice()
+        for (const doc of data) {
+          yield doc
+        }
+      }
+
+      let count = 0
+      await client.helpers.bulk<Document>({
+        datasource: generator(),
+        flushBytes: 1,
+        concurrency: 1,
+        onDocument (_doc) {
+          return {
+            index: { _index: 'test' }
+          }
+        },
+        onSuccess ({ result, document }) {
+          t.same(result, { index: { _index: 'test' }})
+          t.same(document, dataset[count++])
+        },
+        onDrop (_doc) {
+          t.fail('This should never be called')
+        }
+      })
+      t.equal(count, 3)
+      t.end()
+    })
     t.end()
   })
 
@@ -943,6 +1062,8 @@ test('bulk create', t => {
     })
   })
 
+
+
   t.end()
 })
 
@@ -1279,6 +1400,63 @@ test('bulk delete', t => {
     server.stop()
   })
 
+  t.test('should call onSuccess callback with delete action object', async t => {
+    const MockConnection = connection.buildMockConnection({
+      onRequest (params) {
+        // @ts-expect-error
+        let [action, payload] = params.body.split('\n')
+        action = JSON.parse(action)
+        return { body: { errors: false, items: [action] } }
+      }
+    })
+
+    const client = new Client({
+      node: 'http://localhost:9200',
+      Connection: MockConnection
+    })
+
+    let docCount = 0
+    let successCount = 0
+    await client.helpers.bulk<Document>({
+      datasource: dataset.slice(),
+      flushBytes: 1,
+      concurrency: 1,
+      onDocument (_doc) {
+        if (docCount++ === 1) {
+          return {
+            delete: {
+              _index: 'test',
+              _id: String(docCount)
+            }
+          }
+        } else {
+          return {
+            index: { _index: 'test' }
+          }
+        }
+      },
+      onSuccess ({ result, document }) {
+        const item = dataset[successCount]
+        if (successCount++ === 1) {
+          t.same(result, {
+            delete: {
+              _index: 'test',
+              _id: String(successCount)
+            }
+          })
+        } else {
+          t.same(result, { index: { _index: 'test' }})
+          t.same(document, item)
+        }
+      },
+      onDrop (_doc) {
+        t.fail('This should never be called')
+      }
+    })
+
+    t.end()
+  })
+
   t.end()
 })
 
@@ -1594,152 +1772,153 @@ test('Flush interval', t => {
     })
   })
 
+  test(`flush timeout does not lock process when flushInterval is less than server timeout`, async t => {
+    const flushInterval = 500
+
+    async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+      setTimeout(() => {
+        res.writeHead(200, { 'content-type': 'application/json' })
+        res.end(JSON.stringify({ errors: false, items: [{}] }))
+      }, 1000)
+    }
+
+    const [{ port }, server] = await buildServer(handler)
+    const client = new Client({ node: `http://localhost:${port}` })
+
+    async function * generator () {
+      const data = dataset.slice()
+      for (const doc of data) {
+        await sleep(flushInterval)
+        yield doc
+      }
+    }
+
+    const result = await client.helpers.bulk({
+      datasource: Readable.from(generator()),
+      flushBytes: 1,
+      flushInterval: flushInterval,
+      concurrency: 1,
+      onDocument (_) {
+        return {
+          index: { _index: 'test' }
+        }
+      },
+      onDrop (_) {
+        t.fail('This should never be called')
+      }
+    })
+
+    t.type(result.time, 'number')
+    t.type(result.bytes, 'number')
+    t.match(result, {
+      total: 3,
+      successful: 3,
+      retry: 0,
+      failed: 0,
+      aborted: false
+    })
+
+    server.stop()
+  })
+
+  test(`flush timeout does not lock process when flushInterval is greater than server timeout`, async t => {
+    const flushInterval = 500
+
+    async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+      setTimeout(() => {
+        res.writeHead(200, { 'content-type': 'application/json' })
+        res.end(JSON.stringify({ errors: false, items: [{}] }))
+      }, 250)
+    }
+
+    const [{ port }, server] = await buildServer(handler)
+    const client = new Client({ node: `http://localhost:${port}` })
+
+    async function * generator () {
+      const data = dataset.slice()
+      for (const doc of data) {
+        await sleep(flushInterval)
+        yield doc
+      }
+    }
+
+    const result = await client.helpers.bulk({
+      datasource: Readable.from(generator()),
+      flushBytes: 1,
+      flushInterval: flushInterval,
+      concurrency: 1,
+      onDocument (_) {
+        return {
+          index: { _index: 'test' }
+        }
+      },
+      onDrop (_) {
+        t.fail('This should never be called')
+      }
+    })
+
+    t.type(result.time, 'number')
+    t.type(result.bytes, 'number')
+    t.match(result, {
+      total: 3,
+      successful: 3,
+      retry: 0,
+      failed: 0,
+      aborted: false
+    })
+
+    server.stop()
+  })
+
+  test(`flush timeout does not lock process when flushInterval is equal to server timeout`, async t => {
+    const flushInterval = 500
+
+    async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+      setTimeout(() => {
+        res.writeHead(200, { 'content-type': 'application/json' })
+        res.end(JSON.stringify({ errors: false, items: [{}] }))
+      }, flushInterval)
+    }
+
+    const [{ port }, server] = await buildServer(handler)
+    const client = new Client({ node: `http://localhost:${port}` })
+
+    async function * generator () {
+      const data = dataset.slice()
+      for (const doc of data) {
+        await sleep(flushInterval)
+        yield doc
+      }
+    }
+
+    const result = await client.helpers.bulk({
+      datasource: Readable.from(generator()),
+      flushBytes: 1,
+      flushInterval: flushInterval,
+      concurrency: 1,
+      onDocument (_) {
+        return {
+          index: { _index: 'test' }
+        }
+      },
+      onDrop (_) {
+        t.fail('This should never be called')
+      }
+    })
+
+    t.type(result.time, 'number')
+    t.type(result.bytes, 'number')
+    t.match(result, {
+      total: 3,
+      successful: 3,
+      retry: 0,
+      failed: 0,
+      aborted: false
+    })
+
+    server.stop()
+  })
+
   t.end()
 })
 
-test(`flush timeout does not lock process when flushInterval is less than server timeout`, async t => {
-  const flushInterval = 500
-
-  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
-    setTimeout(() => {
-      res.writeHead(200, { 'content-type': 'application/json' })
-      res.end(JSON.stringify({ errors: false, items: [{}] }))
-    }, 1000)
-  }
-
-  const [{ port }, server] = await buildServer(handler)
-  const client = new Client({ node: `http://localhost:${port}` })
-
-  async function * generator () {
-    const data = dataset.slice()
-    for (const doc of data) {
-      await sleep(flushInterval)
-      yield doc
-    }
-  }
-
-  const result = await client.helpers.bulk({
-    datasource: Readable.from(generator()),
-    flushBytes: 1,
-    flushInterval: flushInterval,
-    concurrency: 1,
-    onDocument (_) {
-      return {
-        index: { _index: 'test' }
-      }
-    },
-    onDrop (_) {
-      t.fail('This should never be called')
-    }
-  })
-
-  t.type(result.time, 'number')
-  t.type(result.bytes, 'number')
-  t.match(result, {
-    total: 3,
-    successful: 3,
-    retry: 0,
-    failed: 0,
-    aborted: false
-  })
-
-  server.stop()
-})
-
-test(`flush timeout does not lock process when flushInterval is greater than server timeout`, async t => {
-  const flushInterval = 500
-
-  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
-    setTimeout(() => {
-      res.writeHead(200, { 'content-type': 'application/json' })
-      res.end(JSON.stringify({ errors: false, items: [{}] }))
-    }, 250)
-  }
-
-  const [{ port }, server] = await buildServer(handler)
-  const client = new Client({ node: `http://localhost:${port}` })
-
-  async function * generator () {
-    const data = dataset.slice()
-    for (const doc of data) {
-      await sleep(flushInterval)
-      yield doc
-    }
-  }
-
-  const result = await client.helpers.bulk({
-    datasource: Readable.from(generator()),
-    flushBytes: 1,
-    flushInterval: flushInterval,
-    concurrency: 1,
-    onDocument (_) {
-      return {
-        index: { _index: 'test' }
-      }
-    },
-    onDrop (_) {
-      t.fail('This should never be called')
-    }
-  })
-
-  t.type(result.time, 'number')
-  t.type(result.bytes, 'number')
-  t.match(result, {
-    total: 3,
-    successful: 3,
-    retry: 0,
-    failed: 0,
-    aborted: false
-  })
-
-  server.stop()
-})
-
-test(`flush timeout does not lock process when flushInterval is equal to server timeout`, async t => {
-  const flushInterval = 500
-
-  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
-    setTimeout(() => {
-      res.writeHead(200, { 'content-type': 'application/json' })
-      res.end(JSON.stringify({ errors: false, items: [{}] }))
-    }, flushInterval)
-  }
-
-  const [{ port }, server] = await buildServer(handler)
-  const client = new Client({ node: `http://localhost:${port}` })
-
-  async function * generator () {
-    const data = dataset.slice()
-    for (const doc of data) {
-      await sleep(flushInterval)
-      yield doc
-    }
-  }
-
-  const result = await client.helpers.bulk({
-    datasource: Readable.from(generator()),
-    flushBytes: 1,
-    flushInterval: flushInterval,
-    concurrency: 1,
-    onDocument (_) {
-      return {
-        index: { _index: 'test' }
-      }
-    },
-    onDrop (_) {
-      t.fail('This should never be called')
-    }
-  })
-
-  t.type(result.time, 'number')
-  t.type(result.bytes, 'number')
-  t.match(result, {
-    total: 3,
-    successful: 3,
-    retry: 0,
-    failed: 0,
-    aborted: false
-  })
-
-  server.stop()
-})