Made build work again, partially refactored controller/agent.

This commit adds some one-shot rsync functions that can be used until we
can get the endpoint refactored to re-use rsync engines.

It also removes our multiplexing package - this is probably not the
right way to go.

Finally, it adds a few cleanups to agent and controller code, but much
more awaits.

.travis.yml

@@ -35,7 +35,6 @@ install:
 #   go test -v -race -cover ./...
 script:
   - go version
-  - go test -v -race -cover github.com/havoc-io/mutagen/multiplex
   - go test -v -race -cover github.com/havoc-io/mutagen/rsync
   - go test -v -race -cover github.com/havoc-io/mutagen/sync
   - go test -v -race -cover github.com/havoc-io/mutagen/url

agent/ssh.go

@@ -211,7 +211,7 @@ func connectSSH(remote *url.URL, prompter, mode string) (io.ReadWriteCloser, boo
 	}
 
 	// Create a stream that wrap's the process' standard input/output.
-	stream, err := newProcessStream(process)
+	stream, err := newAgentStream(process)
 	if err != nil {
 		return nil, false, errors.Wrap(err, "unable to create SSH process stream")
 	}

agent/stream.go

@@ -5,18 +5,37 @@ import (
 	"os/exec"
 
 	"github.com/pkg/errors"
-
-	"github.com/havoc-io/mutagen/stream"
 )
 
-// processStream implements io.ReadWriteCloser around the standard input/output
-// of a process.
-type processStream struct {
+// agentStream implements io.ReadWriteCloser around the standard input/output of
+// an agent process.
+type agentStream struct {
 	process *exec.Cmd
-	io.ReadWriteCloser
+	io.Reader
+	io.Writer
+}
+
+// Close closes the agent stream.
+// HACK: Rather than closing the process' standard input/output, this method
+// simply terminates the agent process. The problem with closing the
+// input/output streams is that they'll be OS pipes that might be blocked in
+// reads or writes and won't necessarily unblock if closed, and they might even
+// block the close - it's all platform dependent. But terminating the process
+// will close the remote ends of the pipes and thus unblocks and reads/writes.
+func (s *agentStream) Close() error {
+	// HACK: Accessing the Process field of an os/exec.Cmd could be a bit
+	// dangerous if other code was accessing the Cmd at the same time, but in
+	// our case the Cmd becomes completely encapsulated inside agentStream
+	// before agentStream is returned, so it's okay.
+	if s.process.Process != nil {
+		if err := s.process.Process.Kill(); err != nil {
+			return errors.Wrap(err, "unable to kill underlying process")
+		}
+	}
+	return nil
 }
 
-func newProcessStream(process *exec.Cmd) (io.ReadWriteCloser, error) {
+func newAgentStream(process *exec.Cmd) (io.ReadWriteCloser, error) {
 	// Redirect the process' standard input.
 	standardInput, err := process.StdinPipe()
 	if err != nil {
@@ -30,8 +49,5 @@ func newProcessStream(process *exec.Cmd) (io.ReadWriteCloser, error) {
 	}
 
 	// Create the result.
-	return &processStream{
-		process,
-		stream.New(standardOutput, standardInput, standardInput),
-	}, nil
+	return &agentStream{process, standardOutput, standardInput}, nil
 }

appveyor.yml

@@ -29,7 +29,6 @@ before_test:
 #   go test -v %RACEFLAG% -cover ./...
 test_script:
   - go version
-  - go test -v %RACEFLAG% -cover github.com/havoc-io/mutagen/multiplex
   - go test -v %RACEFLAG% -cover github.com/havoc-io/mutagen/rsync
   - go test -v %RACEFLAG% -cover github.com/havoc-io/mutagen/sync
   - go test -v %RACEFLAG% -cover github.com/havoc-io/mutagen/url

cmd/mutagen-agent/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"io"
 	"os"
 	"os/signal"
 
@@ -10,12 +11,24 @@ import (
 	"github.com/havoc-io/mutagen/agent"
 	"github.com/havoc-io/mutagen/cmd"
 	"github.com/havoc-io/mutagen/session"
-	"github.com/havoc-io/mutagen/stream"
 )
 
 var agentUsage = `usage: mutagen-agent should not be manually invoked
 `
 
+type stdio struct {
+	io.Reader
+	io.Writer
+}
+
+func (s *stdio) Close() error {
+	// HACK: We can't really close standard input/output pipes because doing so
+	// won't necessarily unblock and reads/writes and might also block the
+	// close. Fortunately, we don't need to support this in the agent - the
+	// streams should have the same lifetime as the process.
+	panic("standard input/output closed in agent")
+}
+
 func main() {
 	// Parse flags.
 	flagSet := cmd.NewFlagSet("mutagen-agent", agentUsage, []int{1})
@@ -33,7 +46,7 @@ func main() {
 	agent.Housekeep()
 
 	// Create a stream on standard input/output.
-	stdio := stream.New(os.Stdin, os.Stdout, os.Stdout)
+	stdio := &stdio{os.Stdin, os.Stdout}
 
 	// Perform a handshake.
 	if err := mutagen.SendVersion(stdio); err != nil {