something is working
This commit is contained in:
quoije
+941
@@ -0,0 +1,941 @@
|
||||
import type { Readable } from 'stream';
|
||||
import { promisify } from 'util';
|
||||
|
||||
import type { Binary, Document, Timestamp } from './bson';
|
||||
import { Collection } from './collection';
|
||||
import { CHANGE, CLOSE, END, ERROR, INIT, MORE, RESPONSE, RESUME_TOKEN_CHANGED } from './constants';
|
||||
import type { AbstractCursorEvents, CursorStreamOptions } from './cursor/abstract_cursor';
|
||||
import { ChangeStreamCursor, ChangeStreamCursorOptions } from './cursor/change_stream_cursor';
|
||||
import { Db } from './db';
|
||||
import {
|
||||
AnyError,
|
||||
isResumableError,
|
||||
MongoAPIError,
|
||||
MongoChangeStreamError,
|
||||
MongoRuntimeError
|
||||
} from './error';
|
||||
import { MongoClient } from './mongo_client';
|
||||
import { InferIdType, TypedEventEmitter } from './mongo_types';
|
||||
import type { AggregateOptions } from './operations/aggregate';
|
||||
import type { CollationOptions, OperationParent } from './operations/command';
|
||||
import type { ReadPreference } from './read_preference';
|
||||
import type { ServerSessionId } from './sessions';
|
||||
import { Callback, filterOptions, getTopology, maybePromise, MongoDBNamespace } from './utils';
|
||||
|
||||
/** @internal */
|
||||
const kCursorStream = Symbol('cursorStream');
|
||||
/** @internal */
|
||||
const kClosed = Symbol('closed');
|
||||
/** @internal */
|
||||
const kMode = Symbol('mode');
|
||||
|
||||
const CHANGE_STREAM_OPTIONS = [
|
||||
'resumeAfter',
|
||||
'startAfter',
|
||||
'startAtOperationTime',
|
||||
'fullDocument',
|
||||
'fullDocumentBeforeChange',
|
||||
'showExpandedEvents'
|
||||
] as const;
|
||||
|
||||
const CHANGE_DOMAIN_TYPES = {
|
||||
COLLECTION: Symbol('Collection'),
|
||||
DATABASE: Symbol('Database'),
|
||||
CLUSTER: Symbol('Cluster')
|
||||
};
|
||||
|
||||
const CHANGE_STREAM_EVENTS = [RESUME_TOKEN_CHANGED, END, CLOSE];
|
||||
|
||||
const NO_RESUME_TOKEN_ERROR =
|
||||
'A change stream document has been received that lacks a resume token (_id).';
|
||||
const CHANGESTREAM_CLOSED_ERROR = 'ChangeStream is closed';
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @deprecated Please use the ChangeStreamCursorOptions type instead.
|
||||
*/
|
||||
export interface ResumeOptions {
|
||||
startAtOperationTime?: Timestamp;
|
||||
batchSize?: number;
|
||||
maxAwaitTimeMS?: number;
|
||||
collation?: CollationOptions;
|
||||
readPreference?: ReadPreference;
|
||||
resumeAfter?: ResumeToken;
|
||||
startAfter?: ResumeToken;
|
||||
fullDocument?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Represents the logical starting point for a new ChangeStream or resuming a ChangeStream on the server.
|
||||
* @see https://www.mongodb.com/docs/manual/changeStreams/#std-label-change-stream-resume
|
||||
* @public
|
||||
*/
|
||||
export type ResumeToken = unknown;
|
||||
|
||||
/**
|
||||
* Represents a specific point in time on a server. Can be retrieved by using `db.command()`
|
||||
* @public
|
||||
* @see https://docs.mongodb.com/manual/reference/method/db.runCommand/#response
|
||||
*/
|
||||
export type OperationTime = Timestamp;
|
||||
|
||||
/** @public */
|
||||
export interface PipeOptions {
|
||||
end?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Options that can be passed to a ChangeStream. Note that startAfter, resumeAfter, and startAtOperationTime are all mutually exclusive, and the server will error if more than one is specified.
|
||||
* @public
|
||||
*/
|
||||
export interface ChangeStreamOptions extends AggregateOptions {
|
||||
/**
|
||||
* Allowed values: 'updateLookup', 'whenAvailable', 'required'.
|
||||
*
|
||||
* When set to 'updateLookup', the change notification for partial updates
|
||||
* will include both a delta describing the changes to the document as well
|
||||
* as a copy of the entire document that was changed from some time after
|
||||
* the change occurred.
|
||||
*
|
||||
* When set to 'whenAvailable', configures the change stream to return the
|
||||
* post-image of the modified document for replace and update change events
|
||||
* if the post-image for this event is available.
|
||||
*
|
||||
* When set to 'required', the same behavior as 'whenAvailable' except that
|
||||
* an error is raised if the post-image is not available.
|
||||
*/
|
||||
fullDocument?: string;
|
||||
|
||||
/**
|
||||
* Allowed values: 'whenAvailable', 'required', 'off'.
|
||||
*
|
||||
* The default is to not send a value, which is equivalent to 'off'.
|
||||
*
|
||||
* When set to 'whenAvailable', configures the change stream to return the
|
||||
* pre-image of the modified document for replace, update, and delete change
|
||||
* events if it is available.
|
||||
*
|
||||
* When set to 'required', the same behavior as 'whenAvailable' except that
|
||||
* an error is raised if the pre-image is not available.
|
||||
*/
|
||||
fullDocumentBeforeChange?: string;
|
||||
/** The maximum amount of time for the server to wait on new documents to satisfy a change stream query. */
|
||||
maxAwaitTimeMS?: number;
|
||||
/**
|
||||
* Allows you to start a changeStream after a specified event.
|
||||
* @see https://docs.mongodb.com/manual/changeStreams/#resumeafter-for-change-streams
|
||||
*/
|
||||
resumeAfter?: ResumeToken;
|
||||
/**
|
||||
* Similar to resumeAfter, but will allow you to start after an invalidated event.
|
||||
* @see https://docs.mongodb.com/manual/changeStreams/#startafter-for-change-streams
|
||||
*/
|
||||
startAfter?: ResumeToken;
|
||||
/** Will start the changeStream after the specified operationTime. */
|
||||
startAtOperationTime?: OperationTime;
|
||||
/**
|
||||
* The number of documents to return per batch.
|
||||
* @see https://docs.mongodb.com/manual/reference/command/aggregate
|
||||
*/
|
||||
batchSize?: number;
|
||||
|
||||
/**
|
||||
* When enabled, configures the change stream to include extra change events.
|
||||
*
|
||||
* - createIndexes
|
||||
* - dropIndexes
|
||||
* - modify
|
||||
* - create
|
||||
* - shardCollection
|
||||
* - reshardCollection
|
||||
* - refineCollectionShardKey
|
||||
*/
|
||||
showExpandedEvents?: boolean;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export interface ChangeStreamNameSpace {
|
||||
db: string;
|
||||
coll: string;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export interface ChangeStreamDocumentKey<TSchema extends Document = Document> {
|
||||
/**
|
||||
* For unsharded collections this contains a single field `_id`.
|
||||
* For sharded collections, this will contain all the components of the shard key
|
||||
*/
|
||||
documentKey: { _id: InferIdType<TSchema>; [shardKey: string]: any };
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export interface ChangeStreamDocumentCommon {
|
||||
/**
|
||||
* The id functions as an opaque token for use when resuming an interrupted
|
||||
* change stream.
|
||||
*/
|
||||
_id: ResumeToken;
|
||||
/**
|
||||
* The timestamp from the oplog entry associated with the event.
|
||||
* For events that happened as part of a multi-document transaction, the associated change stream
|
||||
* notifications will have the same clusterTime value, namely the time when the transaction was committed.
|
||||
* On a sharded cluster, events that occur on different shards can have the same clusterTime but be
|
||||
* associated with different transactions or even not be associated with any transaction.
|
||||
* To identify events for a single transaction, you can use the combination of lsid and txnNumber in the change stream event document.
|
||||
*/
|
||||
clusterTime?: Timestamp;
|
||||
|
||||
/**
|
||||
* The transaction number.
|
||||
* Only present if the operation is part of a multi-document transaction.
|
||||
*
|
||||
* **NOTE:** txnNumber can be a Long if promoteLongs is set to false
|
||||
*/
|
||||
txnNumber?: number;
|
||||
|
||||
/**
|
||||
* The identifier for the session associated with the transaction.
|
||||
* Only present if the operation is part of a multi-document transaction.
|
||||
*/
|
||||
lsid?: ServerSessionId;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export interface ChangeStreamDocumentCollectionUUID {
|
||||
/**
|
||||
* The UUID (Binary subtype 4) of the collection that the operation was performed on.
|
||||
*
|
||||
* Only present when the `showExpandedEvents` flag is enabled.
|
||||
*
|
||||
* **NOTE:** collectionUUID will be converted to a NodeJS Buffer if the promoteBuffers
|
||||
* flag is enabled.
|
||||
*
|
||||
* @since 6.1.0
|
||||
*/
|
||||
collectionUUID: Binary;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export interface ChangeStreamDocumentOperationDescription {
|
||||
/**
|
||||
* An description of the operation.
|
||||
*
|
||||
* Only present when the `showExpandedEvents` flag is enabled.
|
||||
*
|
||||
* @since 6.1.0
|
||||
*/
|
||||
operationDescription?: Document;
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/#insert-event
|
||||
*/
|
||||
export interface ChangeStreamInsertDocument<TSchema extends Document = Document>
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentKey<TSchema>,
|
||||
ChangeStreamDocumentCollectionUUID {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'insert';
|
||||
/** This key will contain the document being inserted */
|
||||
fullDocument: TSchema;
|
||||
/** Namespace the insert event occured on */
|
||||
ns: ChangeStreamNameSpace;
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/#update-event
|
||||
*/
|
||||
export interface ChangeStreamUpdateDocument<TSchema extends Document = Document>
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentKey<TSchema>,
|
||||
ChangeStreamDocumentCollectionUUID {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'update';
|
||||
/**
|
||||
* This is only set if `fullDocument` is set to `'updateLookup'`
|
||||
* Contains the point-in-time post-image of the modified document if the
|
||||
* post-image is available and either 'required' or 'whenAvailable' was
|
||||
* specified for the 'fullDocument' option when creating the change stream.
|
||||
*/
|
||||
fullDocument?: TSchema;
|
||||
/** Contains a description of updated and removed fields in this operation */
|
||||
updateDescription: UpdateDescription<TSchema>;
|
||||
/** Namespace the update event occured on */
|
||||
ns: ChangeStreamNameSpace;
|
||||
/**
|
||||
* Contains the pre-image of the modified or deleted document if the
|
||||
* pre-image is available for the change event and either 'required' or
|
||||
* 'whenAvailable' was specified for the 'fullDocumentBeforeChange' option
|
||||
* when creating the change stream. If 'whenAvailable' was specified but the
|
||||
* pre-image is unavailable, this will be explicitly set to null.
|
||||
*/
|
||||
fullDocumentBeforeChange?: TSchema;
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/#replace-event
|
||||
*/
|
||||
export interface ChangeStreamReplaceDocument<TSchema extends Document = Document>
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentKey<TSchema> {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'replace';
|
||||
/** The fullDocument of a replace event represents the document after the insert of the replacement document */
|
||||
fullDocument: TSchema;
|
||||
/** Namespace the replace event occured on */
|
||||
ns: ChangeStreamNameSpace;
|
||||
/**
|
||||
* Contains the pre-image of the modified or deleted document if the
|
||||
* pre-image is available for the change event and either 'required' or
|
||||
* 'whenAvailable' was specified for the 'fullDocumentBeforeChange' option
|
||||
* when creating the change stream. If 'whenAvailable' was specified but the
|
||||
* pre-image is unavailable, this will be explicitly set to null.
|
||||
*/
|
||||
fullDocumentBeforeChange?: TSchema;
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/#delete-event
|
||||
*/
|
||||
export interface ChangeStreamDeleteDocument<TSchema extends Document = Document>
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentKey<TSchema>,
|
||||
ChangeStreamDocumentCollectionUUID {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'delete';
|
||||
/** Namespace the delete event occured on */
|
||||
ns: ChangeStreamNameSpace;
|
||||
/**
|
||||
* Contains the pre-image of the modified or deleted document if the
|
||||
* pre-image is available for the change event and either 'required' or
|
||||
* 'whenAvailable' was specified for the 'fullDocumentBeforeChange' option
|
||||
* when creating the change stream. If 'whenAvailable' was specified but the
|
||||
* pre-image is unavailable, this will be explicitly set to null.
|
||||
*/
|
||||
fullDocumentBeforeChange?: TSchema;
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/#drop-event
|
||||
*/
|
||||
export interface ChangeStreamDropDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'drop';
|
||||
/** Namespace the drop event occured on */
|
||||
ns: ChangeStreamNameSpace;
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/#rename-event
|
||||
*/
|
||||
export interface ChangeStreamRenameDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'rename';
|
||||
/** The new name for the `ns.coll` collection */
|
||||
to: { db: string; coll: string };
|
||||
/** The "from" namespace that the rename occured on */
|
||||
ns: ChangeStreamNameSpace;
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/#dropdatabase-event
|
||||
*/
|
||||
export interface ChangeStreamDropDatabaseDocument extends ChangeStreamDocumentCommon {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'dropDatabase';
|
||||
/** The database dropped */
|
||||
ns: { db: string };
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/#invalidate-event
|
||||
*/
|
||||
export interface ChangeStreamInvalidateDocument extends ChangeStreamDocumentCommon {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'invalidate';
|
||||
}
|
||||
|
||||
/**
|
||||
* Only present when the `showExpandedEvents` flag is enabled.
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/
|
||||
*/
|
||||
export interface ChangeStreamCreateIndexDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID,
|
||||
ChangeStreamDocumentOperationDescription {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'createIndexes';
|
||||
}
|
||||
|
||||
/**
|
||||
* Only present when the `showExpandedEvents` flag is enabled.
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/
|
||||
*/
|
||||
export interface ChangeStreamDropIndexDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID,
|
||||
ChangeStreamDocumentOperationDescription {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'dropIndexes';
|
||||
}
|
||||
|
||||
/**
|
||||
* Only present when the `showExpandedEvents` flag is enabled.
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/
|
||||
*/
|
||||
export interface ChangeStreamCollModDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'modify';
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/
|
||||
*/
|
||||
export interface ChangeStreamCreateDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'create';
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/
|
||||
*/
|
||||
export interface ChangeStreamShardCollectionDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID,
|
||||
ChangeStreamDocumentOperationDescription {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'shardCollection';
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/
|
||||
*/
|
||||
export interface ChangeStreamReshardCollectionDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID,
|
||||
ChangeStreamDocumentOperationDescription {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'reshardCollection';
|
||||
}
|
||||
|
||||
/**
|
||||
* @public
|
||||
* @see https://www.mongodb.com/docs/manual/reference/change-events/
|
||||
*/
|
||||
export interface ChangeStreamRefineCollectionShardKeyDocument
|
||||
extends ChangeStreamDocumentCommon,
|
||||
ChangeStreamDocumentCollectionUUID,
|
||||
ChangeStreamDocumentOperationDescription {
|
||||
/** Describes the type of operation represented in this change notification */
|
||||
operationType: 'refineCollectionShardKey';
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export type ChangeStreamDocument<TSchema extends Document = Document> =
|
||||
| ChangeStreamInsertDocument<TSchema>
|
||||
| ChangeStreamUpdateDocument<TSchema>
|
||||
| ChangeStreamReplaceDocument<TSchema>
|
||||
| ChangeStreamDeleteDocument<TSchema>
|
||||
| ChangeStreamDropDocument
|
||||
| ChangeStreamRenameDocument
|
||||
| ChangeStreamDropDatabaseDocument
|
||||
| ChangeStreamInvalidateDocument
|
||||
| ChangeStreamCreateIndexDocument
|
||||
| ChangeStreamCreateDocument
|
||||
| ChangeStreamCollModDocument
|
||||
| ChangeStreamDropIndexDocument
|
||||
| ChangeStreamShardCollectionDocument
|
||||
| ChangeStreamReshardCollectionDocument
|
||||
| ChangeStreamRefineCollectionShardKeyDocument;
|
||||
|
||||
/** @public */
|
||||
export interface UpdateDescription<TSchema extends Document = Document> {
|
||||
/**
|
||||
* A document containing key:value pairs of names of the fields that were
|
||||
* changed, and the new value for those fields.
|
||||
*/
|
||||
updatedFields?: Partial<TSchema>;
|
||||
|
||||
/**
|
||||
* An array of field names that were removed from the document.
|
||||
*/
|
||||
removedFields?: string[];
|
||||
|
||||
/**
|
||||
* An array of documents which record array truncations performed with pipeline-based updates using one or more of the following stages:
|
||||
* - $addFields
|
||||
* - $set
|
||||
* - $replaceRoot
|
||||
* - $replaceWith
|
||||
*/
|
||||
truncatedArrays?: Array<{
|
||||
/** The name of the truncated field. */
|
||||
field: string;
|
||||
/** The number of elements in the truncated array. */
|
||||
newSize: number;
|
||||
}>;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export type ChangeStreamEvents<
|
||||
TSchema extends Document = Document,
|
||||
TChange extends Document = ChangeStreamDocument<TSchema>
|
||||
> = {
|
||||
resumeTokenChanged(token: ResumeToken): void;
|
||||
init(response: any): void;
|
||||
more(response?: any): void;
|
||||
response(): void;
|
||||
end(): void;
|
||||
error(error: Error): void;
|
||||
change(change: TChange): void;
|
||||
} & AbstractCursorEvents;
|
||||
|
||||
/**
|
||||
* Creates a new Change Stream instance. Normally created using {@link Collection#watch|Collection.watch()}.
|
||||
* @public
|
||||
*/
|
||||
export class ChangeStream<
|
||||
TSchema extends Document = Document,
|
||||
TChange extends Document = ChangeStreamDocument<TSchema>
|
||||
> extends TypedEventEmitter<ChangeStreamEvents<TSchema, TChange>> {
|
||||
pipeline: Document[];
|
||||
options: ChangeStreamOptions;
|
||||
parent: MongoClient | Db | Collection;
|
||||
namespace: MongoDBNamespace;
|
||||
type: symbol;
|
||||
/** @internal */
|
||||
cursor: ChangeStreamCursor<TSchema, TChange>;
|
||||
streamOptions?: CursorStreamOptions;
|
||||
/** @internal */
|
||||
[kCursorStream]?: Readable & AsyncIterable<TChange>;
|
||||
/** @internal */
|
||||
[kClosed]: boolean;
|
||||
/** @internal */
|
||||
[kMode]: false | 'iterator' | 'emitter';
|
||||
|
||||
/** @event */
|
||||
static readonly RESPONSE = RESPONSE;
|
||||
/** @event */
|
||||
static readonly MORE = MORE;
|
||||
/** @event */
|
||||
static readonly INIT = INIT;
|
||||
/** @event */
|
||||
static readonly CLOSE = CLOSE;
|
||||
/**
|
||||
* Fired for each new matching change in the specified namespace. Attaching a `change`
|
||||
* event listener to a Change Stream will switch the stream into flowing mode. Data will
|
||||
* then be passed as soon as it is available.
|
||||
* @event
|
||||
*/
|
||||
static readonly CHANGE = CHANGE;
|
||||
/** @event */
|
||||
static readonly END = END;
|
||||
/** @event */
|
||||
static readonly ERROR = ERROR;
|
||||
/**
|
||||
* Emitted each time the change stream stores a new resume token.
|
||||
* @event
|
||||
*/
|
||||
static readonly RESUME_TOKEN_CHANGED = RESUME_TOKEN_CHANGED;
|
||||
|
||||
/**
|
||||
* @internal
|
||||
*
|
||||
* @param parent - The parent object that created this change stream
|
||||
* @param pipeline - An array of {@link https://docs.mongodb.com/manual/reference/operator/aggregation-pipeline/|aggregation pipeline stages} through which to pass change stream documents
|
||||
*/
|
||||
constructor(
|
||||
parent: OperationParent,
|
||||
pipeline: Document[] = [],
|
||||
options: ChangeStreamOptions = {}
|
||||
) {
|
||||
super();
|
||||
|
||||
this.pipeline = pipeline;
|
||||
this.options = options;
|
||||
|
||||
if (parent instanceof Collection) {
|
||||
this.type = CHANGE_DOMAIN_TYPES.COLLECTION;
|
||||
} else if (parent instanceof Db) {
|
||||
this.type = CHANGE_DOMAIN_TYPES.DATABASE;
|
||||
} else if (parent instanceof MongoClient) {
|
||||
this.type = CHANGE_DOMAIN_TYPES.CLUSTER;
|
||||
} else {
|
||||
throw new MongoChangeStreamError(
|
||||
'Parent provided to ChangeStream constructor must be an instance of Collection, Db, or MongoClient'
|
||||
);
|
||||
}
|
||||
|
||||
this.parent = parent;
|
||||
this.namespace = parent.s.namespace;
|
||||
if (!this.options.readPreference && parent.readPreference) {
|
||||
this.options.readPreference = parent.readPreference;
|
||||
}
|
||||
|
||||
// Create contained Change Stream cursor
|
||||
this.cursor = this._createChangeStreamCursor(options);
|
||||
|
||||
this[kClosed] = false;
|
||||
this[kMode] = false;
|
||||
|
||||
// Listen for any `change` listeners being added to ChangeStream
|
||||
this.on('newListener', eventName => {
|
||||
if (eventName === 'change' && this.cursor && this.listenerCount('change') === 0) {
|
||||
this._streamEvents(this.cursor);
|
||||
}
|
||||
});
|
||||
|
||||
this.on('removeListener', eventName => {
|
||||
if (eventName === 'change' && this.listenerCount('change') === 0 && this.cursor) {
|
||||
this[kCursorStream]?.removeAllListeners('data');
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
get cursorStream(): (Readable & AsyncIterable<TChange>) | undefined {
|
||||
return this[kCursorStream];
|
||||
}
|
||||
|
||||
/** The cached resume token that is used to resume after the most recently returned change. */
|
||||
get resumeToken(): ResumeToken {
|
||||
return this.cursor?.resumeToken;
|
||||
}
|
||||
|
||||
/** Check if there is any document still available in the Change Stream */
|
||||
hasNext(): Promise<boolean>;
|
||||
hasNext(callback: Callback<boolean>): void;
|
||||
hasNext(callback?: Callback): Promise<boolean> | void {
|
||||
this._setIsIterator();
|
||||
// TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
|
||||
// Shadowing is intentional here. We want to override the `callback` variable
|
||||
// from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
|
||||
return maybePromise(callback, callback => {
|
||||
(async () => {
|
||||
try {
|
||||
const hasNext = await this.cursor.hasNext();
|
||||
return hasNext;
|
||||
} catch (error) {
|
||||
try {
|
||||
await this._processErrorIteratorMode(error);
|
||||
const hasNext = await this.cursor.hasNext();
|
||||
return hasNext;
|
||||
} catch (error) {
|
||||
await this.close().catch(err => err);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
})().then(
|
||||
hasNext => callback(undefined, hasNext),
|
||||
error => callback(error)
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/** Get the next available document from the Change Stream. */
|
||||
next(): Promise<TChange>;
|
||||
next(callback: Callback<TChange>): void;
|
||||
next(callback?: Callback<TChange>): Promise<TChange> | void {
|
||||
this._setIsIterator();
|
||||
// TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
|
||||
// Shadowing is intentional here. We want to override the `callback` variable
|
||||
// from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
|
||||
return maybePromise(callback, callback => {
|
||||
(async () => {
|
||||
try {
|
||||
const change = await this.cursor.next();
|
||||
const processedChange = this._processChange(change ?? null);
|
||||
return processedChange;
|
||||
} catch (error) {
|
||||
try {
|
||||
await this._processErrorIteratorMode(error);
|
||||
const change = await this.cursor.next();
|
||||
const processedChange = this._processChange(change ?? null);
|
||||
return processedChange;
|
||||
} catch (error) {
|
||||
await this.close().catch(err => err);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
})().then(
|
||||
change => callback(undefined, change),
|
||||
error => callback(error)
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Try to get the next available document from the Change Stream's cursor or `null` if an empty batch is returned
|
||||
*/
|
||||
tryNext(): Promise<Document | null>;
|
||||
tryNext(callback: Callback<Document | null>): void;
|
||||
tryNext(callback?: Callback<Document | null>): Promise<Document | null> | void {
|
||||
this._setIsIterator();
|
||||
// TOOD(NODE-4319): Add eslint rule preventing accidental variable shadowing
|
||||
// Shadowing is intentional here. We want to override the `callback` variable
|
||||
// from the outer scope so that the inner scope doesn't accidentally call the wrong callback.
|
||||
return maybePromise(callback, callback => {
|
||||
(async () => {
|
||||
try {
|
||||
const change = await this.cursor.tryNext();
|
||||
return change ?? null;
|
||||
} catch (error) {
|
||||
try {
|
||||
await this._processErrorIteratorMode(error);
|
||||
const change = await this.cursor.tryNext();
|
||||
return change ?? null;
|
||||
} catch (error) {
|
||||
await this.close().catch(err => err);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
})().then(
|
||||
change => callback(undefined, change),
|
||||
error => callback(error)
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/** Is the cursor closed */
|
||||
get closed(): boolean {
|
||||
return this[kClosed] || this.cursor.closed;
|
||||
}
|
||||
|
||||
/** Close the Change Stream */
|
||||
close(): Promise<void>;
|
||||
close(callback: Callback): void;
|
||||
close(callback?: Callback): Promise<void> | void {
|
||||
this[kClosed] = true;
|
||||
|
||||
return maybePromise(callback, cb => {
|
||||
const cursor = this.cursor;
|
||||
return cursor.close(err => {
|
||||
this._endStream();
|
||||
return cb(err);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Return a modified Readable stream including a possible transform method.
|
||||
*
|
||||
* NOTE: When using a Stream to process change stream events, the stream will
|
||||
* NOT automatically resume in the case a resumable error is encountered.
|
||||
*
|
||||
* @throws MongoChangeStreamError if the underlying cursor or the change stream is closed
|
||||
*/
|
||||
stream(options?: CursorStreamOptions): Readable & AsyncIterable<TChange> {
|
||||
if (this.closed) {
|
||||
throw new MongoChangeStreamError(CHANGESTREAM_CLOSED_ERROR);
|
||||
}
|
||||
|
||||
this.streamOptions = options;
|
||||
return this.cursor.stream(options);
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
private _setIsEmitter(): void {
|
||||
if (this[kMode] === 'iterator') {
|
||||
// TODO(NODE-3485): Replace with MongoChangeStreamModeError
|
||||
throw new MongoAPIError(
|
||||
'ChangeStream cannot be used as an EventEmitter after being used as an iterator'
|
||||
);
|
||||
}
|
||||
this[kMode] = 'emitter';
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
private _setIsIterator(): void {
|
||||
if (this[kMode] === 'emitter') {
|
||||
// TODO(NODE-3485): Replace with MongoChangeStreamModeError
|
||||
throw new MongoAPIError(
|
||||
'ChangeStream cannot be used as an iterator after being used as an EventEmitter'
|
||||
);
|
||||
}
|
||||
this[kMode] = 'iterator';
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new change stream cursor based on self's configuration
|
||||
* @internal
|
||||
*/
|
||||
private _createChangeStreamCursor(
|
||||
options: ChangeStreamOptions | ChangeStreamCursorOptions
|
||||
): ChangeStreamCursor<TSchema, TChange> {
|
||||
const changeStreamStageOptions = filterOptions(options, CHANGE_STREAM_OPTIONS);
|
||||
if (this.type === CHANGE_DOMAIN_TYPES.CLUSTER) {
|
||||
changeStreamStageOptions.allChangesForCluster = true;
|
||||
}
|
||||
const pipeline = [{ $changeStream: changeStreamStageOptions }, ...this.pipeline];
|
||||
|
||||
const client: MongoClient | null =
|
||||
this.type === CHANGE_DOMAIN_TYPES.CLUSTER
|
||||
? (this.parent as MongoClient)
|
||||
: this.type === CHANGE_DOMAIN_TYPES.DATABASE
|
||||
? (this.parent as Db).s.client
|
||||
: this.type === CHANGE_DOMAIN_TYPES.COLLECTION
|
||||
? (this.parent as Collection).s.db.s.client
|
||||
: null;
|
||||
|
||||
if (client == null) {
|
||||
// This should never happen because of the assertion in the constructor
|
||||
throw new MongoRuntimeError(
|
||||
`Changestream type should only be one of cluster, database, collection. Found ${this.type.toString()}`
|
||||
);
|
||||
}
|
||||
|
||||
const changeStreamCursor = new ChangeStreamCursor<TSchema, TChange>(
|
||||
client,
|
||||
this.namespace,
|
||||
pipeline,
|
||||
options
|
||||
);
|
||||
|
||||
for (const event of CHANGE_STREAM_EVENTS) {
|
||||
changeStreamCursor.on(event, e => this.emit(event, e));
|
||||
}
|
||||
|
||||
if (this.listenerCount(ChangeStream.CHANGE) > 0) {
|
||||
this._streamEvents(changeStreamCursor);
|
||||
}
|
||||
|
||||
return changeStreamCursor;
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
private _closeEmitterModeWithError(error: AnyError): void {
|
||||
this.emit(ChangeStream.ERROR, error);
|
||||
|
||||
this.close(() => {
|
||||
// nothing to do
|
||||
});
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
private _streamEvents(cursor: ChangeStreamCursor<TSchema, TChange>): void {
|
||||
this._setIsEmitter();
|
||||
const stream = this[kCursorStream] ?? cursor.stream();
|
||||
this[kCursorStream] = stream;
|
||||
stream.on('data', change => {
|
||||
try {
|
||||
const processedChange = this._processChange(change);
|
||||
this.emit(ChangeStream.CHANGE, processedChange);
|
||||
} catch (error) {
|
||||
this.emit(ChangeStream.ERROR, error);
|
||||
}
|
||||
});
|
||||
stream.on('error', error => this._processErrorStreamMode(error));
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
private _endStream(): void {
|
||||
const cursorStream = this[kCursorStream];
|
||||
if (cursorStream) {
|
||||
['data', 'close', 'end', 'error'].forEach(event => cursorStream.removeAllListeners(event));
|
||||
cursorStream.destroy();
|
||||
}
|
||||
|
||||
this[kCursorStream] = undefined;
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
private _processChange(change: TChange | null): TChange {
|
||||
if (this[kClosed]) {
|
||||
// TODO(NODE-3485): Replace with MongoChangeStreamClosedError
|
||||
throw new MongoAPIError(CHANGESTREAM_CLOSED_ERROR);
|
||||
}
|
||||
|
||||
// a null change means the cursor has been notified, implicitly closing the change stream
|
||||
if (change == null) {
|
||||
// TODO(NODE-3485): Replace with MongoChangeStreamClosedError
|
||||
throw new MongoRuntimeError(CHANGESTREAM_CLOSED_ERROR);
|
||||
}
|
||||
|
||||
if (change && !change._id) {
|
||||
throw new MongoChangeStreamError(NO_RESUME_TOKEN_ERROR);
|
||||
}
|
||||
|
||||
// cache the resume token
|
||||
this.cursor.cacheResumeToken(change._id);
|
||||
|
||||
// wipe the startAtOperationTime if there was one so that there won't be a conflict
|
||||
// between resumeToken and startAtOperationTime if we need to reconnect the cursor
|
||||
this.options.startAtOperationTime = undefined;
|
||||
|
||||
return change;
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
private _processErrorStreamMode(changeStreamError: AnyError) {
|
||||
// If the change stream has been closed explicitly, do not process error.
|
||||
if (this[kClosed]) return;
|
||||
|
||||
if (isResumableError(changeStreamError, this.cursor.maxWireVersion)) {
|
||||
this._endStream();
|
||||
this.cursor.close().catch(() => null);
|
||||
|
||||
const topology = getTopology(this.parent);
|
||||
topology.selectServer(this.cursor.readPreference, {}, serverSelectionError => {
|
||||
if (serverSelectionError) return this._closeEmitterModeWithError(changeStreamError);
|
||||
this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);
|
||||
});
|
||||
} else {
|
||||
this._closeEmitterModeWithError(changeStreamError);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @internal
|
||||
*
|
||||
* TODO(NODE-4320): promisify selectServer and refactor this code to be async
|
||||
*
|
||||
* we promisify _processErrorIteratorModeCallback until we have a promisifed version of selectServer.
|
||||
*/
|
||||
// eslint-disable-next-line @typescript-eslint/unbound-method
|
||||
private _processErrorIteratorMode = promisify(this._processErrorIteratorModeCallback);
|
||||
|
||||
/** @internal */
|
||||
private _processErrorIteratorModeCallback(changeStreamError: AnyError, callback: Callback) {
|
||||
if (this[kClosed]) {
|
||||
// TODO(NODE-3485): Replace with MongoChangeStreamClosedError
|
||||
return callback(new MongoAPIError(CHANGESTREAM_CLOSED_ERROR));
|
||||
}
|
||||
|
||||
if (isResumableError(changeStreamError, this.cursor.maxWireVersion)) {
|
||||
this.cursor.close().catch(() => null);
|
||||
|
||||
const topology = getTopology(this.parent);
|
||||
topology.selectServer(this.cursor.readPreference, {}, serverSelectionError => {
|
||||
// if the topology can't reconnect, close the stream
|
||||
if (serverSelectionError) return this.close(() => callback(changeStreamError));
|
||||
|
||||
this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);
|
||||
callback();
|
||||
});
|
||||
} else {
|
||||
this.close(() => callback(changeStreamError));
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user