Add asyncexec package to simplify os/exec output streaming (#15)

Issue N/A

Description of changes:
- Add `asyncexec` package to simplify `os/exec.Command` output streaming 

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Author: a-hilaly

Makefile

@@ -17,7 +17,7 @@ install: build
 	cp ./bin/ackdev $(shell go env GOPATH)/bin/ackdev
 
 test:
-	go test -v ./...
+	go test -tags $(shell go env GOOS) -v ./...
 
 .PHONY: test install mocks
 

pkg/asyncexec/cmd.go

@@ -0,0 +1,103 @@
+package asyncexec
+
+import (
+	"bufio"
+	"os/exec"
+)
+
+// New instantiate a new Cmd object.
+func New(cmd *exec.Cmd, buff int) *Cmd {
+	return &Cmd{
+		cmd:      cmd,
+		stopCh:   make(chan struct{}),
+		stdoutCh: make(chan []byte, buff),
+		stderrCh: make(chan []byte, buff),
+	}
+}
+
+// Cmd is a wrapper arround exec.Cmd. Mainly used to execute
+// command asynchronously with/or without output stream.
+type Cmd struct {
+	cmd *exec.Cmd
+
+	stopCh   chan struct{}
+	stdoutCh chan []byte
+	stderrCh chan []byte
+}
+
+// Run runs the command. if streamOutput is true, it will spin
+// two goroutine responsible of streaming the stdout and stderr
+func (c *Cmd) Run() error {
+	cmdStdoutReader, err := c.cmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+	stdoutScanner := bufio.NewScanner(cmdStdoutReader)
+
+	cmdStderrReader, err := c.cmd.StderrPipe()
+	if err != nil {
+		return err
+	}
+	stderrScanner := bufio.NewScanner(cmdStderrReader)
+
+	// Goroutine for stdout
+	go func() {
+		defer close(c.stdoutCh)
+		for stdoutScanner.Scan() {
+			bytes := stdoutScanner.Bytes()
+			c.stdoutCh <- bytes
+		}
+	}()
+
+	// Goroutine for stderr
+	go func() {
+		defer close(c.stderrCh)
+		for stderrScanner.Scan() {
+			bytes := stderrScanner.Bytes()
+			c.stderrCh <- bytes
+		}
+	}()
+
+	err = c.cmd.Start()
+	if err != nil {
+		return err
+	}
+
+	// listening for stop signal
+	go func() {
+		<-c.stopCh
+		c.cmd.Process.Kill()
+	}()
+
+	return nil
+}
+
+// Exited returns true if the command exited, false otherwise.
+func (c *Cmd) Exited() bool {
+	return c.cmd.ProcessState.Exited()
+}
+
+// ExitCode returns the command process exit code.
+func (c *Cmd) ExitCode() int {
+	return c.cmd.ProcessState.ExitCode()
+}
+
+// StdoutStream returns a channel streaming the command Stdout.
+func (c *Cmd) StdoutStream() <-chan []byte {
+	return c.stdoutCh
+}
+
+// StderrStream returns a channel streaming the command Stderr.
+func (c *Cmd) StderrStream() <-chan []byte {
+	return c.stderrCh
+}
+
+// Wait blocks until the command exits
+func (c *Cmd) Wait() error {
+	return c.cmd.Wait()
+}
+
+// Stop signals the Wrapper to kill the process running the command.
+func (c *Cmd) Stop() {
+	c.stopCh <- struct{}{}
+}

pkg/asyncexec/cmd_test.go

@@ -0,0 +1,44 @@
+// +build !windows
+
+package asyncexec_test
+
+import (
+	"fmt"
+	"os/exec"
+
+	"github.com/aws-controllers-k8s/dev-tools/pkg/asyncexec"
+)
+
+func ExampleCmd_Run_withNoStream() {
+	cmd := asyncexec.New(exec.Command("echo", "Hello ACK"), 16)
+	cmd.Run()
+	cmd.Wait()
+	fmt.Println(cmd.ExitCode())
+	// Output: 0
+}
+
+func ExampleCmd_Run_withStream() {
+	cmd := asyncexec.New(exec.Command("echo", "Hello ACK"), 16)
+	cmd.Run()
+
+	done := make(chan struct{})
+	go func() {
+		for b := range cmd.StdoutStream() {
+			fmt.Println(string(b))
+
+		}
+		done <- struct{}{}
+	}()
+	go func() {
+		for b := range cmd.StderrStream() {
+			fmt.Println(string(b))
+
+		}
+		done <- struct{}{}
+	}()
+
+	defer func() { _, _ = <-done, <-done }()
+
+	cmd.Wait()
+	// Output: Hello ACK
+}

pkg/asyncexec/stream.go

@@ -0,0 +1,62 @@
+package asyncexec
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+)
+
+// StreamCommand executes a given command in a context directory and streams
+// the outputs to their according stdeout/stderr.
+func StreamCommand(workDir string, command string, args []string) error {
+	cmd := exec.Command(command, args...)
+	if workDir != "" {
+		cmd.Dir = workDir
+	}
+
+	acmd := New(cmd, 8)
+	err := acmd.Run()
+	if err != nil {
+		return err
+	}
+
+	// done is used to wait for stream readers to finish before exiting the function.
+	done := make(chan struct{})
+
+	go func() {
+		for b := range acmd.StdoutStream() {
+			_, err := os.Stdout.Write(b)
+			if err != nil {
+				msg := fmt.Sprintf("failed to write to Stdout: %v", err)
+				// should never happen, just panic.
+				panic(msg)
+			}
+		}
+		done <- struct{}{}
+	}()
+	go func() {
+		for b := range acmd.StderrStream() {
+			_, err := os.Stderr.Write(b)
+			if err != nil {
+				msg := fmt.Sprintf("failed to write to Stderr: %v", err)
+				// should never happen, just panic.
+				panic(msg)
+			}
+		}
+		done <- struct{}{}
+	}()
+
+	// wait for printers to finish
+	defer func() { _, _ = <-done, <-done }()
+
+	// wait for command to finish
+	err = acmd.Wait()
+	if err != nil {
+		return err
+	}
+
+	if acmd.ExitCode() != 0 {
+		return fmt.Errorf("exited with code %d", acmd.ExitCode())
+	}
+	return nil
+}