code: 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;
  }
docs: Create a new change stream cursor based on self's configuration