meteor/mongo-livedata/oplog_observe_driver.js
function (doc) {
    var self = this;
    var id = doc._id;
    if (self._published.has(id))
      throw Error("tried to add something already published " + id);
    if (self._limit && self._unpublishedBuffer.has(id))
      throw Error("tried to add something already existed in buffer " + id);

    var limit = self._limit;
    var comparator = self._comparator;
    var maxPublished = (limit && self._published.size() > 0) ?
      self._published.get(self._published.maxElementId()) : null;
    var maxBuffered = (limit && self._unpublishedBuffer.size() > 0) ?
      self._unpublishedBuffer.get(self._unpublishedBuffer.maxElementId()) : null;
    // The query is unlimited or didn't publish enough documents yet or the new
    // document would fit into published set pushing the maximum element out,
    // then we need to publish the doc.
    var toPublish = ! limit || self._published.size() < limit ||
                    comparator(doc, maxPublished) < 0;

    // Otherwise we might need to buffer it (only in case of limited query).
    // Buffering is allowed if the buffer is not filled up yet and all matching
    // docs are either in the published set or in the buffer.
    var canAppendToBuffer = !toPublish && self._safeAppendToBuffer &&
                            self._unpublishedBuffer.size() < limit;

    // Or if it is small enough to be safely inserted to the middle or the
    // beginning of the buffer.
    var canInsertIntoBuffer = !toPublish && maxBuffered &&
                              comparator(doc, maxBuffered) <= 0;

    var toBuffer = canAppendToBuffer || canInsertIntoBuffer;

    if (toPublish) {
      self._addPublished(id, doc);
    } else if (toBuffer) {
      self._addBuffered(id, doc);
    } else {
      // dropping it and not saving to the cache
      self._safeAppendToBuffer = false;
    }
  }
function (doc) {
    var self = this;
    var id = doc._id;
    if (self._published.has(id))
      throw Error("tried to add something already published " + id);
    if (self._limit && self._unpublishedBuffer.has(id))
      throw Error("tried to add something already existed in buffer " + id);

    var limit = self._limit;
    var comparator = self._comparator;
    var maxPublished = (limit && self._published.size() > 0) ?
      self._published.get(self._published.maxElementId()) : null;
    var maxBuffered = (limit && self._unpublishedBuffer.size() > 0) ?
      self._unpublishedBuffer.get(self._unpublishedBuffer.maxElementId()) : null;
    // The query is unlimited or didn't publish enough documents yet or the new
    // document would fit into published set pushing the maximum element out,
    // then we need to publish the doc.
    var toPublish = ! limit || self._published.size() < limit ||
                    comparator(doc, maxPublished) < 0;

    // Otherwise we might need to buffer it (only in case of limited query).
    // Buffering is allowed if the buffer is not filled up yet and all matching
    // docs are either in the published set or in the buffer.
    var canAppendToBuffer = !toPublish && self._safeAppendToBuffer &&
                            self._unpublishedBuffer.size() < limit;

    // Or if it is small enough to be safely inserted to the middle or the
    // beginning of the buffer.
    var canInsertIntoBuffer = !toPublish && maxBuffered &&
                              comparator(doc, maxBuffered) <= 0;

    var toBuffer = canAppendToBuffer || canInsertIntoBuffer;

    if (toPublish) {
      self._addPublished(id, doc);
    } else if (toBuffer) {
      self._addBuffered(id, doc);
    } else {
      // dropping it and not saving to the cache
      self._safeAppendToBuffer = false;
    }
  }
function ( doc ) { var self = this ; var id = doc . _id ; if ( self . _published . has ( id ) ) throw Error ( "tried to add something already published " + id ) ; if ( self . _limit && self . _unpublishedBuffer . has ( id ) ) throw Error ( "tried to add something already existed in buffer " + id ) ; var limit = self . _limit ; var comparator = self . _comparator ; var maxPublished = ( limit && self . _published . size ( ) > 0 ) ? self . _published . get ( self . _published . maxElementId ( ) ) : null ; var maxBuffered = ( limit && self . _unpublishedBuffer . size ( ) > 0 ) ? self . _unpublishedBuffer . get ( self . _unpublishedBuffer . maxElementId ( ) ) : null ; // The query is unlimited or didn't publish enough documents yet or the new // document would fit into published set pushing the maximum element out, // then we need to publish the doc. var toPublish = ! limit || self . _published . size ( ) < limit || comparator ( doc , maxPublished ) < 0 ; // Otherwise we might need to buffer it (only in case of limited query). // Buffering is allowed if the buffer is not filled up yet and all matching // docs are either in the published set or in the buffer. var canAppendToBuffer = ! toPublish && self . _safeAppendToBuffer && self . _unpublishedBuffer . size ( ) < limit ; // Or if it is small enough to be safely inserted to the middle or the // beginning of the buffer. var canInsertIntoBuffer = ! toPublish && maxBuffered && comparator ( doc , maxBuffered ) <= 0 ; var toBuffer = canAppendToBuffer || canInsertIntoBuffer ; if ( toPublish ) { self . _addPublished ( id , doc ) ; } else if ( toBuffer ) { self . _addBuffered ( id , doc ) ; } else { // dropping it and not saving to the cache self . _safeAppendToBuffer = false ; } }
Called when a document has joined the "Matching" results set. Takes responsibility of keeping _unpublishedBuffer in sync with _published and the effect of limit enforced.
Called when a document has joined the Matching results set . Takes responsibility of keeping _unpublishedBuffer in sync with _published and the effect of limit enforced .