You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
63 lines
1.7 KiB
63 lines
1.7 KiB
3 years ago
|
import type {
|
||
|
AnyBulkWriteOperation,
|
||
|
BulkOperationBase,
|
||
|
BulkWriteOptions,
|
||
|
BulkWriteResult
|
||
|
} from '../bulk/common';
|
||
|
import type { Collection } from '../collection';
|
||
|
import type { Server } from '../sdam/server';
|
||
|
import type { ClientSession } from '../sessions';
|
||
|
import type { Callback } from '../utils';
|
||
|
import { AbstractOperation, Aspect, defineAspects } from './operation';
|
||
|
|
||
|
/** @internal */
|
||
|
export class BulkWriteOperation extends AbstractOperation<BulkWriteResult> {
|
||
|
override options: BulkWriteOptions;
|
||
|
collection: Collection;
|
||
|
operations: AnyBulkWriteOperation[];
|
||
|
|
||
|
constructor(
|
||
|
collection: Collection,
|
||
|
operations: AnyBulkWriteOperation[],
|
||
|
options: BulkWriteOptions
|
||
|
) {
|
||
|
super(options);
|
||
|
this.options = options;
|
||
|
this.collection = collection;
|
||
|
this.operations = operations;
|
||
|
}
|
||
|
|
||
|
override execute(
|
||
|
server: Server,
|
||
|
session: ClientSession | undefined,
|
||
|
callback: Callback<BulkWriteResult>
|
||
|
): void {
|
||
|
const coll = this.collection;
|
||
|
const operations = this.operations;
|
||
|
const options = { ...this.options, ...this.bsonOptions, readPreference: this.readPreference };
|
||
|
|
||
|
// Create the bulk operation
|
||
|
const bulk: BulkOperationBase =
|
||
|
options.ordered === false
|
||
|
? coll.initializeUnorderedBulkOp(options)
|
||
|
: coll.initializeOrderedBulkOp(options);
|
||
|
|
||
|
// for each op go through and add to the bulk
|
||
|
try {
|
||
|
for (let i = 0; i < operations.length; i++) {
|
||
|
bulk.raw(operations[i]);
|
||
|
}
|
||
|
} catch (err) {
|
||
|
return callback(err);
|
||
|
}
|
||
|
|
||
|
// Execute the bulk
|
||
|
bulk.execute({ ...options, session }).then(
|
||
|
result => callback(undefined, result),
|
||
|
error => callback(error)
|
||
|
);
|
||
|
}
|
||
|
}
|
||
|
|
||
|
defineAspects(BulkWriteOperation, [Aspect.WRITE_OPERATION]);
|