Using command stdin/out/err to pipe stuff from and to Go.

cmd-pipe.go

package main

import (
	"encoding/json"
	"errors"
	"io"
	"io/ioutil"
	"log"
	"os"
	"os/exec"
	"sync"
	"time"
)

type InputData struct {
	Text   string
	Number int
}

type OutputData struct {
	A string
	B int
}

func main() {
	filename, err := tempFile(`
		cat - > stdin.log
		echo {\"A\":\"foobar\", \"B\":13}
		#echo {\"A\":\"foobar\", \"B\":13} 1>&2
	`)
	if err != nil {
		log.Fatal(err)
	}
	in, out := &InputData{Text: "hello world", Number: 1e8}, &OutputData{}
	err = runCommand(2*time.Second, in, &out, "sh", filename)
	os.Remove(filename)
	if err != nil {
		log.Fatal("stderr:", err)
	}
	log.Printf("in:%#v out:%#v", in, out)
	log.Println("See stdin.log")
}

func tempFile(content string) (string, error) {
	f, err := ioutil.TempFile(".", "_test")
	if err != nil {
		return "", err
	}
	defer f.Close()
	if _, err := f.WriteString(content); err != nil {
		return "", err
	}
	return f.Name(), nil
}

// runCommand executes a command piping JSON to its stdin and stdout. If the
// command writes to stderr, the execution is aborted after stderr is closed
// and the error is returned.
func runCommand(timeout time.Duration, in interface{}, out interface{}, command string, arg ...string) error {
	p := &pipe{
		cmd:  exec.Command(command, arg...),
		out:  out,
		errc: make(chan error, 3), // in,out,err
	}
	fin, err := p.cmd.StdinPipe()
	if err != nil {
		return err
	}
	defer fin.Close()
	fout, err := p.cmd.StdoutPipe()
	if err != nil {
		return err
	}
	defer fout.Close()
	ferr, err := p.cmd.StderrPipe()
	if err != nil {
		return err
	}
	defer ferr.Close()
	if err := p.cmd.Start(); err != nil {
		return err
	}
	outc := make(chan interface{}, 1)
	go p.writeStdin(fin, in)
	go p.readStdout(fout, outc)
	go p.readStderr(ferr)
	p.rw.Add(3) // the 3 above
	go p.waitFinish()
	select {
	case out = <-outc:
		return nil
	case err := <-p.errc:
		return err
	case <-time.After(timeout):
		return errors.New("Timed out running command")
	}
}

type pipe struct {
	cmd  *exec.Cmd
	out  interface{}
	errc chan error
	rw   sync.WaitGroup // wait for stdin, stdout and stderr to finish
}

func (p *pipe) writeStdin(f io.WriteCloser, data interface{}) {
	if err := json.NewEncoder(f).Encode(data); err != nil {
		p.errc <- err
	}
	f.Close()
	p.rw.Done()
}

func (p *pipe) readStdout(f io.Reader, outc chan interface{}) {
	if err := json.NewDecoder(f).Decode(p.out); err != nil {
		p.errc <- err
	} else {
		outc <- p.out
	}
	p.rw.Done()
}

func (p *pipe) readStderr(f io.Reader) {
	if m, err := ioutil.ReadAll(f); err != nil {
		p.errc <- err
	} else if len(m) > 0 {
		p.errc <- errors.New(string(m))
	}
	p.rw.Done()
}

func (p *pipe) waitFinish() {
	p.rw.Wait()
	p.cmd.Wait()
}