Method getLateFeed

db/changes.go:1280–1331

Feed to process late sequences for the channel. Updates lastSequence as it works the feed. Error indicates previous position in late sequence feed isn't available, and caller should reset to low sequence.

(ctx context.Context, feedHandler *lateSequenceFeed, singleChannelCache SingleChannelCache, versionType ChangesVersionType)

// Feed to process late sequences for the channel.  Updates lastSequence as it works the feed.  Error indicates
// previous position in late sequence feed isn't available, and caller should reset to low sequence.
func (db *DatabaseCollectionWithUser) getLateFeed(ctx context.Context, feedHandler *lateSequenceFeed, singleChannelCache SingleChannelCache, versionType ChangesVersionType) (<-chan *ChangeEntry, error) {

	if !singleChannelCache.SupportsLateFeed() {
		return nil, errors.New("Cache doesn't support late feeds")
	}
	// If the associated cache instance for this feedHandler doesn't match SingleChannelCache, it means the channel cache
	// has been evicted/recreated, and the current feedHandler is no longer valid
	if feedHandler.lateSequenceUUID != singleChannelCache.LateSequenceUUID() {
		return nil, errors.New("Cache/handler mismatch")
	}

	// Use LogPriorityQueue for late entries, to utilize the existing Len/Less/Swap methods on LogPriorityQueue for sort
	var logs LogPriorityQueue
	logs, lastSequence, err := singleChannelCache.GetLateSequencesSince(feedHandler.lastSequence)
	if err != nil {
		return nil, err
	}
	if logs == nil || len(logs) == 0 {
		// There are no late entries newer than lastSequence
		feed := make(chan *ChangeEntry)
		close(feed)
		return feed, nil
	}

	// Sort late sequences, to ensure duplicates aren't sent in a single continuous _changes iteration when multiple
	// channels have late arrivals
	sort.Sort(logs)

	feed := make(chan *ChangeEntry, 1)
	go func() {
		defer close(feed)
		// Write each log entry to the 'feed' channel in turn:
		for _, logEntry := range logs {
			// We don't need TriggeredBy handling here, because when backfilling from a
			// channel in response to a user being added to the channel, we don't need to worry about
			// late arrived sequences
			seqID := SequenceID{
				Seq: logEntry.Sequence,
			}
			change := makeChangeEntry(ctx, logEntry, seqID, singleChannelCache.ChannelID(), versionType)
			select {
			case <-ctx.Done():
				return

			case feed <- &change:
			}
		}
	}()

	feedHandler.lastSequence = lastSequence
	return feed, nil
}

// Closes a single late sequence feed.
func (db *DatabaseCollectionWithUser) closeLateFeed(ctx context.Context, feedHandler *lateSequenceFeed) {