Add streamOverrideWrapper to intercepts command.Cli stream methods and transparently returns custom streams when provided via options

Add new GetConfiguredStreams function to Compose API definition

Signed-off-by: Guillaume Lours <[email protected]>

cmd/compose/logs.go

@@ -86,11 +86,12 @@ func runLogs(ctx context.Context, dockerCli command.Cli, backendOptions *Backend
 		}
 	}
 
-	consumer := formatter.NewLogConsumer(ctx, dockerCli.Out(), dockerCli.Err(), !opts.noColor, !opts.noPrefix, false)
 	backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...)
 	if err != nil {
 		return err
 	}
+	outStream, errStream, _ := backend.GetConfiguredStreams()
+	consumer := formatter.NewLogConsumer(ctx, outStream, errStream, !opts.noColor, !opts.noPrefix, false)
 	return backend.Logs(ctx, name, consumer, api.LogOptions{
 		Project:    project,
 		Services:   services,

cmd/compose/up.go

@@ -293,7 +293,8 @@ func runUp(
 	var consumer api.LogConsumer
 	var attach []string
 	if !upOptions.Detach {
-		consumer = formatter.NewLogConsumer(ctx, dockerCli.Out(), dockerCli.Err(), !upOptions.noColor, !upOptions.noPrefix, upOptions.timestamp)
+		outStream, errStream, _ := backend.GetConfiguredStreams()
+		consumer = formatter.NewLogConsumer(ctx, outStream, errStream, !upOptions.noColor, !upOptions.noPrefix, upOptions.timestamp)
 
 		var attachSet utils.Set[string]
 		if len(upOptions.attach) != 0 {

cmd/compose/watch.go

@@ -121,11 +121,12 @@ func runWatch(ctx context.Context, dockerCli command.Cli, backendOptions *Backen
 		}
 	}
 
-	consumer := formatter.NewLogConsumer(ctx, dockerCli.Out(), dockerCli.Err(), false, false, false)
 	backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...)
 	if err != nil {
 		return err
 	}
+	outStream, errStream, _ := backend.GetConfiguredStreams()
+	consumer := formatter.NewLogConsumer(ctx, outStream, errStream, false, false, false)
 	return backend.Watch(ctx, project, api.WatchOptions{
 		Build:    &build,
 		LogTo:    consumer,

pkg/api/api.go

@@ -99,6 +99,9 @@ type Compose interface {
 	Generate(ctx context.Context, options GenerateOptions) (*types.Project, error)
 	// Volumes executes the equivalent to a `docker volume ls`
 	Volumes(ctx context.Context, project string, options VolumesOptions) ([]VolumesSummary, error)
+	// GetConfiguredStreams returns the configured I/O streams (stdout, stderr, stdin).
+	// If no custom streams were configured, it returns the dockerCli streams.
+	GetConfiguredStreams() (stdout io.Writer, stderr io.Writer, stdin io.Reader)
 }
 
 type VolumesOptions struct {

pkg/compose/build.go

@@ -177,7 +177,7 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti
 		if options.Progress == progress.ModeAuto {
 			options.Progress = os.Getenv("BUILDKIT_PROGRESS")
 		}
-		w, err = xprogress.NewPrinter(progressCtx, os.Stdout, progressui.DisplayMode(options.Progress),
+		w, err = xprogress.NewPrinter(progressCtx, s.stdout(), progressui.DisplayMode(options.Progress),
 			xprogress.WithDesc(
 				fmt.Sprintf("building with %q instance using %s driver", b.Name, b.Driver),
 				fmt.Sprintf("%s:%s", b.Driver, b.Name),

pkg/compose/build_bake.go

@@ -142,7 +142,7 @@ func (s *composeService) doBuildBake(ctx context.Context, project *types.Project
 		if displayMode == progress.ModeAuto && !s.stdout().IsTerminal() {
 			displayMode = progressui.PlainMode
 		}
-		out = os.Stdout // should be s.stdout(), but NewDisplay require access to the underlying *File
+		out = s.stdout()
 	}
 	display, err := progressui.NewDisplay(out, displayMode)
 	if err != nil {

pkg/compose/compose.go

@@ -94,6 +94,12 @@ func NewComposeService(dockerCli command.Cli, options ...Option) (api.Compose, e
 			return defaultValue, nil
 		}
 	}
+
+	// If custom streams were provided, wrap the Docker CLI to use them
+	if s.outStream != nil || s.errStream != nil || s.inStream != nil {
+		s.dockerCli = s.wrapDockerCliWithStreams(dockerCli)
+	}
+
 	return s, nil
 }
 
@@ -141,7 +147,7 @@ func WithContextInfo(info api.ContextInfo) Option {
 
 // WithProxyConfig sets custom HTTP proxy configuration for builds
 func WithProxyConfig(config map[string]string) Option {
-	return func(s *composeService) error{
+	return func(s *composeService) error {
 		s.proxyConfig = config
 		return nil
 	}
@@ -238,28 +244,15 @@ func (s *composeService) getProxyConfig() map[string]string {
 	return storeutil.GetProxyConfig(s.dockerCli)
 }
 
-
 func (s *composeService) stdout() *streams.Out {
-	// If stream overrides are provided, use them
-	if s.outStream != nil {
-		return streams.NewOut(s.outStream)
-	}
 	return s.dockerCli.Out()
 }
 
 func (s *composeService) stdin() *streams.In {
-	// If stream overrides are provided, use them
-	if s.inStream != nil {
-		return streams.NewIn(&readCloserAdapter{r: s.inStream})
-	}
 	return s.dockerCli.In()
 }
 
 func (s *composeService) stderr() *streams.Out {
-	// If stream overrides are provided, use them
-	if s.errStream != nil {
-		return streams.NewOut(s.errStream)
-	}
 	return s.dockerCli.Err()
 }
 
@@ -270,6 +263,11 @@ func (s *composeService) stdinfo() *streams.Out {
 	return s.stderr()
 }
 
+// GetConfiguredStreams returns the configured I/O streams (implements api.Compose interface)
+func (s *composeService) GetConfiguredStreams() (io.Writer, io.Writer, io.Reader) {
+	return s.stdout(), s.stderr(), s.stdin()
+}
+
 // readCloserAdapter adapts io.Reader to io.ReadCloser
 type readCloserAdapter struct {
 	r io.Reader
@@ -283,6 +281,55 @@ func (r *readCloserAdapter) Close() error {
 	return nil
 }
 
+// wrapDockerCliWithStreams wraps the Docker CLI to intercept and override stream methods
+func (s *composeService) wrapDockerCliWithStreams(baseCli command.Cli) command.Cli {
+	wrapper := &streamOverrideWrapper{
+		Cli: baseCli,
+	}
+
+	// Wrap custom streams in Docker CLI's stream types
+	if s.outStream != nil {
+		wrapper.outStream = streams.NewOut(s.outStream)
+	}
+	if s.errStream != nil {
+		wrapper.errStream = streams.NewOut(s.errStream)
+	}
+	if s.inStream != nil {
+		wrapper.inStream = streams.NewIn(&readCloserAdapter{r: s.inStream})
+	}
+
+	return wrapper
+}
+
+// streamOverrideWrapper wraps command.Cli to override streams with custom implementations
+type streamOverrideWrapper struct {
+	command.Cli
+	outStream *streams.Out
+	errStream *streams.Out
+	inStream  *streams.In
+}
+
+func (w *streamOverrideWrapper) Out() *streams.Out {
+	if w.outStream != nil {
+		return w.outStream
+	}
+	return w.Cli.Out()
+}
+
+func (w *streamOverrideWrapper) Err() *streams.Out {
+	if w.errStream != nil {
+		return w.errStream
+	}
+	return w.Cli.Err()
+}
+
+func (w *streamOverrideWrapper) In() *streams.In {
+	if w.inStream != nil {
+		return w.inStream
+	}
+	return w.Cli.In()
+}
+
 func getCanonicalContainerName(c container.Summary) string {
 	if len(c.Names) == 0 {
 		// corner case, sometime happens on removal. return short ID as a safeguard value