Finish up remote exec implementation

Signed-off-by: Peter Hunt <pehunt@redhat.com>

3 files changed, 78 insertions, 48 deletions

diff --git a/pkg/varlinkapi/attach.go b/pkg/varlinkapi/attach.go
index 97ba525a5..1f8d48eb9 100644
--- a/pkg/varlinkapi/attach.go
+++ b/pkg/varlinkapi/attach.go
@@ -68,7 +68,7 @@ func (i *LibpodAPI) Attach(call iopodman.VarlinkCall, name string, detachKeys st
 	reader, writer, _, pw, streams := setupStreams(call)
 
 	go func() {
-		if err := virtwriter.Reader(reader, nil, nil, pw, resize); err != nil {
+		if err := virtwriter.Reader(reader, nil, nil, pw, resize, nil); err != nil {
 			errChan <- err
 		}
 	}()
@@ -83,7 +83,7 @@ func (i *LibpodAPI) Attach(call iopodman.VarlinkCall, name string, detachKeys st
 		logrus.Error(finalErr)
 	}
 
-	if err = virtwriter.HangUp(writer); err != nil {
+	if err = virtwriter.HangUp(writer, 0); err != nil {
 		logrus.Errorf("Failed to HANG-UP attach to %s: %s", ctr.ID(), err.Error())
 	}
 	return call.Writer.Flush()
diff --git a/pkg/varlinkapi/containers.go b/pkg/varlinkapi/containers.go
index 19a8bfd2e..4714f3fa0 100644
--- a/pkg/varlinkapi/containers.go
+++ b/pkg/varlinkapi/containers.go
@@ -796,45 +796,55 @@ func (i *LibpodAPI) ExecContainer(call iopodman.VarlinkCall, opts iopodman.ExecO
 	if opts.Workdir != nil {
 		workDir = *opts.Workdir
 	}
+	// ACK the client upgrade request
+	call.ReplyExecContainer()
 
 	resizeChan := make(chan remotecommand.TerminalSize)
 	errChan := make(chan error)
 
 	reader, writer, _, pipeWriter, streams := setupStreams(call)
+	//reader, _, _, pipeWriter, streams := setupStreams(call)
+	ecChan := make(chan uint32, 1)
 
 	go func() {
 		fmt.Printf("ExecContainer Start Reader\n")
-		if err := virtwriter.Reader(reader, nil, nil, pipeWriter, resizeChan); err != nil {
-			fmt.Printf("ExecContainer Reader err %s, %s\n", err.Error(), errors.Cause(err).Error())
-			errChan <- err
+		if err := virtwriter.Reader(reader, nil, nil, pipeWriter, resizeChan, nil); err != nil {
+			//fmt.Printf("ExecContainer Reader err %s, %s\n", err.Error(), errors.Cause(err).Error())
+			errChan <- errors.Wrapf(err, "error")
 		}
 	}()
 
-	// Debugging...
-	time.Sleep(5 * time.Second)
-
+	fmt.Printf("ExecContainer Start ctr.Exec\n")
+	// TODO detach keys and resize
+	// TODO add handling for exit code
+	// TODO capture exit code and return to main thread
 	go func() {
-		fmt.Printf("ExecContainer Start ctr.Exec\n")
-		// TODO detach keys and resize
-		// TODO add handling for exit code
-		// TODO capture exit code and return to main thread
-		_, err := ctr.Exec(opts.Tty, opts.Privileged, envs, opts.Cmd, user, workDir, streams, 0, nil, "")
+		ec, err := ctr.Exec(opts.Tty, opts.Privileged, envs, opts.Cmd, user, workDir, streams, 0, resizeChan, "")
 		if err != nil {
-			fmt.Printf("ExecContainer Exec err %s, %s\n", err.Error(), errors.Cause(err).Error())
-			errChan <- errors.Wrapf(err, "ExecContainer failed for container %s", ctr.ID())
+			logrus.Errorf("ExecContainer Exec err %s, %s\n", err.Error(), errors.Cause(err).Error())
+			errChan <-err
 		}
-	}()
+		ecChan <-uint32(ec)
 
-	execErr := <-errChan
+	}()
 
-	if execErr != nil && errors.Cause(execErr) != io.EOF {
-		fmt.Printf("ExecContainer err: %s\n", execErr.Error())
-		return call.ReplyErrorOccurred(execErr.Error())
+	ec := uint32(125)
+	var execErr error
+	select {
+		case execErr = <-errChan:
+			fmt.Println(execErr.Error())
+		case ec = <-ecChan:
+			fmt.Println("found", ec)
 	}
 
-	if err = virtwriter.HangUp(writer); err != nil {
-		fmt.Printf("ExecContainer hangup err: %s\n", err.Error())
+	// TODO FIXME prevent all of vthese conversions
+	if err = virtwriter.HangUp(writer, int(ec)); err != nil {
 		logrus.Errorf("ExecContainer failed to HANG-UP on %s: %s", ctr.ID(), err.Error())
 	}
-	return call.Writer.Flush()
+	defer fmt.Println("Succeeded in exec'ing")
+	if err := call.Writer.Flush(); err != nil {
+		logrus.Errorf("Exec Container err: %s", err.Error())
+	}
+
+	return execErr
 }
diff --git a/pkg/varlinkapi/virtwriter/virtwriter.go b/pkg/varlinkapi/virtwriter/virtwriter.go
index 0da2a91fc..23f945704 100644
--- a/pkg/varlinkapi/virtwriter/virtwriter.go
+++ b/pkg/varlinkapi/virtwriter/virtwriter.go
@@ -89,10 +89,14 @@ func (v VirtWriteCloser) Write(input []byte) (int, error) {
 }
 
 // Reader decodes the content that comes over the wire and directs it to the proper destination.
-func Reader(r *bufio.Reader, output io.Writer, errput io.Writer, input io.Writer, resize chan remotecommand.TerminalSize) error {
+func Reader(r *bufio.Reader, output, errput, input io.Writer, resize chan remotecommand.TerminalSize, execEcChan chan int) error {
 	var messageSize int64
 	headerBytes := make([]byte, 8)
 
+	if r == nil {
+		return errors.Errorf("Reader must not be nil")
+	}
+
 	for {
 		n, err := io.ReadFull(r, headerBytes)
 		if err != nil {
@@ -106,44 +110,57 @@ func Reader(r *bufio.Reader, output io.Writer, errput io.Writer, input io.Writer
 
 		switch IntToSocketDest(int(headerBytes[0])) {
 		case ToStdout:
-			_, err := io.CopyN(output, r, messageSize)
-			if err != nil {
-				return err
+			if output != nil {
+				_, err := io.CopyN(output, r, messageSize)
+				if err != nil {
+						return errors.Wrapf(err, "issue stdout")
+				}
 			}
 		case ToStderr:
-			_, err := io.CopyN(errput, r, messageSize)
-			if err != nil {
-				return err
+			if errput != nil {
+				_, err := io.CopyN(errput, r, messageSize)
+				if err != nil {
+					return err
+						return errors.Wrapf(err, "issue stderr")
+				}
 			}
 		case ToStdin:
-			_, err := io.CopyN(input, r, messageSize)
-			if err != nil {
-				return err
-			}
-		case TerminalResize:
-			out := make([]byte, messageSize)
-			if messageSize > 0 {
-				_, err = io.ReadFull(r, out)
-
+			if input != nil {
+				_, err := io.CopyN(input, r, messageSize)
 				if err != nil {
-					return err
+						return errors.Wrapf(err, "issue stdin")
 				}
 			}
-			// Resize events come over in bytes, need to be reserialized
-			resizeEvent := remotecommand.TerminalSize{}
-			if err := json.Unmarshal(out, &resizeEvent); err != nil {
-				return err
+		case TerminalResize:
+			if resize != nil {
+				out := make([]byte, messageSize)
+				if messageSize > 0 {
+					_, err = io.ReadFull(r, out)
+
+					if err != nil {
+						return errors.Wrapf(err, "issue resizing")
+					}
+				}
+				// Resize events come over in bytes, need to be reserialized
+				resizeEvent := remotecommand.TerminalSize{}
+				if err := json.Unmarshal(out, &resizeEvent); err != nil {
+					return errors.Wrapf(err, "issue resizing")
+				}
+				resize <- resizeEvent
 			}
-			resize <- resizeEvent
 		case Quit:
 			out := make([]byte, messageSize)
 			if messageSize > 0 {
 				_, err = io.ReadFull(r, out)
 
 				if err != nil {
-					return err
+					return errors.Wrapf(err, "issue quitting")
 				}
 			}
+			if execEcChan != nil {
+				ecInt := binary.BigEndian.Uint32(out)
+				execEcChan <-int(ecInt)
+			}
 			return nil
 
 		default:
@@ -154,9 +171,12 @@ func Reader(r *bufio.Reader, output io.Writer, errput io.Writer, input io.Writer
 }
 
 // HangUp sends message to peer to close connection
-func HangUp(writer *bufio.Writer) (err error) {
+func HangUp(writer *bufio.Writer, ec int) (err error) {
 	n := 0
-	msg := []byte("HANG-UP")
+	msg := make([]byte, 4)
+
+	binary.LittleEndian.PutUint32(msg, uint32(ec))
+
 
 	writeQuit := NewVirtWriteCloser(writer, Quit)
 	if n, err = writeQuit.Write(msg); err != nil {