func buildContainerImagesInParallel()

in internal/pkg/cli/deploy/workload.go [486:556]
68 lines of code
12 McCabe index (conditional complexity)

func buildContainerImagesInParallel(in *ImageActionInput, uri string, buildArgsPerContainer map[string]*dockerengine.BuildArguments, buildFunc func(ctx context.Context, args *dockerengine.BuildArguments, w io.Writer) (string, error), out *UploadArtifactsOutput) error {
	var digestsMu sync.Mutex
	out.ImageDigests = make(map[string]ContainerImageIdentifier, len(buildArgsPerContainer))
	var labeledBuffers []*syncbuffer.LabeledSyncBuffer
	g, ctx := errgroup.WithContext(context.Background())
	cursor := cursor.New()
	cursor.Hide()
	for name, buildArgs := range buildArgsPerContainer {
		// create a copy of loop variables to avoid data race.
		name := name
		buildArgs := buildArgs

		buildArgs.URI = uri
		buildArgsList, err := buildArgs.GenerateDockerBuildArgs(dockerengine.New(exec.NewCmd()))
		if err != nil {
			return fmt.Errorf("generate docker build args for %q: %w", name, err)
		}
		buf := syncbuffer.New()
		labeledBuffers = append(labeledBuffers, buf.WithLabel(fmt.Sprintf("Building your container image %q: docker %s", name, strings.Join(buildArgsList, " "))))
		pr, pw := io.Pipe()
		g.Go(func() error {
			defer pw.Close()
			digest, err := buildFunc(ctx, buildArgs, pw)
			if err != nil {
				return fmt.Errorf("build and push the image %q: %w", name, err)
			}

			id := ContainerImageIdentifier{
				Digest:            digest,
				CustomTag:         in.CustomTag,
				GitShortCommitTag: in.GitShortCommitTag,
			}
			for _, tag := range buildArgs.Tags {
				id.RepoTags = append(id.RepoTags, fmt.Sprintf("%s:%s", uri, tag))
			}
			sort.Strings(id.RepoTags)
			digestsMu.Lock()
			defer digestsMu.Unlock()
			out.ImageDigests[name] = id
			return nil
		})
		g.Go(func() error {
			if err := buf.Copy(pr); err != nil {
				return fmt.Errorf("copy build and push output for %q: %w", name, err)
			}
			return nil
		})
	}
	opts := []syncbuffer.LabeledTermPrinterOption{syncbuffer.WithPadding(paddingInSpacesForBuildAndPush)}
	if os.Getenv("CI") != "true" {
		opts = append(opts, syncbuffer.WithNumLines(defaultNumLinesForBuildAndPush))
	}
	ltp := in.LabeledTermPrinter(os.Stderr, labeledBuffers, opts...)
	g.Go(func() error {
		for {
			ltp.Print()
			if ltp.IsDone() {
				return nil
			}
			select {
			case <-ctx.Done():
				return nil
			case <-time.After(pollIntervalForBuildAndPush):
			}
		}
	})
	if err := g.Wait(); err != nil {
		return err
	}
	return nil
}