Add cancellable attach.

client.go

@@ -554,24 +554,37 @@ type hijackOptions struct {
 	data           interface{}
 }
 
-func (c *Client) hijack(method, path string, hijackOptions hijackOptions) error {
+type CloseWaiter interface {
+	io.Closer
+	Wait() error
+}
+
+type waiterFunc func() error
+
+func (w waiterFunc) Wait() error { return w() }
+
+type closerFunc func() error
+
+func (c closerFunc) Close() error { return c() }
+
+func (c *Client) hijack(method, path string, hijackOptions hijackOptions) (CloseWaiter, error) {
 	if path != "/version" && !c.SkipServerVersionCheck && c.expectedAPIVersion == nil {
 		err := c.checkAPIVersion()
 		if err != nil {
-			return err
+			return nil, err
 		}
 	}
 	var params io.Reader
 	if hijackOptions.data != nil {
 		buf, err := json.Marshal(hijackOptions.data)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		params = bytes.NewBuffer(buf)
 	}
 	req, err := http.NewRequest(method, c.getURL(path), params)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	req.Header.Set("Content-Type", "plain/text")
 	req.Header.Set("Connection", "Upgrade")
@@ -586,74 +599,103 @@ func (c *Client) hijack(method, path string, hijackOptions hijackOptions) error
 	if c.TLSConfig != nil && protocol != "unix" {
 		dial, err = tlsDialWithDialer(c.Dialer, protocol, address, c.TLSConfig)
 		if err != nil {
-			return err
+			return nil, err
 		}
 	} else {
 		dial, err = c.Dialer.Dial(protocol, address)
 		if err != nil {
-			return err
+			return nil, err
 		}
 	}
-	clientconn := httputil.NewClientConn(dial, nil)
-	defer clientconn.Close()
-	clientconn.Do(req)
-	if hijackOptions.success != nil {
-		hijackOptions.success <- struct{}{}
-		<-hijackOptions.success
-	}
-	rwc, br := clientconn.Hijack()
-	defer rwc.Close()
-	errChanOut := make(chan error, 1)
-	errChanIn := make(chan error, 1)
-	if hijackOptions.stdout == nil && hijackOptions.stderr == nil {
-		close(errChanOut)
-	} else {
-		// Only copy if hijackOptions.stdout and/or hijackOptions.stderr is actually set.
-		// Otherwise, if the only stream you care about is stdin, your attach session
-		// will "hang" until the container terminates, even though you're not reading
-		// stdout/stderr
-		if hijackOptions.stdout == nil {
-			hijackOptions.stdout = ioutil.Discard
-		}
-		if hijackOptions.stderr == nil {
-			hijackOptions.stderr = ioutil.Discard
-		}
 
-		go func() {
-			defer func() {
-				if hijackOptions.in != nil {
-					if closer, ok := hijackOptions.in.(io.Closer); ok {
-						closer.Close()
+	errs := make(chan error)
+	quit := make(chan struct{})
+	go func() {
+		clientconn := httputil.NewClientConn(dial, nil)
+		defer clientconn.Close()
+		clientconn.Do(req)
+		if hijackOptions.success != nil {
+			hijackOptions.success <- struct{}{}
+			<-hijackOptions.success
+		}
+		rwc, br := clientconn.Hijack()
+		defer rwc.Close()
+
+		errChanOut := make(chan error, 1)
+		errChanIn := make(chan error, 1)
+		if hijackOptions.stdout == nil && hijackOptions.stderr == nil {
+			close(errChanOut)
+		} else {
+			// Only copy if hijackOptions.stdout and/or hijackOptions.stderr is actually set.
+			// Otherwise, if the only stream you care about is stdin, your attach session
+			// will "hang" until the container terminates, even though you're not reading
+			// stdout/stderr
+			if hijackOptions.stdout == nil {
+				hijackOptions.stdout = ioutil.Discard
+			}
+			if hijackOptions.stderr == nil {
+				hijackOptions.stderr = ioutil.Discard
+			}
+
+			go func() {
+				defer func() {
+					if hijackOptions.in != nil {
+						if closer, ok := hijackOptions.in.(io.Closer); ok {
+							closer.Close()
+						}
+						errChanIn <- nil
 					}
-					errChanIn <- nil
+				}()
+
+				var err error
+				if hijackOptions.setRawTerminal {
+					_, err = io.Copy(hijackOptions.stdout, br)
+				} else {
+					_, err = stdcopy.StdCopy(hijackOptions.stdout, hijackOptions.stderr, br)
 				}
+				errChanOut <- err
 			}()
+		}
 
+		go func() {
 			var err error
-			if hijackOptions.setRawTerminal {
-				_, err = io.Copy(hijackOptions.stdout, br)
-			} else {
-				_, err = stdcopy.StdCopy(hijackOptions.stdout, hijackOptions.stderr, br)
+			if hijackOptions.in != nil {
+				_, err = io.Copy(rwc, hijackOptions.in)
 			}
-			errChanOut <- err
+			errChanIn <- err
+			rwc.(interface {
+				CloseWrite() error
+			}).CloseWrite()
 		}()
-	}
-	go func() {
-		var err error
-		if hijackOptions.in != nil {
-			_, err = io.Copy(rwc, hijackOptions.in)
-		}
-		errChanIn <- err
-		rwc.(interface {
-			CloseWrite() error
-		}).CloseWrite()
+
+		var errIn error
+		select {
+		case errIn = <-errChanIn:
+		case <-quit:
+			return
+		}
+
+		var errOut error
+		select {
+		case errOut = <-errChanOut:
+		case <-quit:
+			return
+		}
+
+		if errIn != nil {
+			errs <- errIn
+		} else {
+			errs <- errOut
+		}
 	}()
-	errIn := <-errChanIn
-	errOut := <-errChanOut
-	if errIn != nil {
-		return errIn
-	}
-	return errOut
+
+	return struct {
+		closerFunc
+		waiterFunc
+	}{
+		closerFunc(func() error { close(quit); return nil }),
+		waiterFunc(func() error { return <-errs }),
+	}, nil
 }
 
 func (c *Client) getURL(path string) string {

container.go

@@ -1036,8 +1036,20 @@ type AttachToContainerOptions struct {
 //
 // See https://goo.gl/NKpkFk for more details.
 func (c *Client) AttachToContainer(opts AttachToContainerOptions) error {
+	cw, err := c.AttachToContainerNonBlocking(opts)
+	if err != nil {
+		return err
+	}
+	return cw.Wait()
+}
+
+// AttachToContainerNonBlocking attaches to a container, using the given options.
+// This function does not block.
+//
+// See https://goo.gl/NKpkFk for more details.
+func (c *Client) AttachToContainerNonBlocking(opts AttachToContainerOptions) (CloseWaiter, error) {
 	if opts.Container == "" {
-		return &NoSuchContainer{ID: opts.Container}
+		return nil, &NoSuchContainer{ID: opts.Container}
 	}
 	path := "/containers/" + opts.Container + "/attach?" + queryString(opts)
 	return c.hijack("POST", path, hijackOptions{

exec.go

@@ -83,8 +83,24 @@ type StartExecOptions struct {
 //
 // See https://goo.gl/iQCnto for more details
 func (c *Client) StartExec(id string, opts StartExecOptions) error {
+	cw, err := c.StartExecNonBlocking(id, opts)
+	if err != nil {
+		return err
+	}
+	if cw != nil {
+		return cw.Wait()
+	}
+	return nil
+}
+
+// StartExecNonBlocking starts a previously set up exec instance id. If opts.Detach is
+// true, it returns after starting the exec command. Otherwise, it sets up an
+// interactive session with the exec command.
+//
+// See https://goo.gl/iQCnto for more details
+func (c *Client) StartExecNonBlocking(id string, opts StartExecOptions) (CloseWaiter, error) {
 	if id == "" {
-		return &NoSuchExec{ID: id}
+		return nil, &NoSuchExec{ID: id}
 	}
 
 	path := fmt.Sprintf("/exec/%s/start", id)
@@ -93,12 +109,12 @@ func (c *Client) StartExec(id string, opts StartExecOptions) error {
 		resp, err := c.do("POST", path, doOptions{data: opts})
 		if err != nil {
 			if e, ok := err.(*Error); ok && e.Status == http.StatusNotFound {
-				return &NoSuchExec{ID: id}
+				return nil, &NoSuchExec{ID: id}
 			}
-			return err
+			return nil, err
 		}
 		defer resp.Body.Close()
-		return nil
+		return nil, nil
 	}
 
 	return c.hijack("POST", path, hijackOptions{