func (a *Agent) remoteExecWriteAck(event *remoteExecEvent) bool {
	if err := a.remoteExecWriteKey(event, remoteExecAckSuffix, nil); err != nil {
		a.logger.Printf("[ERR] agent: failed to ack remote exec job: %v", err)
		return false
	}
	return true
}
remoteExecWriteAck is used to write an ack. Returns if execution should
continue.
package agent
import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"os/exec"
	"path"
	"strconv"
	"sync"
	"syscall"
	"time"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/consul/consul/structs"
)
const (
	remoteExecFileName = "job"
	remoteExecAckSuffix = "ack"
	remoteExecExitSuffix = "exit"
	remoteExecOutputDivider = "out"
	remoteExecOutputSize = 4 * 1024
	remoteExecOutputDeadline = 500 * time.Millisecond
)
type remoteExecEvent struct {
	Prefix  string
	Session string
}
type remoteExecSpec struct {
	Command string
	Script  []byte
	Wait    time.Duration
}
type rexecWriter struct {
	BufCh    chan []byte
	BufSize  int
	BufIdle  time.Duration
	CancelCh chan struct{}
	buf     []byte
	bufLen  int
	bufLock sync.Mutex
	flush   *time.Timer
}
func (r *rexecWriter) Write(b []byte) (int, error) {
	r.bufLock.Lock()
	defer r.bufLock.Unlock()
	if r.flush != nil {
		r.flush.Stop()
		r.flush = nil
	}
	inpLen := len(b)
	if r.buf == nil {
		r.buf = make([]byte, r.BufSize)
	}
COPY:
	remain := len(r.buf) - r.bufLen
	if remain > len(b) {
		copy(r.buf[r.bufLen:], b)
		r.bufLen += len(b)
	} else {
		copy(r.buf[r.bufLen:], b[:remain])
		b = b[remain:]
		r.bufLen += remain
		r.bufLock.Unlock()
		r.Flush()
		r.bufLock.Lock()
		goto COPY
	}
	r.flush = time.AfterFunc(r.BufIdle, r.Flush)
	return inpLen, nil
}
func (r *rexecWriter) Flush() {
	r.bufLock.Lock()
	defer r.bufLock.Unlock()
	if r.flush != nil {
		r.flush.Stop()
		r.flush = nil
	}
	if r.bufLen == 0 {
		return
	}
	select {
	case r.BufCh <- r.buf[:r.bufLen]:
		r.buf = make([]byte, r.BufSize)
		r.bufLen = 0
	case <-r.CancelCh:
		r.bufLen = 0
	}
}
func (a *Agent) handleRemoteExec(msg *UserEvent) {
	a.logger.Printf("[DEBUG] agent: received remote exec event (ID: %s)", msg.ID)
	var event remoteExecEvent
	if err := json.Unmarshal(msg.Payload, &event); err != nil {
		a.logger.Printf("[ERR] agent: failed to decode remote exec event: %v", err)
		return
	}
	var spec remoteExecSpec
	if !a.remoteExecGetSpec(&event, &spec) {
		return
	}
	if !a.remoteExecWriteAck(&event) {
		return
	}
	exitCode := 0
	defer a.remoteExecWriteExitCode(&event, &exitCode)
	var script string
	if len(spec.Script) != 0 {
		tmpFile, err := ioutil.TempFile("", "rexec")
		if err != nil {
			a.logger.Printf("[DEBUG] agent: failed to make tmp file: %v", err)
			exitCode = 255
			return
		}
		defer os.Remove(tmpFile.Name())
		os.Chmod(tmpFile.Name(), 0750)
		tmpFile.Write(spec.Script)
		tmpFile.Close()
		script = tmpFile.Name()
	} else {
		script = spec.Command
	}
	a.logger.Printf("[INFO] agent: remote exec '%s'", script)
	cmd, err := ExecScript(script)
	if err != nil {
		a.logger.Printf("[DEBUG] agent: failed to start remote exec: %v", err)
		exitCode = 255
		return
	}
	writer := &rexecWriter{
		BufCh:    make(chan []byte, 16),
		BufSize:  remoteExecOutputSize,
		BufIdle:  remoteExecOutputDeadline,
		CancelCh: make(chan struct{}),
	}
	cmd.Stdout = writer
	cmd.Stderr = writer
	err = cmd.Start()
	if err != nil {
		a.logger.Printf("[DEBUG] agent: failed to start remote exec: %v", err)
		exitCode = 255
		return
	}
	exitCh := make(chan int, 1)
	go func() {
		err := cmd.Wait()
		writer.Flush()
		close(writer.BufCh)
		if err == nil {
			exitCh <- 0
			return
		}
		if exitErr, ok := err.(*exec.ExitError); ok {
			if status, ok := exitErr.Sys().(syscall.WaitStatus); ok {
				exitCh <- status.ExitStatus()
				return
			}
		}
		exitCh <- 1
	}()
WAIT:
	for num := 0; ; num++ {
		select {
		case out := <-writer.BufCh:
			if out == nil {
				break WAIT
			}
			if !a.remoteExecWriteOutput(&event, num, out) {
				close(writer.CancelCh)
				exitCode = 255
				return
			}
		case <-time.After(spec.Wait):
			if !a.remoteExecWriteOutput(&event, num, nil) {
				close(writer.CancelCh)
				exitCode = 255
				return
			}
		}
	}
	exitCode = <-exitCh
}
func (a *Agent) remoteExecGetSpec(event *remoteExecEvent, spec *remoteExecSpec) bool {
	get := structs.KeyRequest{
		Datacenter: a.config.Datacenter,
		Key:        path.Join(event.Prefix, event.Session, remoteExecFileName),
		QueryOptions: structs.QueryOptions{
			AllowStale: true,  
		},
	}
	var out structs.IndexedDirEntries
QUERY:
	if err := a.RPC("KVS.Get", &get, &out); err != nil {
		a.logger.Printf("[ERR] agent: failed to get remote exec job: %v", err)
		return false
	}
	if len(out.Entries) == 0 {
		if get.QueryOptions.AllowStale {
			a.logger.Printf("[DEBUG] agent: trying consistent fetch of remote exec job spec")
			get.QueryOptions.AllowStale = false
			goto QUERY
		} else {
			a.logger.Printf("[DEBUG] agent: remote exec aborted, job spec missing")
			return false
		}
	}
	if err := json.Unmarshal(out.Entries[0].Value, &spec); err != nil {
		a.logger.Printf("[ERR] agent: failed to decode remote exec spec: %v", err)
		return false
	}
	return true
}