Method _createChangeStreamCursor

src/change_stream.ts:902–948

* Create a new change stream cursor based on self's configuration * @internal

(
    options: ChangeStreamOptions | ChangeStreamCursorOptions
  )

   * @internal
   */
  private _createChangeStreamCursor(
    options: ChangeStreamOptions | ChangeStreamCursorOptions
  ): ChangeStreamCursor<TSchema, TChange> {
    const changeStreamStageOptions: Document = filterOutOptions(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).client
          : this.type === CHANGE_DOMAIN_TYPES.COLLECTION
            ? (this.parent as Collection).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,
        timeoutContext: this.timeoutContext
          ? new CursorTimeoutContext(this.timeoutContext, this.contextOwner)
          : undefined
      }
    );

    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 {