DNM: testing

cli/command/image/build.go

@@ -362,7 +362,7 @@ func runBuild(ctx context.Context, dockerCli command.Cli, options buildOptions)
 		}
 	}
 
-	err = jsonstream.Display(ctx, response.Body, streams.NewOut(buildBuff), jsonstream.WithAuxCallback(aux))
+	err = jsonstream.DisplayStream(ctx, response.Body, streams.NewOut(buildBuff), jsonstream.WithAuxCallback(aux))
 	if err != nil {
 		var jerr *jsonstream.JSONError
 		if errors.As(err, &jerr) {

cli/command/image/import.go

@@ -104,5 +104,5 @@ func runImport(ctx context.Context, dockerCli command.Cli, options importOptions
 	}
 	defer responseBody.Close()
 
-	return jsonstream.Display(ctx, responseBody, dockerCli.Out())
+	return jsonstream.DisplayStream(ctx, responseBody, dockerCli.Out())
 }

cli/command/image/load.go

@@ -97,5 +97,5 @@ func runLoad(ctx context.Context, dockerCli command.Cli, opts loadOptions) error
 	}
 	defer func() { _ = res.Close() }()
 
-	return jsonstream.Display(ctx, res, dockerCli.Out())
+	return jsonstream.DisplayStream(ctx, res, dockerCli.Out())
 }

cli/command/plugin/install.go

@@ -100,7 +100,7 @@ func runInstall(ctx context.Context, dockerCLI command.Cli, opts pluginOptions)
 	defer func() {
 		_ = responseBody.Close()
 	}()
-	if err := jsonstream.Display(ctx, responseBody, dockerCLI.Out()); err != nil {
+	if err := jsonstream.DisplayStream(ctx, responseBody, dockerCLI.Out()); err != nil {
 		return err
 	}
 	_, _ = fmt.Fprintln(dockerCLI.Out(), "Installed plugin", opts.remote) // todo: return proper values from the API for this result

cli/command/plugin/push.go

@@ -56,5 +56,5 @@ func runPush(ctx context.Context, dockerCli command.Cli, name string) error {
 	defer func() {
 		_ = responseBody.Close()
 	}()
-	return jsonstream.Display(ctx, responseBody, dockerCli.Out())
+	return jsonstream.DisplayStream(ctx, responseBody, dockerCli.Out())
 }

cli/command/plugin/upgrade.go

@@ -90,7 +90,7 @@ func runUpgrade(ctx context.Context, dockerCLI command.Cli, opts pluginOptions)
 	defer func() {
 		_ = responseBody.Close()
 	}()
-	if err := jsonstream.Display(ctx, responseBody, dockerCLI.Out()); err != nil {
+	if err := jsonstream.DisplayStream(ctx, responseBody, dockerCLI.Out()); err != nil {
 		return err
 	}
 	_, _ = fmt.Fprintf(dockerCLI.Out(), "Upgraded plugin %s to %s\n", opts.localName, opts.remote) // todo: return proper values from the API for this result

cli/command/service/helpers.go

@@ -24,7 +24,7 @@ func WaitOnService(ctx context.Context, dockerCli command.Cli, serviceID string,
 		return <-errChan
 	}
 
-	err := jsonstream.Display(ctx, pipeReader, dockerCli.Out())
+	err := jsonstream.DisplayStream(ctx, pipeReader, dockerCli.Out())
 	if err == nil {
 		err = <-errChan
 	}

cli/command/swarm/ca.go

@@ -124,7 +124,7 @@ func attach(ctx context.Context, dockerCLI command.Cli, opts caOptions) error {
 		return <-errChan
 	}
 
-	err := jsonstream.Display(ctx, pipeReader, dockerCLI.Out())
+	err := jsonstream.DisplayStream(ctx, pipeReader, dockerCLI.Out())
 	if err == nil {
 		err = <-errChan
 	}

internal/jsonstream/display.go

@@ -3,6 +3,7 @@ package jsonstream
 import (
 	"context"
 	"io"
+	"iter"
 
 	"github.com/docker/cli/cli/streams"
 	"github.com/moby/moby/api/types/jsonstream"
@@ -41,13 +42,41 @@ func WithAuxCallback(cb func(JSONMessage)) Options {
 	}
 }
 
+type ProgressResponse interface {
+	io.ReadCloser
+	JSONMessages(ctx context.Context) iter.Seq2[jsonstream.Message, error]
+}
+
 // Display prints the JSON messages from the given reader to the given stream.
 //
+// It wraps the [jsonmessage.DisplayJSONMessages] function to make it
+// "context aware" and appropriately returns why the function was canceled.
+//
+// It returns an error if the context is canceled, but not if the input reader / stream is closed.
+func Display(ctx context.Context, in ProgressResponse, stream *streams.Out, opts ...Options) error {
+	if ctx.Err() != nil {
+		return ctx.Err()
+	}
+
+	// reader := &ctxReader{err: make(chan error, 1), r: in}
+	// stopFunc := context.AfterFunc(ctx, func() { reader.err <- ctx.Err() })
+	// defer stopFunc()
+	//
+	o := options{}
+	for _, opt := range opts {
+		opt(&o)
+	}
+
+	return jsonmessage.DisplayJSONMessages(in.JSONMessages(ctx), stream, stream.FD(), stream.IsTerminal(), o.AuxCallback)
+}
+
+// DisplayStream prints the JSON messages from the given reader to the given stream.
+//
 // It wraps the [jsonmessage.DisplayJSONMessagesStream] function to make it
 // "context aware" and appropriately returns why the function was canceled.
 //
 // It returns an error if the context is canceled, but not if the input reader / stream is closed.
-func Display(ctx context.Context, in io.Reader, stream *streams.Out, opts ...Options) error {
+func DisplayStream(ctx context.Context, in io.Reader, stream *streams.Out, opts ...Options) error {
 	if ctx.Err() != nil {
 		return ctx.Err()
 	}

internal/jsonstream/display_test.go

@@ -36,7 +36,7 @@ func TestDisplay(t *testing.T) {
 
 	done := make(chan error)
 	go func() {
-		done <- Display(streamCtx, client, streams.NewOut(io.Discard))
+		done <- DisplayStream(streamCtx, client, streams.NewOut(io.Discard))
 	}()
 
 	cancelStream()