Added async generator support in bulk helper (#1138)

* Added async generator support in bulk helper

* Updated test

* Updated docs

* Improved type definitions

* Updated onDrop callback type definition

test/types/helpers.test-d.ts

@@ -9,7 +9,8 @@ import {
   BulkHelper,
   BulkStats,
   BulkHelperOptions,
-  ScrollSearchResponse
+  ScrollSearchResponse,
+  OnDropDocument
 } from '../../lib/Helpers'
 
 const client = new Client({
@@ -18,7 +19,7 @@ const client = new Client({
 
 /// .helpers.bulk
 
-const b = client.helpers.bulk({
+const b = client.helpers.bulk<Record<string, any>>({
   datasource: [],
   onDocument (doc) {
     expectType<Record<string, any>>(doc)
@@ -29,7 +30,7 @@ const b = client.helpers.bulk({
   retries: 3,
   wait: 5000,
   onDrop (doc) {
-    expectType<Record<string, any>>(doc)
+    expectType<OnDropDocument<Record<string, any>>>(doc)
   },
   refreshOnCompletion: true,
   pipeline: 'my-pipeline'
@@ -59,7 +60,7 @@ expectError(
       return { index: { _index: 'test' } } 
     }
   }
-  expectAssignable<BulkHelperOptions>(options)
+  expectAssignable<BulkHelperOptions<Record<string, any>>>(options)
 }
 // create
 {
@@ -69,20 +70,20 @@ expectError(
       return { create: { _index: 'test' } }
     }
   }
-  expectAssignable<BulkHelperOptions>(options)
+  expectAssignable<BulkHelperOptions<Record<string, any>>>(options)
 }
 // update
 {
   // without `:BulkHelperOptions` this test cannot pass
   // but if we write these options inline inside
   // a `.helper.bulk`, it works as expected
-  const options: BulkHelperOptions = {
+  const options: BulkHelperOptions<Record<string, any>> = {
     datasource: [],
     onDocument (doc: Record<string, any>) {
       return [{ update: { _index: 'test' } }, doc]
     }
   }
-  expectAssignable<BulkHelperOptions>(options)
+  expectAssignable<BulkHelperOptions<Record<string, any>>>(options)
 }
 // delete
 {
@@ -92,7 +93,7 @@ expectError(
       return { delete: { _index: 'test' } }
     }
   }
-  expectAssignable<BulkHelperOptions>(options)
+  expectAssignable<BulkHelperOptions<Record<string, any>>>(options)
 }
 
 /// .helpers.scrollSearch

test/unit/helpers/bulk.test.js

@@ -653,6 +653,59 @@ test('bulk index', t => {
     t.end()
   })
 
+  t.test('datasource as async generator', t => {
+    t.test('Should perform a bulk request', async t => {
+      let count = 0
+      const MockConnection = connection.buildMockConnection({
+        onRequest (params) {
+          t.strictEqual(params.path, '/_bulk')
+          t.match(params.headers, { 'Content-Type': 'application/x-ndjson' })
+          const [action, payload] = params.body.split('\n')
+          t.deepEqual(JSON.parse(action), { index: { _index: 'test' } })
+          t.deepEqual(JSON.parse(payload), dataset[count++])
+          return { body: { errors: false, items: [{}] } }
+        }
+      })
+
+      const client = new Client({
+        node: 'http://localhost:9200',
+        Connection: MockConnection
+      })
+
+      async function * generator () {
+        const data = dataset.slice()
+        for (const doc of data) {
+          yield doc
+        }
+      }
+
+      const result = await client.helpers.bulk({
+        datasource: generator(),
+        flushBytes: 1,
+        concurrency: 1,
+        onDocument (doc) {
+          return {
+            index: { _index: 'test' }
+          }
+        },
+        onDrop (doc) {
+          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
+      })
+    })
+    t.end()
+  })
+
   t.end()
 })
 
@@ -896,7 +949,7 @@ test('errors', t => {
       })
     } catch (err) {
       t.true(err instanceof errors.ConfigurationError)
-      t.is(err.message, 'bulk helper: the datasource must be an array or a buffer or a readable stream')
+      t.is(err.message, 'bulk helper: the datasource must be an array or a buffer or a readable stream or an async generator')
     }
   })