Function withFetchProgress

daemon/pkg/plugin/fetch_linux.go:198–293

withFetchProgress is a fetch handler which registers a descriptor with a progress

(cs content.Store, out progress.Output, ref reference.Named)

// withFetchProgress is a fetch handler which registers a descriptor with a progress
func withFetchProgress(cs content.Store, out progress.Output, ref reference.Named) c8dimages.HandlerFunc {
	return func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
		switch desc.MediaType {
		case ocispec.MediaTypeImageManifest, c8dimages.MediaTypeDockerSchema2Manifest:
			tn := reference.TagNameOnly(ref)
			var tagOrDigest string
			if tagged, ok := tn.(reference.Tagged); ok {
				tagOrDigest = tagged.Tag()
			} else {
				tagOrDigest = tn.String()
			}
			progress.Messagef(out, tagOrDigest, "Pulling from %s", reference.FamiliarName(ref))
			progress.Messagef(out, "", "Digest: %s", desc.Digest.String())
			return nil, nil
		case
			c8dimages.MediaTypeDockerSchema2LayerGzip,
			c8dimages.MediaTypeDockerSchema2Layer,
			ocispec.MediaTypeImageLayer,
			ocispec.MediaTypeImageLayerGzip:
		default:
			return nil, nil
		}

		id := stringid.TruncateID(desc.Digest.String())

		if _, err := cs.Info(ctx, desc.Digest); err == nil {
			out.WriteProgress(progress.Progress{ID: id, Action: "Already exists", LastUpdate: true})
			return nil, nil
		}

		progress.Update(out, id, "Waiting")

		key := remotes.MakeRefKey(ctx, desc)

		go func() {
			timer := time.NewTimer(100 * time.Millisecond)
			if !timer.Stop() {
				<-timer.C
			}
			defer timer.Stop()

			var pulling bool
			var (
				// make sure we can still fetch from the content store
				// if the main context is cancelled
				// TODO: Might need to add some sort of timeout; see https://github.com/moby/moby/issues/49413
				ctxErr      error
				noCancelCTX = context.WithoutCancel(ctx)
			)

			for {
				timer.Reset(100 * time.Millisecond)

				select {
				case <-ctx.Done():
					ctxErr = ctx.Err()
				case <-timer.C:
				}