Skip to content

Commit 6fca2d3

Browse files
committed
Backport missing patch from elastic/elasticsearch-js#2027
1 parent 1caa7f4 commit 6fca2d3

File tree

2 files changed

+373
-14
lines changed

2 files changed

+373
-14
lines changed

src/helpers.ts

Lines changed: 16 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -74,11 +74,11 @@ export interface BulkStats {
7474
aborted: boolean
7575
}
7676

77-
interface IndexAction {
77+
interface IndexActionOperation {
7878
index: T.BulkIndexOperation
7979
}
8080

81-
interface CreateAction {
81+
interface CreateActionOperation {
8282
create: T.BulkCreateOperation
8383
}
8484

@@ -90,7 +90,9 @@ interface DeleteAction {
9090
delete: T.BulkDeleteOperation
9191
}
9292

93-
type UpdateAction = [UpdateActionOperation, Record<string, any>]
93+
type CreateAction = CreateActionOperation | [CreateActionOperation, unknown]
94+
type IndexAction = IndexActionOperation | [IndexActionOperation, unknown]
95+
type UpdateAction = [UpdateActionOperation, T.BulkUpdateAction]
9496
type Action = IndexAction | CreateAction | UpdateAction | DeleteAction
9597

9698
export interface OnDropDocument<TDocument = unknown> {
@@ -646,22 +648,21 @@ export default class Helpers {
646648
for await (const chunk of datasource) {
647649
if (shouldAbort) break
648650
timeoutRef.refresh()
649-
const action = onDocument(chunk)
650-
const operation = Array.isArray(action)
651-
? Object.keys(action[0])[0]
652-
: Object.keys(action)[0]
651+
const result = onDocument(chunk)
652+
const [action, payload] = Array.isArray(result) ? result : [result, chunk]
653+
const operation = Object.keys(action)[0]
653654
if (operation === 'index' || operation === 'create') {
654655
actionBody = serializer.serialize(action)
655-
payloadBody = typeof chunk === 'string' ? chunk : serializer.serialize(chunk)
656+
payloadBody = typeof payload === 'string'
657+
? payload
658+
: serializer.serialize(payload)
656659
chunkBytes += Buffer.byteLength(actionBody) + Buffer.byteLength(payloadBody)
657660
bulkBody.push(actionBody, payloadBody)
658661
} else if (operation === 'update') {
659-
// @ts-expect-error in case of update action is an array
660-
actionBody = serializer.serialize(action[0])
662+
actionBody = serializer.serialize(action)
661663
payloadBody = typeof chunk === 'string'
662664
? `{"doc":${chunk}}`
663-
// @ts-expect-error in case of update action is an array
664-
: serializer.serialize({ doc: chunk, ...action[1] })
665+
: serializer.serialize({ doc: chunk, ...payload })
665666
chunkBytes += Buffer.byteLength(actionBody) + Buffer.byteLength(payloadBody)
666667
bulkBody.push(actionBody, payloadBody)
667668
} else if (operation === 'delete') {
@@ -675,10 +676,11 @@ export default class Helpers {
675676

676677
if (chunkBytes >= flushBytes) {
677678
stats.bytes += chunkBytes
678-
const send = await semaphore()
679-
send(bulkBody.slice())
679+
const bulkBodyCopy = bulkBody.slice()
680680
bulkBody.length = 0
681681
chunkBytes = 0
682+
const send = await semaphore()
683+
send(bulkBodyCopy)
682684
}
683685
}
684686

0 commit comments

Comments
 (0)