lib/change_stream.js
function(self) {
  if (self.resumeToken) {
    self.options.resumeAfter = self.resumeToken;
  }

  var changeStreamCursor = buildChangeStreamAggregationCommand(self);

  /**
   * 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 ChangeStream#change
   * @type {object}
   */
  if (self.listenerCount('change') > 0) {
    changeStreamCursor.on('data', function(change) {
      processNewChange({ changeStream: self, change, eventEmitter: true });
    });
  }

  /**
   * Change stream close event
   *
   * @event ChangeStream#close
   * @type {null}
   */
  changeStreamCursor.on('close', function() {
    self.emit('close');
  });

  /**
   * Change stream end event
   *
   * @event ChangeStream#end
   * @type {null}
   */
  changeStreamCursor.on('end', function() {
    self.emit('end');
  });

  /**
   * Fired when the stream encounters an error.
   *
   * @event ChangeStream#error
   * @type {Error}
   */
  changeStreamCursor.on('error', function(error) {
    processNewChange({ changeStream: self, error, eventEmitter: true });
  });

  if (self.pipeDestinations) {
    const cursorStream = changeStreamCursor.stream(self.streamOptions);
    for (let pipeDestination in self.pipeDestinations) {
      cursorStream.pipe(pipeDestination);
    }
  }

  return changeStreamCursor;
}
function(self) {
  if (self.resumeToken) {
    self.options.resumeAfter = self.resumeToken;
  }

  var changeStreamCursor = buildChangeStreamAggregationCommand(self);

  /**
   * 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 ChangeStream#change
   * @type {object}
   */
  if (self.listenerCount('change') > 0) {
    changeStreamCursor.on('data', function(change) {
      processNewChange({ changeStream: self, change, eventEmitter: true });
    });
  }

  /**
   * Change stream close event
   *
   * @event ChangeStream#close
   * @type {null}
   */
  changeStreamCursor.on('close', function() {
    self.emit('close');
  });

  /**
   * Change stream end event
   *
   * @event ChangeStream#end
   * @type {null}
   */
  changeStreamCursor.on('end', function() {
    self.emit('end');
  });

  /**
   * Fired when the stream encounters an error.
   *
   * @event ChangeStream#error
   * @type {Error}
   */
  changeStreamCursor.on('error', function(error) {
    processNewChange({ changeStream: self, error, eventEmitter: true });
  });

  if (self.pipeDestinations) {
    const cursorStream = changeStreamCursor.stream(self.streamOptions);
    for (let pipeDestination in self.pipeDestinations) {
      cursorStream.pipe(pipeDestination);
    }
  }

  return changeStreamCursor;
}
function ( self ) { if ( self . resumeToken ) { self . options . resumeAfter = self . resumeToken ; } var changeStreamCursor = buildChangeStreamAggregationCommand ( self ) ; /**
   * 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 ChangeStream#change
   * @type {object}
   */ if ( self . listenerCount ( 'change' ) > 0 ) { changeStreamCursor . on ( 'data' , function ( change ) { processNewChange ( { changeStream : self , change , eventEmitter : true } ) ; } ) ; } /**
   * Change stream close event
   *
   * @event ChangeStream#close
   * @type {null}
   */ changeStreamCursor . on ( 'close' , function ( ) { self . emit ( 'close' ) ; } ) ; /**
   * Change stream end event
   *
   * @event ChangeStream#end
   * @type {null}
   */ changeStreamCursor . on ( 'end' , function ( ) { self . emit ( 'end' ) ; } ) ; /**
   * Fired when the stream encounters an error.
   *
   * @event ChangeStream#error
   * @type {Error}
   */ changeStreamCursor . on ( 'error' , function ( error ) { processNewChange ( { changeStream : self , error , eventEmitter : true } ) ; } ) ; if ( self . pipeDestinations ) { const cursorStream = changeStreamCursor . stream ( self . streamOptions ) ; for ( let pipeDestination in self . pipeDestinations ) { cursorStream . pipe ( pipeDestination ) ; } } return changeStreamCursor ; }
Create a new change stream cursor based on self's configuration
Create a new change stream cursor based on self s configuration