@@ -21,7 +21,7 @@ import (

 	"github.com/docker/docker/api/types/events"

 	mounttypes "github.com/docker/docker/api/types/mount"

 	swarmtypes "github.com/docker/docker/api/types/swarm"

-	"github.com/docker/docker/container/stream"

+	"github.com/docker/docker/daemon/internal/stream"

 	"github.com/docker/docker/daemon/logger"

 	"github.com/docker/docker/daemon/logger/jsonfilelog"

 	"github.com/docker/docker/daemon/logger/local"

@@ -7,7 +7,7 @@ import (

 	"github.com/containerd/containerd/v2/pkg/cio"

 	"github.com/containerd/log"

-	"github.com/docker/docker/container/stream"

+	"github.com/docker/docker/daemon/internal/stream"

 	"github.com/docker/docker/libcontainerd/types"

 	"github.com/docker/docker/pkg/stringid"

 )

deleted file mode 100644

@@ -1,183 +0,0 @@

-package stream

-

-import (

-	"context"

-	"io"

-

-	"github.com/containerd/log"

-	"github.com/docker/docker/pkg/pools"

-	"github.com/moby/term"

-	"github.com/pkg/errors"

-	"golang.org/x/sync/errgroup"

-)

-

-var defaultEscapeSequence = []byte{16, 17} // ctrl-p, ctrl-q

-

-// AttachConfig is the config struct used to attach a client to a stream's stdio

-type AttachConfig struct {

-	// Tells the attach copier that the stream's stdin is a TTY and to look for

-	// escape sequences in stdin to detach from the stream.

-	// When true the escape sequence is not passed to the underlying stream

-	TTY bool

-	// Specifies the detach keys the client will be using

-	// Only useful when `TTY` is true

-	DetachKeys []byte

-

-	// CloseStdin signals that once done, stdin for the attached stream should be closed

-	// For example, this would close the attached container's stdin.

-	CloseStdin bool

-

-	// UseStd* indicate whether the client has requested to be connected to the

-	// given stream or not.  These flags are used instead of checking Std* != nil

-	// at points before the client streams Std* are wired up.

-	UseStdin, UseStdout, UseStderr bool

-

-	// CStd* are the streams directly connected to the container

-	CStdin           io.WriteCloser

-	CStdout, CStderr io.ReadCloser

-

-	// Provide client streams to wire up to

-	Stdin          io.ReadCloser

-	Stdout, Stderr io.Writer

-}

-

-// AttachStreams attaches the container's streams to the AttachConfig

-func (c *Config) AttachStreams(cfg *AttachConfig) {

-	if cfg.UseStdin {

-		cfg.CStdin = c.StdinPipe()

-	}

-

-	if cfg.UseStdout {

-		cfg.CStdout = c.StdoutPipe()

-	}

-

-	if cfg.UseStderr {

-		cfg.CStderr = c.StderrPipe()

-	}

-}

-

-// CopyStreams starts goroutines to copy data in and out to/from the container

-func (c *Config) CopyStreams(ctx context.Context, cfg *AttachConfig) <-chan error {

-	var group errgroup.Group

-

-	// Connect stdin of container to the attach stdin stream.

-	if cfg.Stdin != nil {

-		group.Go(func() error {

-			log.G(ctx).Debug("attach: stdin: begin")

-			defer log.G(ctx).Debug("attach: stdin: end")

-

-			defer func() {

-				if cfg.CloseStdin && !cfg.TTY {

-					cfg.CStdin.Close()

-				} else {

-					// No matter what, when stdin is closed (io.Copy unblock), close stdout and stderr

-					if cfg.CStdout != nil {

-						cfg.CStdout.Close()

-					}

-					if cfg.CStderr != nil {

-						cfg.CStderr.Close()

-					}

-				}

-			}()

-

-			var err error

-			if cfg.TTY {

-				_, err = copyEscapable(cfg.CStdin, cfg.Stdin, cfg.DetachKeys)

-			} else {

-				_, err = pools.Copy(cfg.CStdin, cfg.Stdin)

-			}

-			if errors.Is(err, io.ErrClosedPipe) {

-				err = nil

-			}

-			if err != nil {

-				log.G(ctx).WithError(err).Debug("error on attach stdin")

-				return errors.Wrap(err, "error on attach stdin")

-			}

-			return nil

-		})

-	}

-

-	attachStream := func(name string, stream io.Writer, streamPipe io.ReadCloser) error {

-		log.G(ctx).Debugf("attach: %s: begin", name)

-		defer log.G(ctx).Debugf("attach: %s: end", name)

-		defer func() {

-			// Make sure stdin gets closed

-			if cfg.Stdin != nil {

-				cfg.Stdin.Close()

-			}

-			streamPipe.Close()

-		}()

-

-		_, err := pools.Copy(stream, streamPipe)

-		if errors.Is(err, io.ErrClosedPipe) {

-			err = nil

-		}

-		if err != nil {

-			log.G(ctx).WithError(err).Debugf("attach: %s", name)

-			return errors.Wrapf(err, "error attaching %s stream", name)

-		}

-		return nil

-	}

-

-	if cfg.Stdout != nil {

-		group.Go(func() error {

-			return attachStream("stdout", cfg.Stdout, cfg.CStdout)

-		})

-	}

-	if cfg.Stderr != nil {

-		group.Go(func() error {

-			return attachStream("stderr", cfg.Stderr, cfg.CStderr)

-		})

-	}

-

-	errs := make(chan error, 1)

-	go func() {

-		defer log.G(ctx).Debug("attach done")

-		groupErr := make(chan error, 1)

-		go func() {

-			groupErr <- group.Wait()

-		}()

-		select {

-		case <-ctx.Done():

-			// close all pipes

-			if cfg.CStdin != nil {

-				cfg.CStdin.Close()

-			}

-			if cfg.CStdout != nil {

-				cfg.CStdout.Close()

-			}

-			if cfg.CStderr != nil {

-				cfg.CStderr.Close()

-			}

-

-			if cfg.Stdin != nil {

-				// In this case, `cfg.Stdin` is a stream from the client.

-				// The way `io.Copy` works we may get stuck waiting to read from `cfg.Stdin` even if the container has exited.

-				// This will cause the `io.Copy` to never return and the `group.Wait()` to never return.

-				// By closing cfg.Stdin we will cause the `io.Copy` to return and the `group.Wait()` to return.

-				cfg.Stdin.Close()

-			}

-

-			// Now with these closed, wait should return.

-			if err := group.Wait(); err != nil {

-				errs <- err

-				return

-			}

-			errs <- ctx.Err()

-		case err := <-groupErr:

-			errs <- err

-		}

-	}()

-

-	return errs

-}

-

-func copyEscapable(dst io.Writer, src io.ReadCloser, keys []byte) (written int64, _ error) {

-	if len(keys) == 0 {

-		keys = defaultEscapeSequence

-	}

-	pr := term.NewEscapeProxy(src, keys)

-	defer src.Close()

-

-	return pools.Copy(dst, pr)

-}

deleted file mode 100644

@@ -1,102 +0,0 @@

-package stream

-

-import (

-	"context"

-	"io"

-	"testing"

-	"time"

-

-	"gotest.tools/v3/assert"

-)

-

-// Make sure when there is no I/O on a stream that the goroutines do not get blocked after the container exits.

-func TestAttachNoIO(t *testing.T) {

-	t.Run("stdin only", func(t *testing.T) {

-		stdinR, _ := io.Pipe()

-		defer stdinR.Close()

-		testStreamCopy(t, stdinR, nil, nil)

-	})

-

-	t.Run("stdout only", func(t *testing.T) {

-		_, w := io.Pipe()

-		defer w.Close()

-		testStreamCopy(t, nil, w, nil)

-	})

-

-	t.Run("stderr only", func(t *testing.T) {

-		_, w := io.Pipe()

-		defer w.Close()

-		testStreamCopy(t, nil, nil, w)

-	})

-

-	t.Run("stdout+stderr", func(t *testing.T) {

-		_, stdoutW := io.Pipe()

-		defer stdoutW.Close()

-		_, stderrW := io.Pipe()

-		defer stderrW.Close()

-

-		testStreamCopy(t, nil, stdoutW, stderrW)

-	})

-

-	t.Run("stdin+stdout", func(t *testing.T) {

-		stdin, _ := io.Pipe()

-		defer stdin.Close()

-		_, stdout := io.Pipe()

-		defer stdout.Close()

-

-		testStreamCopy(t, stdin, stdout, nil)

-	})

-

-	t.Run("stdin+stderr", func(t *testing.T) {

-		stdin, _ := io.Pipe()

-		defer stdin.Close()

-		_, stderr := io.Pipe()

-		defer stderr.Close()

-

-		testStreamCopy(t, stdin, nil, stderr)

-	})

-

-	t.Run("stdin+stdout+stderr", func(t *testing.T) {

-		stdinR, _ := io.Pipe()

-		defer stdinR.Close()

-		stdoutR, stdoutW := io.Pipe()

-		defer stdoutR.Close()

-		stderrR, stderrW := io.Pipe()

-		defer stderrR.Close()

-		testStreamCopy(t, stdinR, stdoutW, stderrW)

-	})

-}

-

-func testStreamCopy(t *testing.T, stdin io.ReadCloser, stdout, stderr io.WriteCloser) {

-	cfg := AttachConfig{

-		UseStdin:  stdin != nil,

-		UseStdout: stdout != nil,

-		UseStderr: stderr != nil,

-		Stdin:     stdin,

-		Stdout:    stdout,

-		Stderr:    stderr,

-	}

-

-	sc := NewConfig()

-	sc.AttachStreams(&cfg)

-	defer sc.CloseStreams()

-

-	ctx, cancel := context.WithCancel(context.Background())

-	defer cancel()

-	chErr := sc.CopyStreams(ctx, &cfg)

-

-	select {

-	case err := <-chErr:

-		assert.NilError(t, err)

-	default:

-	}

-

-	cancel()

-

-	select {

-	case err := <-chErr:

-		assert.ErrorIs(t, err, context.Canceled)

-	case <-time.After(10 * time.Second):

-		t.Fatal("timeout waiting for CopyStreams to exit")

-	}

-}

deleted file mode 100644

@@ -1,187 +0,0 @@

-package stream

-

-import (

-	"context"

-	"errors"

-	"fmt"

-	"io"

-	"sync"

-	"sync/atomic"

-

-	"github.com/containerd/containerd/v2/pkg/cio"

-	"github.com/containerd/log"

-	"github.com/docker/docker/daemon/internal/stream/bytespipe"

-	"github.com/docker/docker/pkg/pools"

-)

-

-// Config holds information about I/O streams managed together.

-//

-// config.StdinPipe returns a WriteCloser which can be used to feed data

-// to the standard input of the streamConfig's active process.

-// config.StdoutPipe and streamConfig.StderrPipe each return a ReadCloser

-// which can be used to retrieve the standard output (and error) generated

-// by the container's active process. The output (and error) are actually

-// copied and delivered to all StdoutPipe and StderrPipe consumers, using

-// a kind of "broadcaster".

-type Config struct {

-	wg        sync.WaitGroup

-	stdout    *unbuffered

-	stderr    *unbuffered

-	stdin     io.ReadCloser

-	stdinPipe io.WriteCloser

-	dio       *cio.DirectIO

-	// closed is set to true when CloseStreams is called

-	closed atomic.Bool

-}

-

-// NewConfig creates a stream config and initializes

-// the standard err and standard out to new unbuffered broadcasters.

-func NewConfig() *Config {

-	return &Config{

-		stderr: new(unbuffered),

-		stdout: new(unbuffered),

-	}

-}

-

-// Stdout returns the standard output in the configuration.

-func (c *Config) Stdout() io.Writer {

-	return c.stdout

-}

-

-// Stderr returns the standard error in the configuration.

-func (c *Config) Stderr() io.Writer {

-	return c.stderr

-}

-

-// Stdin returns the standard input in the configuration.

-func (c *Config) Stdin() io.ReadCloser {

-	return c.stdin

-}

-

-// StdinPipe returns an input writer pipe as an io.WriteCloser.

-func (c *Config) StdinPipe() io.WriteCloser {

-	return c.stdinPipe

-}

-

-// StdoutPipe creates a new io.ReadCloser with an empty bytes pipe.

-// It adds this new out pipe to the Stdout broadcaster.

-// This will block stdout if unconsumed.

-func (c *Config) StdoutPipe() io.ReadCloser {

-	bytesPipe := bytespipe.New()

-	c.stdout.Add(bytesPipe)

-	return bytesPipe

-}

-

-// StderrPipe creates a new io.ReadCloser with an empty bytes pipe.

-// It adds this new err pipe to the Stderr broadcaster.

-// This will block stderr if unconsumed.

-func (c *Config) StderrPipe() io.ReadCloser {

-	bytesPipe := bytespipe.New()

-	c.stderr.Add(bytesPipe)

-	return bytesPipe

-}

-

-// NewInputPipes creates new pipes for both standard inputs, Stdin and StdinPipe.

-func (c *Config) NewInputPipes() {

-	c.stdin, c.stdinPipe = io.Pipe()

-}

-

-// NewNopInputPipe creates a new input pipe that will silently drop all messages in the input.

-func (c *Config) NewNopInputPipe() {

-	c.stdinPipe = &nopWriteCloser{io.Discard}

-}

-

-type nopWriteCloser struct {

-	io.Writer

-}

-

-func (w *nopWriteCloser) Close() error { return nil }

-

-// CloseStreams ensures that the configured streams are properly closed.

-func (c *Config) CloseStreams() error {

-	var errs error

-

-	c.closed.Store(true)

-

-	if c.stdin != nil {

-		if err := c.stdin.Close(); err != nil {

-			errs = errors.Join(errs, fmt.Errorf("error close stdin: %w", err))

-		}

-	}

-

-	if err := c.stdout.Clean(); err != nil {

-		errs = errors.Join(errs, fmt.Errorf("error close stdout: %w", err))

-	}

-

-	if err := c.stderr.Clean(); err != nil {

-		errs = errors.Join(errs, fmt.Errorf("error close stderr: %w", err))

-	}

-

-	return errs

-}

-

-// CopyToPipe connects streamconfig with a libcontainerd.IOPipe

-func (c *Config) CopyToPipe(iop *cio.DirectIO) {

-	ctx := context.TODO()

-

-	c.dio = iop

-	copyFunc := func(name string, w io.Writer, r io.ReadCloser) {

-		c.wg.Add(1)

-		go func() {

-			defer c.wg.Done()

-			if _, err := pools.Copy(w, r); err != nil {

-				if c.closed.Load() {

-					return

-				}

-				log.G(ctx).WithFields(log.Fields{"stream": name, "error": err}).Error("copy stream failed")

-			}

-			if err := r.Close(); err != nil && !c.closed.Load() {

-				log.G(ctx).WithFields(log.Fields{"stream": name, "error": err}).Warn("close stream failed")

-			}

-		}()

-	}

-

-	if iop.Stdout != nil {

-		copyFunc("stdout", c.Stdout(), iop.Stdout)

-	}

-	if iop.Stderr != nil {

-		copyFunc("stderr", c.Stderr(), iop.Stderr)

-	}

-

-	if stdin := c.Stdin(); stdin != nil {

-		if iop.Stdin != nil {

-			go func() {

-				_, err := pools.Copy(iop.Stdin, stdin)

-				if err != nil {

-					if c.closed.Load() {

-						return

-					}

-					log.G(ctx).WithFields(log.Fields{"stream": "stdin", "error": err}).Error("copy stream failed")

-				}

-				if err := iop.Stdin.Close(); err != nil && !c.closed.Load() {

-					log.G(ctx).WithFields(log.Fields{"stream": "stdin", "error": err}).Warn("close stream failed")

-				}

-			}()

-		}

-	}

-}

-

-// Wait for the stream to close

-// Wait supports timeouts via the context to unblock and forcefully

-// close the io streams

-func (c *Config) Wait(ctx context.Context) {

-	done := make(chan struct{}, 1)

-	go func() {

-		c.wg.Wait()

-		close(done)

-	}()

-	select {

-	case <-done:

-	case <-ctx.Done():

-		if c.dio != nil {

-			c.dio.Cancel()

-			c.dio.Wait()

-			c.dio.Close()

-		}

-	}

-}

deleted file mode 100644

@@ -1,49 +0,0 @@

-package stream

-

-import (

-	"io"

-	"sync"

-)

-

-// unbuffered accumulates multiple io.WriteCloser by stream.

-type unbuffered struct {

-	mu      sync.Mutex

-	writers []io.WriteCloser

-}

-

-// Add adds new io.WriteCloser.

-func (w *unbuffered) Add(writer io.WriteCloser) {

-	w.mu.Lock()

-	w.writers = append(w.writers, writer)

-	w.mu.Unlock()

-}

-

-// Write writes bytes to all writers. Failed writers will be evicted during

-// this call.

-func (w *unbuffered) Write(p []byte) (int, error) {

-	w.mu.Lock()

-	var evict []int

-	for i, sw := range w.writers {

-		if n, err := sw.Write(p); err != nil || n != len(p) {

-			// On error, evict the writer

-			evict = append(evict, i)

-		}

-	}

-	for n, i := range evict {

-		w.writers = append(w.writers[:i-n], w.writers[i-n+1:]...)

-	}

-	w.mu.Unlock()

-	return len(p), nil

-}

-

-// Clean closes and removes all writers. Last non-eol-terminated part of data

-// will be saved.

-func (w *unbuffered) Clean() error {

-	w.mu.Lock()

-	for _, sw := range w.writers {

-		sw.Close()

-	}

-	w.writers = nil

-	w.mu.Unlock()

-	return nil

-}

deleted file mode 100644

@@ -1,164 +0,0 @@

-package stream

-

-import (

-	"bytes"

-	"errors"

-	"strings"

-	"testing"

-)

-

-type dummyWriter struct {

-	buffer      bytes.Buffer

-	failOnWrite bool

-}

-

-func (dw *dummyWriter) Write(p []byte) (int, error) {

-	if dw.failOnWrite {

-		return 0, errors.New("Fake fail")

-	}

-	return dw.buffer.Write(p)

-}

-

-func (dw *dummyWriter) String() string {

-	return dw.buffer.String()

-}

-

-func (dw *dummyWriter) Close() error {

-	return nil

-}

-

-func TestUnbuffered(t *testing.T) {

-	writer := new(unbuffered)

-

-	// Test 1: Both bufferA and bufferB should contain "foo"

-	bufferA := &dummyWriter{}

-	writer.Add(bufferA)

-	bufferB := &dummyWriter{}

-	writer.Add(bufferB)

-	writer.Write([]byte("foo"))

-

-	if bufferA.String() != "foo" {

-		t.Errorf("Buffer contains %v", bufferA.String())

-	}

-

-	if bufferB.String() != "foo" {

-		t.Errorf("Buffer contains %v", bufferB.String())

-	}

-

-	// Test2: bufferA and bufferB should contain "foobar",

-	// while bufferC should only contain "bar"

-	bufferC := &dummyWriter{}

-	writer.Add(bufferC)

-	writer.Write([]byte("bar"))

-

-	if bufferA.String() != "foobar" {

-		t.Errorf("Buffer contains %v", bufferA.String())

-	}

-

-	if bufferB.String() != "foobar" {

-		t.Errorf("Buffer contains %v", bufferB.String())

-	}

-

-	if bufferC.String() != "bar" {

-		t.Errorf("Buffer contains %v", bufferC.String())

-	}

-

-	// Test3: Test eviction on failure

-	bufferA.failOnWrite = true

-	writer.Write([]byte("fail"))

-	if bufferA.String() != "foobar" {

-		t.Errorf("Buffer contains %v", bufferA.String())

-	}

-	if bufferC.String() != "barfail" {

-		t.Errorf("Buffer contains %v", bufferC.String())

-	}

-	// Even though we reset the flag, no more writes should go in there

-	bufferA.failOnWrite = false

-	writer.Write([]byte("test"))

-	if bufferA.String() != "foobar" {

-		t.Errorf("Buffer contains %v", bufferA.String())

-	}

-	if bufferC.String() != "barfailtest" {

-		t.Errorf("Buffer contains %v", bufferC.String())

-	}

-

-	// Test4: Test eviction on multiple simultaneous failures

-	bufferB.failOnWrite = true

-	bufferC.failOnWrite = true

-	bufferD := &dummyWriter{}

-	writer.Add(bufferD)

-	writer.Write([]byte("yo"))

-	writer.Write([]byte("ink"))

-	if strings.Contains(bufferB.String(), "yoink") {

-		t.Errorf("bufferB received write. contents: %q", bufferB)

-	}

-	if strings.Contains(bufferC.String(), "yoink") {

-		t.Errorf("bufferC received write. contents: %q", bufferC)

-	}

-	if g, w := bufferD.String(), "yoink"; g != w {

-		t.Errorf("bufferD = %q, want %q", g, w)

-	}

-

-	writer.Clean()

-}

-

-type devNullCloser int

-

-func (d devNullCloser) Close() error {

-	return nil

-}

-

-func (d devNullCloser) Write(buf []byte) (int, error) {

-	return len(buf), nil

-}

-

-// This test checks for races. It is only useful when run with the race detector.

-func TestRaceUnbuffered(t *testing.T) {

-	writer := new(unbuffered)

-	c := make(chan bool)

-	go func() {

-		writer.Add(devNullCloser(0))

-		c <- true

-	}()

-	_, err := writer.Write([]byte("hello"))

-	if err != nil {

-		t.Error(err)

-	}

-	<-c

-}

-

-func BenchmarkUnbuffered(b *testing.B) {

-	writer := new(unbuffered)

-	setUpWriter := func() {

-		for i := 0; i < 100; i++ {

-			writer.Add(devNullCloser(0))

-			writer.Add(devNullCloser(0))

-			writer.Add(devNullCloser(0))

-		}

-	}

-	testLine := "Line that thinks that it is log line from docker"

-	var buf bytes.Buffer

-	for i := 0; i < 100; i++ {

-		buf.WriteString(testLine + "\n")

-	}

-	// line without eol

-	buf.WriteString(testLine)

-	testText := buf.Bytes()

-	b.SetBytes(int64(5 * len(testText)))

-	b.ResetTimer()

-	for i := 0; i < b.N; i++ {

-		b.StopTimer()

-		setUpWriter()

-		b.StartTimer()

-

-		for j := 0; j < 5; j++ {

-			if _, err := writer.Write(testText); err != nil {

-				b.Fatal(err)

-			}

-		}

-

-		b.StopTimer()

-		writer.Clean()

-		b.StartTimer()

-	}

-}

@@ -10,7 +10,7 @@ import (

 	containertypes "github.com/docker/docker/api/types/container"

 	"github.com/docker/docker/api/types/events"

 	"github.com/docker/docker/container"

-	"github.com/docker/docker/container/stream"

+	"github.com/docker/docker/daemon/internal/stream"

 	"github.com/docker/docker/daemon/logger"

 	"github.com/docker/docker/errdefs"

 	"github.com/docker/docker/pkg/stdcopy"

@@ -15,7 +15,7 @@ import (

 	containertypes "github.com/docker/docker/api/types/container"

 	"github.com/docker/docker/api/types/events"

 	"github.com/docker/docker/container"

-	"github.com/docker/docker/container/stream"

+	"github.com/docker/docker/daemon/internal/stream"

 	"github.com/docker/docker/errdefs"

 	"github.com/docker/docker/pkg/pools"

 	"github.com/moby/sys/signal"

new file mode 100644

@@ -0,0 +1,183 @@

+package stream

+

+import (

+	"context"

+	"io"

+

+	"github.com/containerd/log"

+	"github.com/docker/docker/pkg/pools"

+	"github.com/moby/term"

+	"github.com/pkg/errors"

+	"golang.org/x/sync/errgroup"

+)

+

+var defaultEscapeSequence = []byte{16, 17} // ctrl-p, ctrl-q

+

+// AttachConfig is the config struct used to attach a client to a stream's stdio

+type AttachConfig struct {

+	// Tells the attach copier that the stream's stdin is a TTY and to look for

+	// escape sequences in stdin to detach from the stream.

+	// When true the escape sequence is not passed to the underlying stream

+	TTY bool

+	// Specifies the detach keys the client will be using

+	// Only useful when `TTY` is true

+	DetachKeys []byte

+

+	// CloseStdin signals that once done, stdin for the attached stream should be closed

+	// For example, this would close the attached container's stdin.

+	CloseStdin bool

+

+	// UseStd* indicate whether the client has requested to be connected to the

+	// given stream or not.  These flags are used instead of checking Std* != nil

+	// at points before the client streams Std* are wired up.

+	UseStdin, UseStdout, UseStderr bool

+

+	// CStd* are the streams directly connected to the container

+	CStdin           io.WriteCloser

+	CStdout, CStderr io.ReadCloser

+

+	// Provide client streams to wire up to

+	Stdin          io.ReadCloser

+	Stdout, Stderr io.Writer

+}

+

+// AttachStreams attaches the container's streams to the AttachConfig

+func (c *Config) AttachStreams(cfg *AttachConfig) {

+	if cfg.UseStdin {

+		cfg.CStdin = c.StdinPipe()

+	}

+

+	if cfg.UseStdout {

+		cfg.CStdout = c.StdoutPipe()

+	}

+

+	if cfg.UseStderr {

+		cfg.CStderr = c.StderrPipe()

+	}

+}

+

+// CopyStreams starts goroutines to copy data in and out to/from the container

+func (c *Config) CopyStreams(ctx context.Context, cfg *AttachConfig) <-chan error {

+	var group errgroup.Group

+

+	// Connect stdin of container to the attach stdin stream.

+	if cfg.Stdin != nil {

+		group.Go(func() error {

+			log.G(ctx).Debug("attach: stdin: begin")

+			defer log.G(ctx).Debug("attach: stdin: end")

+

+			defer func() {

+				if cfg.CloseStdin && !cfg.TTY {

+					cfg.CStdin.Close()

+				} else {

+					// No matter what, when stdin is closed (io.Copy unblock), close stdout and stderr

+					if cfg.CStdout != nil {

+						cfg.CStdout.Close()

+					}

+					if cfg.CStderr != nil {

+						cfg.CStderr.Close()

+					}

+				}

+			}()

+

+			var err error

+			if cfg.TTY {

+				_, err = copyEscapable(cfg.CStdin, cfg.Stdin, cfg.DetachKeys)

+			} else {

+				_, err = pools.Copy(cfg.CStdin, cfg.Stdin)

+			}

+			if errors.Is(err, io.ErrClosedPipe) {

+				err = nil

+			}

+			if err != nil {

+				log.G(ctx).WithError(err).Debug("error on attach stdin")

+				return errors.Wrap(err, "error on attach stdin")

+			}

+			return nil

+		})

+	}

+

+	attachStream := func(name string, stream io.Writer, streamPipe io.ReadCloser) error {

+		log.G(ctx).Debugf("attach: %s: begin", name)

+		defer log.G(ctx).Debugf("attach: %s: end", name)

+		defer func() {

+			// Make sure stdin gets closed

+			if cfg.Stdin != nil {

+				cfg.Stdin.Close()

+			}

+			streamPipe.Close()

+		}()

+

+		_, err := pools.Copy(stream, streamPipe)

+		if errors.Is(err, io.ErrClosedPipe) {

+			err = nil

+		}

+		if err != nil {

+			log.G(ctx).WithError(err).Debugf("attach: %s", name)

+			return errors.Wrapf(err, "error attaching %s stream", name)

+		}

+		return nil

+	}

+

+	if cfg.Stdout != nil {

+		group.Go(func() error {

+			return attachStream("stdout", cfg.Stdout, cfg.CStdout)

+		})

+	}

+	if cfg.Stderr != nil {

+		group.Go(func() error {

+			return attachStream("stderr", cfg.Stderr, cfg.CStderr)

+		})

+	}

+

+	errs := make(chan error, 1)

+	go func() {

+		defer log.G(ctx).Debug("attach done")