Fix stdin handling in engine.Sender and engine.Receiver

This introduces a superficial change to the Beam API:

* `beam.SendPipe` is renamed to the more accurate `beam.SendRPipe`
* `beam.SendWPipe` is introduced as a mirror to `SendRPipe`

There is no other change in the beam API.

Docker-DCO-1.1-Signed-off-by: Solomon Hykes <solomon@docker.com> (github: shykes)

engine/remote.go

@@ -92,17 +92,17 @@ func (rcv *Receiver) Run() error {
 		}
 		cmd := data.Message(p).Get("cmd")
 		job := rcv.Engine.Job(cmd[0], cmd[1:]...)
-		stdout, err := beam.SendPipe(peer, data.Empty().Set("cmd", "log", "stdout").Bytes())
+		stdout, err := beam.SendRPipe(peer, data.Empty().Set("cmd", "log", "stdout").Bytes())
 		if err != nil {
 			return err
 		}
 		job.Stdout.Add(stdout)
-		stderr, err := beam.SendPipe(peer, data.Empty().Set("cmd", "log", "stderr").Bytes())
+		stderr, err := beam.SendRPipe(peer, data.Empty().Set("cmd", "log", "stderr").Bytes())
 		if err != nil {
 			return err
 		}
 		job.Stderr.Add(stderr)
-		stdin, err := beam.SendPipe(peer, data.Empty().Set("cmd", "log", "stdin").Bytes())
+		stdin, err := beam.SendWPipe(peer, data.Empty().Set("cmd", "log", "stdin").Bytes())
 		if err != nil {
 			return err
 		}

pkg/beam/beam.go

@@ -29,17 +29,48 @@ type ReceiveSender interface {
 	Sender
 }
 
-func SendPipe(dst Sender, data []byte) (*os.File, error) {
+const (
+	R int = 1 << (32 - 1 - iota)
+	W
+)
+
+func sendPipe(dst Sender, data []byte, mode int) (*os.File, error) {
 	r, w, err := os.Pipe()
 	if err != nil {
 		return nil, err
 	}
-	if err := dst.Send(data, r); err != nil {
+	var (
-		r.Close()
+		remote *os.File
-		w.Close()
+		local  *os.File
+	)
+	if mode == R {
+		remote = r
+		local = w
+	} else if mode == W {
+		remote = w
+		local = r
+	}
+	if err := dst.Send(data, remote); err != nil {
+		local.Close()
+		remote.Close()
 		return nil, err
 	}
-	return w, nil
+	return local, nil
+
+}
+
+// SendRPipe create a pipe and sends its *read* end attached in a beam message
+// to `dst`, with `data` as the message payload.
+// It returns the *write* end of the pipe, or an error.
+func SendRPipe(dst Sender, data []byte) (*os.File, error) {
+	return sendPipe(dst, data, R)
+}
+
+// SendWPipe create a pipe and sends its *read* end attached in a beam message
+// to `dst`, with `data` as the message payload.
+// It returns the *write* end of the pipe, or an error.
+func SendWPipe(dst Sender, data []byte) (*os.File, error) {
+	return sendPipe(dst, data, W)
 }
 
 func SendConn(dst Sender, data []byte) (conn *UnixConn, err error) {

pkg/beam/examples/beamsh/beamsh.go

@@ -257,12 +257,12 @@ func Handlers(sink beam.Sender) (*beam.UnixConn, error) {
 				if handler == nil {
 					return
 				}
-				stdout, err := beam.SendPipe(conn, data.Empty().Set("cmd", "log", "stdout").Set("fromcmd", cmd...).Bytes())
+				stdout, err := beam.SendRPipe(conn, data.Empty().Set("cmd", "log", "stdout").Set("fromcmd", cmd...).Bytes())
 				if err != nil {
 					return
 				}
 				defer stdout.Close()
-				stderr, err := beam.SendPipe(conn, data.Empty().Set("cmd", "log", "stderr").Set("fromcmd", cmd...).Bytes())
+				stderr, err := beam.SendRPipe(conn, data.Empty().Set("cmd", "log", "stderr").Set("fromcmd", cmd...).Bytes())
 				if err != nil {
 					return
 				}

pkg/beam/examples/beamsh/builtins.go

@@ -272,7 +272,7 @@ func CmdPrint(args []string, stdout, stderr io.Writer, in beam.Receiver, out bea
 		}
 		// Skip commands
 		if a != nil && data.Message(payload).Get("cmd") == nil {
-			dup, err := beam.SendPipe(out, payload)
+			dup, err := beam.SendRPipe(out, payload)
 			if err != nil {
 				a.Close()
 				return

pkg/beam/router.go

@@ -78,7 +78,7 @@ func (route *Route) Tee(dst Sender) *Route {
 			return inner(payload, attachment)
 		}
 		// Setup the tee
-		w, err := SendPipe(dst, payload)
+		w, err := SendRPipe(dst, payload)
 		if err != nil {
 			return err
 		}