Method _streamEvents

src/change_stream.ts:958–972

@internal

(cursor: ChangeStreamCursor<TSchema, TChange>)

  /** @internal */
  private _streamEvents(cursor: ChangeStreamCursor<TSchema, TChange>): void {
    this._setIsEmitter();
    const stream = this.cursorStream ?? cursor.stream();
    this.cursorStream = stream;
    stream.on('data', change => {
      try {
        const processedChange = this._processChange(change);
        this.emit(ChangeStream.CHANGE, processedChange);
      } catch (error) {
        this.emit(ChangeStream.ERROR, error);
      }
      this.timeoutContext?.refresh();
    });
    stream.on('error', error => this._processErrorStreamMode(error, this.cursor.id != null));
  }

  /** @internal */
  private _endStream(): void {