Method constructor

src/change_stream.ts:641–700

* @internal * * @param parent - The parent object that created this change stream * @param pipeline - An array of {@link https://www.mongodb.com/docs/manual/reference/operator/aggregation-pipeline/|aggregation pipeline stages} through which to pass change stream documents

(
    parent: OperationParent,
    pipeline: Document[] = [],
    options: ChangeStreamOptions = {}
  )

   * @param pipeline - An array of {@link https://www.mongodb.com/docs/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 };
    let serverSelectionTimeoutMS: number;
    delete this.options.writeConcern;

    if (parent instanceof Collection) {
      this.type = CHANGE_DOMAIN_TYPES.COLLECTION;
      serverSelectionTimeoutMS = parent.s.db.client.options.serverSelectionTimeoutMS;
    } else if (parent instanceof Db) {
      this.type = CHANGE_DOMAIN_TYPES.DATABASE;
      serverSelectionTimeoutMS = parent.client.options.serverSelectionTimeoutMS;
    } else if (parent instanceof MongoClient) {
      this.type = CHANGE_DOMAIN_TYPES.CLUSTER;
      serverSelectionTimeoutMS = parent.options.serverSelectionTimeoutMS;
    } else {
      throw new MongoChangeStreamError(
        'Parent provided to ChangeStream constructor must be an instance of Collection, Db, or MongoClient'
      );
    }

    this.contextOwner = Symbol();
    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.isClosed = false;
    this.mode = 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.cursorStream?.removeAllListeners('data');
      }
    });

    if (this.options.timeoutMS != null) {
      this.timeoutContext = new CSOTTimeoutContext({
        timeoutMS: this.options.timeoutMS,
        serverSelectionTimeoutMS
      });