Method enumerateQueuedBlobs

pkg/server/sync.go:543–560

enumerateQueuedBlobs yields blobs from the on-disk sorted.KeyValue store. This differs from enumeratePendingBlobs, which sends from the in-memory pending list.

(dst chan<- blob.SizedRef, intr <-chan struct{})

// enumerateQueuedBlobs yields blobs from the on-disk sorted.KeyValue store.
// This differs from enumeratePendingBlobs, which sends from the in-memory pending list.
func (sh *SyncHandler) enumerateQueuedBlobs(dst chan<- blob.SizedRef, intr <-chan struct{}) error {
	defer close(dst)
	it := sh.queue.Find("", "")
	for it.Next() {
		br, ok := blob.Parse(it.Key())
		size, err := strconv.ParseUint(it.Value(), 10, 32)
		if !ok || err != nil {
			sh.logf("ERROR: bogus sync queue entry: %q => %q", it.Key(), it.Value())
			continue
		}
		select {
		case dst <- blob.SizedRef{Ref: br, Size: uint32(size)}:
		case <-intr:
			return it.Close()
		}
	}
	return it.Close()
}

func (sh *SyncHandler) runSync(syncType string, enumSrc func(chan<- blob.SizedRef, <-chan struct{}) error) int {
	enumch := make(chan blob.SizedRef, 8)