130 lines
2.6 KiB
Go
130 lines
2.6 KiB
Go
package core
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"log"
|
|
"os"
|
|
"os/exec"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
func readStdout(stdout io.ReadCloser, items chan Item, cparse chan bool) {
|
|
var item Item
|
|
parseError := false
|
|
scanout := bufio.NewScanner(stdout)
|
|
for scanout.Scan() {
|
|
data := scanout.Bytes()
|
|
err := json.Unmarshal(data, &item)
|
|
if err != nil {
|
|
log.Printf("[stdout] %s\n", strings.TrimSpace(string(data)))
|
|
parseError = true
|
|
} else {
|
|
log.Printf("[item] %s\n", item.Id)
|
|
items <- item
|
|
}
|
|
}
|
|
// Only send the parsing result at the end, to block main until stdout is drained
|
|
cparse <- parseError
|
|
close(items)
|
|
}
|
|
|
|
func readStderr(stderr io.ReadCloser, done chan bool) {
|
|
scanerr := bufio.NewScanner(stderr)
|
|
for scanerr.Scan() {
|
|
text := strings.TrimSpace(scanerr.Text())
|
|
log.Printf("[stderr] %s\n", text)
|
|
}
|
|
done <- true
|
|
}
|
|
|
|
func writeStdin(stdin io.WriteCloser, text string) {
|
|
defer stdin.Close()
|
|
io.WriteString(stdin, text)
|
|
}
|
|
|
|
func Execute(
|
|
argv []string,
|
|
env []string,
|
|
input string,
|
|
timeout time.Duration,
|
|
) ([]Item, error) {
|
|
log.Printf("Executing %v", argv)
|
|
|
|
if len(argv) == 0 {
|
|
return nil, errors.New("error: empty argv")
|
|
}
|
|
|
|
env = append(env, "STATE_PATH=")
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
defer cancel()
|
|
cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
|
|
cmd.Env = append(os.Environ(), env...)
|
|
cmd.WaitDelay = time.Second * 5
|
|
|
|
// Open pipes to the command
|
|
stdin, err := cmd.StdinPipe()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
stdout, err := cmd.StdoutPipe()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
stderr, err := cmd.StderrPipe()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cout := make(chan Item)
|
|
cparse := make(chan bool)
|
|
cerr := make(chan bool)
|
|
|
|
// Sink routine for items produced
|
|
var items []Item
|
|
go func() {
|
|
for item := range cout {
|
|
items = append(items, item)
|
|
}
|
|
}()
|
|
|
|
// Routines handling the process i/o
|
|
go writeStdin(stdin, input)
|
|
go readStdout(stdout, cout, cparse)
|
|
go readStderr(stderr, cerr)
|
|
|
|
// Kick off the command
|
|
err = cmd.Start()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Block until std{out,err} close
|
|
<-cerr
|
|
parseError := <-cparse
|
|
|
|
err = cmd.Wait()
|
|
if ctx.Err() == context.DeadlineExceeded {
|
|
log.Printf("Timed out after %v\n", timeout)
|
|
return nil, err
|
|
} else if exiterr, ok := err.(*exec.ExitError); ok {
|
|
log.Printf("error: %s failed with exit code %d\n", argv[0], exiterr.ExitCode())
|
|
return nil, err
|
|
} else if err != nil {
|
|
log.Printf("error: %s failed with error: %s\n", argv[0], err)
|
|
return nil, err
|
|
}
|
|
|
|
if parseError {
|
|
log.Printf("error: could not parse item\n")
|
|
return nil, errors.New("invalid JSON")
|
|
}
|
|
|
|
return items, nil
|
|
}
|