7 files changed, 620 insertions, 17 deletions

diff --git a/pkg/adapter/containers.go b/pkg/adapter/containers.go
index 19ff2270c..d4e8e30d9 100644
--- a/pkg/adapter/containers.go
+++ b/pkg/adapter/containers.go
@@ -255,14 +255,14 @@ func (r *LocalRuntime) Log(c *cliconfig.LogsValues, options *libpod.LogOptions)
 
 // CreateContainer creates a libpod container
 func (r *LocalRuntime) CreateContainer(ctx context.Context, c *cliconfig.CreateValues) (string, error) {
-	results := shared.NewIntermediateLayer(&c.PodmanCommand)
+	results := shared.NewIntermediateLayer(&c.PodmanCommand, false)
 	ctr, _, err := shared.CreateContainer(ctx, &results, r.Runtime)
 	return ctr.ID(), err
 }
 
 // Run a libpod container
 func (r *LocalRuntime) Run(ctx context.Context, c *cliconfig.RunValues, exitCode int) (int, error) {
-	results := shared.NewIntermediateLayer(&c.PodmanCommand)
+	results := shared.NewIntermediateLayer(&c.PodmanCommand, false)
 
 	ctr, createConfig, err := shared.CreateContainer(ctx, &results, r.Runtime)
 	if err != nil {
@@ -397,3 +397,49 @@ func ReadExitFile(runtimeTmp, ctrID string) (int, error) {
 
 	return exitCode, nil
 }
+
+// Ps ...
+func (r *LocalRuntime) Ps(c *cliconfig.PsValues, opts shared.PsOptions) ([]shared.PsContainerOutput, error) {
+	maxWorkers := shared.Parallelize("ps")
+	if c.GlobalIsSet("max-workers") {
+		maxWorkers = c.GlobalFlags.MaxWorks
+	}
+	logrus.Debugf("Setting maximum workers to %d", maxWorkers)
+	return shared.GetPsContainerOutput(r.Runtime, opts, c.Filter, maxWorkers)
+}
+
+// Attach ...
+func (r *LocalRuntime) Attach(ctx context.Context, c *cliconfig.AttachValues) error {
+	var (
+		ctr *libpod.Container
+		err error
+	)
+
+	if c.Latest {
+		ctr, err = r.Runtime.GetLatestContainer()
+	} else {
+		ctr, err = r.Runtime.LookupContainer(c.InputArgs[0])
+	}
+
+	if err != nil {
+		return errors.Wrapf(err, "unable to exec into %s", c.InputArgs[0])
+	}
+
+	conState, err := ctr.State()
+	if err != nil {
+		return errors.Wrapf(err, "unable to determine state of %s", ctr.ID())
+	}
+	if conState != libpod.ContainerStateRunning {
+		return errors.Errorf("you can only attach to running containers")
+	}
+
+	inputStream := os.Stdin
+	if c.NoStdin {
+		inputStream = nil
+	}
+	// If the container is in a pod, also set to recursively start dependencies
+	if err := StartAttachCtr(ctx, ctr, os.Stdout, os.Stderr, inputStream, c.DetachKeys, c.SigProxy, false, ctr.PodID() != ""); err != nil && errors.Cause(err) != libpod.ErrDetach {
+		return errors.Wrapf(err, "error attaching to container %s", ctr.ID())
+	}
+	return nil
+}
diff --git a/pkg/adapter/containers_remote.go b/pkg/adapter/containers_remote.go
index 1892629ea..9ca823760 100644
--- a/pkg/adapter/containers_remote.go
+++ b/pkg/adapter/containers_remote.go
@@ -6,19 +6,25 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
+	"os"
 	"strconv"
 	"syscall"
 	"time"
 
 	"github.com/containers/libpod/cmd/podman/cliconfig"
 	"github.com/containers/libpod/cmd/podman/shared"
-	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
-
 	iopodman "github.com/containers/libpod/cmd/podman/varlink"
 	"github.com/containers/libpod/libpod"
 	"github.com/containers/libpod/pkg/inspect"
+	"github.com/containers/libpod/pkg/varlinkapi/virtwriter"
+	"github.com/docker/docker/pkg/term"
+	"github.com/opencontainers/runtime-spec/specs-go"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 	"github.com/varlink/go/varlink"
+	"golang.org/x/crypto/ssh/terminal"
+	"k8s.io/client-go/tools/remotecommand"
 )
 
 // Inspect returns an inspect struct from varlink
@@ -71,6 +77,19 @@ func (r *LocalRuntime) ContainerState(name string) (*libpod.ContainerState, erro
 
 }
 
+// Spec obtains the container spec.
+func (r *LocalRuntime) Spec(name string) (*specs.Spec, error) {
+	reply, err := iopodman.Spec().Call(r.Conn, name)
+	if err != nil {
+		return nil, err
+	}
+	data := specs.Spec{}
+	if err := json.Unmarshal([]byte(reply), &data); err != nil {
+		return nil, err
+	}
+	return &data, nil
+}
+
 // LookupContainer gets basic information about container over a varlink
 // connection and then translates it to a *Container
 func (r *LocalRuntime) LookupContainer(idOrName string) (*Container, error) {
@@ -79,10 +98,6 @@ func (r *LocalRuntime) LookupContainer(idOrName string) (*Container, error) {
 		return nil, err
 	}
 	config := r.Config(idOrName)
-	if err != nil {
-		return nil, err
-	}
-
 	return &Container{
 		remoteContainer{
 			r,
@@ -316,26 +331,209 @@ func (r *LocalRuntime) CreateContainer(ctx context.Context, c *cliconfig.CreateV
 		// TODO need to add attach when that function becomes available
 		return "", errors.New("the remote client only supports detached containers")
 	}
-	results := shared.NewIntermediateLayer(&c.PodmanCommand)
+	results := shared.NewIntermediateLayer(&c.PodmanCommand, true)
 	return iopodman.CreateContainer().Call(r.Conn, results.MakeVarlink())
 }
 
 // Run creates a container overvarlink and then starts it
 func (r *LocalRuntime) Run(ctx context.Context, c *cliconfig.RunValues, exitCode int) (int, error) {
+	// FIXME
+	// podman-remote run -it alpine ls DOES NOT WORK YET
+	// podman-remote run -it alpine /bin/sh does, i suspect there is some sort of
+	// timing issue between the socket availability and terminal setup and the command
+	// being run.
+
 	// TODO the exit codes for run need to be figured out for remote connections
-	if !c.Bool("detach") {
-		return 0, errors.New("the remote client only supports detached containers")
-	}
-	results := shared.NewIntermediateLayer(&c.PodmanCommand)
+	results := shared.NewIntermediateLayer(&c.PodmanCommand, true)
 	cid, err := iopodman.CreateContainer().Call(r.Conn, results.MakeVarlink())
 	if err != nil {
 		return 0, err
 	}
-	fmt.Println(cid)
-	_, err = iopodman.StartContainer().Call(r.Conn, cid)
-	return 0, err
+	if c.Bool("detach") {
+		_, err := iopodman.StartContainer().Call(r.Conn, cid)
+		fmt.Println(cid)
+		return 0, err
+	}
+
+	errChan, err := r.attach(ctx, os.Stdin, os.Stdout, cid, true)
+	if err != nil {
+		return 0, err
+	}
+	finalError := <-errChan
+	return 0, finalError
 }
 
 func ReadExitFile(runtimeTmp, ctrID string) (int, error) {
 	return 0, libpod.ErrNotImplemented
 }
+
+// Ps ...
+func (r *LocalRuntime) Ps(c *cliconfig.PsValues, opts shared.PsOptions) ([]shared.PsContainerOutput, error) {
+	var psContainers []shared.PsContainerOutput
+	last := int64(c.Last)
+	PsOpts := iopodman.PsOpts{
+		All:     c.All,
+		Filters: &c.Filter,
+		Last:    &last,
+		Latest:  &c.Latest,
+		NoTrunc: &c.NoTrunct,
+		Pod:     &c.Pod,
+		Quiet:   &c.Quiet,
+		Sort:    &c.Sort,
+		Sync:    &c.Sync,
+	}
+	containers, err := iopodman.Ps().Call(r.Conn, PsOpts)
+	if err != nil {
+		return nil, err
+	}
+	for _, ctr := range containers {
+		createdAt, err := time.Parse(time.RFC3339Nano, ctr.CreatedAt)
+		if err != nil {
+			return nil, err
+		}
+		exitedAt, err := time.Parse(time.RFC3339Nano, ctr.ExitedAt)
+		if err != nil {
+			return nil, err
+		}
+		startedAt, err := time.Parse(time.RFC3339Nano, ctr.StartedAt)
+		if err != nil {
+			return nil, err
+		}
+		containerSize := shared.ContainerSize{
+			RootFsSize: ctr.RootFsSize,
+			RwSize:     ctr.RwSize,
+		}
+		state, err := libpod.StringToContainerStatus(ctr.State)
+		if err != nil {
+			return nil, err
+		}
+		psc := shared.PsContainerOutput{
+			ID:        ctr.Id,
+			Image:     ctr.Image,
+			Command:   ctr.Command,
+			Created:   ctr.Created,
+			Ports:     ctr.Ports,
+			Names:     ctr.Names,
+			IsInfra:   ctr.IsInfra,
+			Status:    ctr.Status,
+			State:     state,
+			Pid:       int(ctr.PidNum),
+			Size:      &containerSize,
+			Pod:       ctr.Pod,
+			CreatedAt: createdAt,
+			ExitedAt:  exitedAt,
+			StartedAt: startedAt,
+			Labels:    ctr.Labels,
+			PID:       ctr.NsPid,
+			Cgroup:    ctr.Cgroup,
+			IPC:       ctr.Ipc,
+			MNT:       ctr.Mnt,
+			NET:       ctr.Net,
+			PIDNS:     ctr.PidNs,
+			User:      ctr.User,
+			UTS:       ctr.Uts,
+			Mounts:    ctr.Mounts,
+		}
+		psContainers = append(psContainers, psc)
+	}
+	return psContainers, nil
+}
+
+func (r *LocalRuntime) attach(ctx context.Context, stdin, stdout *os.File, cid string, start bool) (chan error, error) {
+	var (
+		oldTermState *term.State
+	)
+	errChan := make(chan error)
+	spec, err := r.Spec(cid)
+	if err != nil {
+		return nil, err
+	}
+	resize := make(chan remotecommand.TerminalSize)
+	haveTerminal := terminal.IsTerminal(int(os.Stdin.Fd()))
+
+	// Check if we are attached to a terminal. If we are, generate resize
+	// events, and set the terminal to raw mode
+	if haveTerminal && spec.Process.Terminal {
+		logrus.Debugf("Handling terminal attach")
+
+		subCtx, cancel := context.WithCancel(ctx)
+		defer cancel()
+
+		resizeTty(subCtx, resize)
+		oldTermState, err = term.SaveState(os.Stdin.Fd())
+		if err != nil {
+			return nil, errors.Wrapf(err, "unable to save terminal state")
+		}
+
+		logrus.SetFormatter(&RawTtyFormatter{})
+		term.SetRawTerminal(os.Stdin.Fd())
+
+	}
+	// TODO add detach keys support
+	_, err = iopodman.Attach().Send(r.Conn, varlink.Upgrade, cid, "", start)
+	if err != nil {
+		restoreTerminal(oldTermState)
+		return nil, err
+	}
+
+	// These are the varlink sockets
+	reader := r.Conn.Reader
+	writer := r.Conn.Writer
+
+	// These are the special writers that encode input from the client.
+	varlinkStdinWriter := virtwriter.NewVirtWriteCloser(writer, virtwriter.ToStdin)
+	varlinkResizeWriter := virtwriter.NewVirtWriteCloser(writer, virtwriter.TerminalResize)
+
+	go func() {
+		// Read from the wire and direct to stdout or stderr
+		err := virtwriter.Reader(reader, stdout, os.Stderr, nil, nil)
+		defer restoreTerminal(oldTermState)
+		errChan <- err
+	}()
+
+	go func() {
+		for termResize := range resize {
+			b, err := json.Marshal(termResize)
+			if err != nil {
+				defer restoreTerminal(oldTermState)
+				errChan <- err
+			}
+			_, err = varlinkResizeWriter.Write(b)
+			if err != nil {
+				defer restoreTerminal(oldTermState)
+				errChan <- err
+			}
+		}
+	}()
+
+	// Takes stdinput and sends it over the wire after being encoded
+	go func() {
+		if _, err := io.Copy(varlinkStdinWriter, stdin); err != nil {
+			defer restoreTerminal(oldTermState)
+			errChan <- err
+		}
+
+	}()
+	return errChan, nil
+
+}
+
+// Attach to a remote terminal
+func (r *LocalRuntime) Attach(ctx context.Context, c *cliconfig.AttachValues) error {
+	ctr, err := r.LookupContainer(c.InputArgs[0])
+	if err != nil {
+		return nil
+	}
+	if ctr.state.State != libpod.ContainerStateRunning {
+		return errors.New("you can only attach to running containers")
+	}
+	inputStream := os.Stdin
+	if c.NoStdin {
+		inputStream = nil
+	}
+	errChan, err := r.attach(ctx, inputStream, os.Stdout, c.InputArgs[0], false)
+	if err != nil {
+		return err
+	}
+	return <-errChan
+}
diff --git a/pkg/spec/spec.go b/pkg/spec/spec.go
index 9b6bd089e..0371b6d4d 100644
--- a/pkg/spec/spec.go
+++ b/pkg/spec/spec.go
@@ -132,6 +132,9 @@ func CreateConfigToOCISpec(config *CreateConfig) (*spec.Spec, error) { //nolint
 			Options:     []string{"rprivate", "nosuid", "noexec", "nodev", r, "rbind"},
 		}
 		g.AddMount(sysMnt)
+		if !config.Privileged && isRootless {
+			g.AddLinuxMaskedPaths("/sys/kernel")
+		}
 	}
 	if isRootless {
 		nGids, err := getAvailableGids()
diff --git a/pkg/varlinkapi/attach.go b/pkg/varlinkapi/attach.go
new file mode 100644
index 000000000..9e2a265be
--- /dev/null
+++ b/pkg/varlinkapi/attach.go
@@ -0,0 +1,103 @@
+// +build varlink
+
+package varlinkapi
+
+import (
+	"bufio"
+	"io"
+
+	"github.com/containers/libpod/cmd/podman/varlink"
+	"github.com/containers/libpod/libpod"
+	"github.com/containers/libpod/pkg/varlinkapi/virtwriter"
+	"github.com/sirupsen/logrus"
+	"k8s.io/client-go/tools/remotecommand"
+)
+
+func setupStreams(call iopodman.VarlinkCall) (*bufio.Reader, *bufio.Writer, *io.PipeReader, *io.PipeWriter, *libpod.AttachStreams) {
+
+	// These are the varlink sockets
+	reader := call.Call.Reader
+	writer := call.Call.Writer
+
+	// This pipe is used to pass stdin from the client to the input stream
+	// once the msg has been "decoded"
+	pr, pw := io.Pipe()
+
+	stdoutWriter := virtwriter.NewVirtWriteCloser(writer, virtwriter.ToStdout)
+	// TODO if runc ever starts passing stderr, we can too
+	//stderrWriter := NewVirtWriteCloser(writer, ToStderr)
+
+	streams := libpod.AttachStreams{
+		OutputStream: stdoutWriter,
+		InputStream:  pr,
+		// Runc eats the error stream
+		ErrorStream:  stdoutWriter,
+		AttachInput:  true,
+		AttachOutput: true,
+		// Runc eats the error stream
+		AttachError: true,
+	}
+	return reader, writer, pr, pw, &streams
+}
+
+// Attach connects to a containers console
+func (i *LibpodAPI) Attach(call iopodman.VarlinkCall, name string, detachKeys string, start bool) error {
+	var finalErr error
+	resize := make(chan remotecommand.TerminalSize)
+	errChan := make(chan error)
+
+	if !call.WantsUpgrade() {
+		return call.ReplyErrorOccurred("client must use upgraded connection to attach")
+	}
+	ctr, err := i.Runtime.LookupContainer(name)
+	if err != nil {
+		return call.ReplyErrorOccurred(err.Error())
+	}
+
+	reader, writer, _, pw, streams := setupStreams(call)
+
+	go func() {
+		if err := virtwriter.Reader(reader, nil, nil, pw, resize); err != nil {
+			errChan <- err
+		}
+	}()
+
+	if start {
+		finalErr = startAndAttach(ctr, streams, detachKeys, resize, errChan)
+	} else {
+		finalErr = attach(ctr, streams, detachKeys, resize, errChan)
+	}
+
+	if finalErr != libpod.ErrDetach && finalErr != nil {
+		logrus.Error(finalErr)
+	}
+	quitWriter := virtwriter.NewVirtWriteCloser(writer, virtwriter.Quit)
+	_, err = quitWriter.Write([]byte("HANG-UP"))
+	// TODO error handling is not quite right here yet
+	return call.Writer.Flush()
+}
+
+func attach(ctr *libpod.Container, streams *libpod.AttachStreams, detachKeys string, resize chan remotecommand.TerminalSize, errChan chan error) error {
+	go func() {
+		if err := ctr.Attach(streams, detachKeys, resize); err != nil {
+			errChan <- err
+		}
+	}()
+	attachError := <-errChan
+	return attachError
+}
+
+func startAndAttach(ctr *libpod.Container, streams *libpod.AttachStreams, detachKeys string, resize chan remotecommand.TerminalSize, errChan chan error) error {
+	var finalErr error
+	attachChan, err := ctr.StartAndAttach(getContext(), streams, detachKeys, resize, false)
+	if err != nil {
+		return err
+	}
+	select {
+	case attachChanErr := <-attachChan:
+		finalErr = attachChanErr
+	case chanError := <-errChan:
+		finalErr = chanError
+	}
+	return finalErr
+}
diff --git a/pkg/varlinkapi/containers.go b/pkg/varlinkapi/containers.go
index ac1352dac..17792ccfe 100644
--- a/pkg/varlinkapi/containers.go
+++ b/pkg/varlinkapi/containers.go
@@ -47,6 +47,55 @@ func (i *LibpodAPI) ListContainers(call iopodman.VarlinkCall) error {
 	return call.ReplyListContainers(listContainers)
 }
 
+func (i *LibpodAPI) Ps(call iopodman.VarlinkCall, opts iopodman.PsOpts) error {
+	var (
+		containers []iopodman.PsContainer
+	)
+	maxWorkers := shared.Parallelize("ps")
+	psOpts := makePsOpts(opts)
+	filters := []string{}
+	if opts.Filters != nil {
+		filters = *opts.Filters
+	}
+	psContainerOutputs, err := shared.GetPsContainerOutput(i.Runtime, psOpts, filters, maxWorkers)
+	if err != nil {
+		return call.ReplyErrorOccurred(err.Error())
+	}
+
+	for _, ctr := range psContainerOutputs {
+		container := iopodman.PsContainer{
+			Id:         ctr.ID,
+			Image:      ctr.Image,
+			Command:    ctr.Command,
+			Created:    ctr.Created,
+			Ports:      ctr.Ports,
+			Names:      ctr.Names,
+			IsInfra:    ctr.IsInfra,
+			Status:     ctr.Status,
+			State:      ctr.State.String(),
+			PidNum:     int64(ctr.Pid),
+			RootFsSize: ctr.Size.RootFsSize,
+			RwSize:     ctr.Size.RwSize,
+			Pod:        ctr.Pod,
+			CreatedAt:  ctr.CreatedAt.Format(time.RFC3339Nano),
+			ExitedAt:   ctr.ExitedAt.Format(time.RFC3339Nano),
+			StartedAt:  ctr.StartedAt.Format(time.RFC3339Nano),
+			Labels:     ctr.Labels,
+			NsPid:      ctr.PID,
+			Cgroup:     ctr.Cgroup,
+			Ipc:        ctr.Cgroup,
+			Mnt:        ctr.MNT,
+			Net:        ctr.NET,
+			PidNs:      ctr.PIDNS,
+			User:       ctr.User,
+			Uts:        ctr.UTS,
+			Mounts:     ctr.Mounts,
+		}
+		containers = append(containers, container)
+	}
+	return call.ReplyPs(containers)
+}
+
 // GetContainer ...
 func (i *LibpodAPI) GetContainer(call iopodman.VarlinkCall, id string) error {
 	ctr, err := i.Runtime.LookupContainer(id)
@@ -585,6 +634,22 @@ func (i *LibpodAPI) GetContainerStatsWithHistory(call iopodman.VarlinkCall, prev
 	return call.ReplyGetContainerStatsWithHistory(cStats)
 }
 
+// Spec ...
+func (i *LibpodAPI) Spec(call iopodman.VarlinkCall, name string) error {
+	ctr, err := i.Runtime.LookupContainer(name)
+	if err != nil {
+		return call.ReplyErrorOccurred(err.Error())
+	}
+
+	spec := ctr.Spec()
+	b, err := json.Marshal(spec)
+	if err != nil {
+		return call.ReplyErrorOccurred(err.Error())
+	}
+
+	return call.ReplySpec(string(b))
+}
+
 // GetContainersLogs is the varlink endpoint to obtain one or more container logs
 func (i *LibpodAPI) GetContainersLogs(call iopodman.VarlinkCall, names []string, follow, latest bool, since string, tail int64, timestamps bool) error {
 	var wg sync.WaitGroup
diff --git a/pkg/varlinkapi/util.go b/pkg/varlinkapi/util.go
index 3c4b9b79a..8716c963a 100644
--- a/pkg/varlinkapi/util.go
+++ b/pkg/varlinkapi/util.go
@@ -162,3 +162,36 @@ func stringPullPolicyToType(s string) buildah.PullPolicy {
 	}
 	return buildah.PullIfMissing
 }
+
+func derefBool(inBool *bool) bool {
+	if inBool == nil {
+		return false
+	}
+	return *inBool
+}
+
+func derefString(in *string) string {
+	if in == nil {
+		return ""
+	}
+	return *in
+}
+
+func makePsOpts(inOpts iopodman.PsOpts) shared.PsOptions {
+	last := 0
+	if inOpts.Last != nil {
+		lastT := *inOpts.Last
+		last = int(lastT)
+	}
+	return shared.PsOptions{
+		All:       inOpts.All,
+		Last:      last,
+		Latest:    derefBool(inOpts.Latest),
+		NoTrunc:   derefBool(inOpts.NoTrunc),
+		Pod:       derefBool(inOpts.Pod),
+		Size:      true,
+		Sort:      derefString(inOpts.Sort),
+		Namespace: true,
+		Sync:      derefBool(inOpts.Sync),
+	}
+}
diff --git a/pkg/varlinkapi/virtwriter/virtwriter.go b/pkg/varlinkapi/virtwriter/virtwriter.go
new file mode 100644
index 000000000..3adaf6e17
--- /dev/null
+++ b/pkg/varlinkapi/virtwriter/virtwriter.go
@@ -0,0 +1,155 @@
+package virtwriter
+
+import (
+	"bufio"
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+
+	"k8s.io/client-go/tools/remotecommand"
+)
+
+// SocketDest is the "key" to where IO should go on the varlink
+// multiplexed socket
+type SocketDest int
+
+const (
+	// ToStdout indicates traffic should go stdout
+	ToStdout SocketDest = iota
+	// ToStdin indicates traffic came from stdin
+	ToStdin SocketDest = iota
+	// ToStderr indicates traffuc should go to stderr
+	ToStderr SocketDest = iota
+	// TerminalResize indicates a terminal resize event has occurred
+	// and data should be passed to resizer
+	TerminalResize SocketDest = iota
+	// Quit and detach
+	Quit SocketDest = iota
+)
+
+// IntToSocketDest returns a socketdest based on integer input
+func IntToSocketDest(i int) SocketDest {
+	switch i {
+	case ToStdout.Int():
+		return ToStdout
+	case ToStderr.Int():
+		return ToStderr
+	case ToStdin.Int():
+		return ToStdin
+	case TerminalResize.Int():
+		return TerminalResize
+	case Quit.Int():
+		return Quit
+	default:
+		return ToStderr
+	}
+}
+
+// Int returns the integer representation of the socket dest
+func (sd SocketDest) Int() int {
+	return int(sd)
+}
+
+// VirtWriteCloser are writers for attach which include the dest
+// of the data
+type VirtWriteCloser struct {
+	writer *bufio.Writer
+	dest   SocketDest
+}
+
+// NewVirtWriteCloser is a constructor
+func NewVirtWriteCloser(w *bufio.Writer, dest SocketDest) VirtWriteCloser {
+	return VirtWriteCloser{w, dest}
+}
+
+// Close is a required method for a writecloser
+func (v VirtWriteCloser) Close() error {
+	return nil
+}
+
+// Write prepends a header to the input message.  The header is
+// 8bytes.  Position one contains the destination.  Positions
+// 5,6,7,8 are a big-endian encoded uint32 for len of the message.
+func (v VirtWriteCloser) Write(input []byte) (int, error) {
+	header := []byte{byte(v.dest), 0, 0, 0}
+	// Go makes us define the byte for big endian
+	mlen := make([]byte, 4)
+	binary.BigEndian.PutUint32(mlen, uint32(len(input)))
+	// append the message len to the header
+	msg := append(header, mlen...)
+	// append the message to the header
+	msg = append(msg, input...)
+	_, err := v.writer.Write(msg)
+	if err != nil {
+		return 0, err
+	}
+	err = v.writer.Flush()
+	return len(input), err
+}
+
+// Reader decodes the content that comes over the wire and directs it to the proper destination.
+func Reader(r *bufio.Reader, output, errput *os.File, input *io.PipeWriter, resize chan remotecommand.TerminalSize) error {
+	var saveb []byte
+	var eom int
+	for {
+		readb := make([]byte, 32*1024)
+		n, err := r.Read(readb)
+		// TODO, later may be worth checking in len of the read is 0
+		if err != nil {
+			return err
+		}
+		b := append(saveb, readb[0:n]...)
+		// no sense in reading less than the header len
+		for len(b) > 7 {
+			eom = int(binary.BigEndian.Uint32(b[4:8])) + 8
+			// The message and header are togther
+			if len(b) >= eom {
+				out := append([]byte{}, b[8:eom]...)
+
+				switch IntToSocketDest(int(b[0])) {
+				case ToStdout:
+					n, err := output.Write(out)
+					if err != nil {
+						return err
+					}
+					if n < len(out) {
+						return errors.New("short write error occurred on stdout")
+					}
+				case ToStderr:
+					n, err := errput.Write(out)
+					if err != nil {
+						return err
+					}
+					if n < len(out) {
+						return errors.New("short write error occurred on stderr")
+					}
+				case ToStdin:
+					n, err := input.Write(out)
+					if err != nil {
+						return err
+					}
+					if n < len(out) {
+						return errors.New("short write error occurred on stdin")
+					}
+				case TerminalResize:
+					// Resize events come over in bytes, need to be reserialized
+					resizeEvent := remotecommand.TerminalSize{}
+					if err := json.Unmarshal(out, &resizeEvent); err != nil {
+						return err
+					}
+					resize <- resizeEvent
+				case Quit:
+					return nil
+				}
+				b = b[eom:]
+			} else {
+				//	We do not have the header and full message, need to slurp again
+				saveb = b
+				break
+			}
+		}
+	}
+	return nil
+}