Function ListMissingDestinationBlobs

pkg/blobserver/sync.go:30–72

ListMissingDestinationBlobs reads from 'srcch' and 'dstch' (sorted enumerations of blobs from two blob servers) and sends to 'destMissing' any blobs which appear on the source but not at the destination. destMissing is closed at the end. If an invalid (zero) blob from srcch or dstch arrives, ListM

(destMissing chan<- blob.SizedRef, sizeMismatch func(blob.Ref), srcch, dstch <-chan blob.SizedRef)

// If an invalid (zero) blob from srcch or dstch arrives,
// ListMissingDestinationBlobs stops.
func ListMissingDestinationBlobs(destMissing chan<- blob.SizedRef, sizeMismatch func(blob.Ref), srcch, dstch <-chan blob.SizedRef) {
	defer close(destMissing)

	src := &blob.ChanPeeker{Ch: srcch}
	dst := &blob.ChanPeeker{Ch: dstch}

	for {
		srcSized, ok := src.Peek()
		if !ok {
			break
		}
		if !srcSized.Ref.Valid() {
			return
		}

		// If the destination has reached its end, anything
		// remaining in the source is needed.
		if _, ok := dst.Peek(); !ok {
			destMissing <- src.MustTake()
			continue
		}

		srcBlob := src.MustPeek().Ref
		dstRef := dst.MustPeek().Ref
		if !dstRef.Valid() {
			return
		}

		switch {
		case srcBlob == dstRef:
			// Skip both
			sb := src.MustTake()
			db := dst.MustTake()
			if sb.Size != db.Size {
				sizeMismatch(sb.Ref)
			}
		case srcBlob.Less(dstRef):
			destMissing <- src.MustTake()
		case dstRef.Less(srcBlob):
			dst.Take()
		}
	}
}