chore(helper): add Pipeline func

Signed-off-by: Bo-Yi Wu <[email protected]>

pkg/dbdump/mysql/mysql.go

@@ -39,7 +39,7 @@ func (d Dump) Exec() error {
 		return err
 	}
 
-	flags := []string{}
+	flags := []string{"mysqldump"}
 	if d.Name != "" {
 		flags = append(flags, "-d", d.Name)
 	}
@@ -64,17 +64,17 @@ func (d Dump) Exec() error {
 		flags = append(flags, d.Name)
 	}
 
-	// gzip > dump.sql.gz
+	// add gzip command
 	flags = append(flags, "|", "gzip", ">", "dump.sql.gz")
 
-	envs := []string{}
+	envs := os.Environ()
 	if d.Password != "" {
 		// See the MySQL Environment Variables
 		// ref: https://dev.mysql.com/doc/refman/8.0/en/environment-variables.html
 		envs = append(envs, fmt.Sprintf("MYSQL_PWD=%s", d.Password))
 	}
 
-	cmd = exec.Command("mysqldump", flags...)
+	cmd = exec.Command("bash", "-c", strings.Join(flags, " "))
 	cmd.Env = envs
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr

pkg/dbdump/postgres/postgres.go

@@ -39,7 +39,7 @@ func (d Dump) Exec() error {
 		return err
 	}
 
-	flags := []string{}
+	flags := []string{"pg_dump"}
 	host, port := getHostPort(d.Host)
 	if host != "" {
 		flags = append(flags, "-h", host)
@@ -60,15 +60,15 @@ func (d Dump) Exec() error {
 		flags = append(flags, d.Name)
 	}
 
-	// gzip > dump.sql.gz
+	// add gzip command
 	flags = append(flags, "|", "gzip", ">", "dump.sql.gz")
 
-	envs := []string{}
+	envs := os.Environ()
 	if d.Password != "" {
 		envs = append(envs, fmt.Sprintf("PGPASSWORD=%s", d.Password))
 	}
 
-	cmd = exec.Command("pg_dump", flags...)
+	cmd = exec.Command("bash", "-c", strings.Join(flags, " "))
 	cmd.Env = envs
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr

pkg/helper/cmd.go

@@ -0,0 +1,56 @@
+package helper
+
+import (
+	"bytes"
+	"os/exec"
+)
+
+// Pipeline strings together the given exec.Cmd commands in a similar fashion
+// to the Unix pipeline.  Each command's standard output is connected to the
+// standard input of the next command, and the output of the final command in
+// the pipeline is returned, along with the collected standard error of all
+// commands and the first error found (if any).
+//
+// To provide input to the pipeline, assign an io.Reader to the first's Stdin.
+// ref: https://gist.github.com/kylelemons/1525278
+func Pipeline(cmds ...*exec.Cmd) (pipeLineOutput, collectedStandardError []byte, pipeLineError error) {
+	// Require at least one command
+	if len(cmds) < 1 {
+		return nil, nil, nil
+	}
+
+	// Collect the output from the command(s)
+	var output bytes.Buffer
+	var stderr bytes.Buffer
+
+	last := len(cmds) - 1
+	for i, cmd := range cmds[:last] {
+		var err error
+		// Connect each command's stdin to the previous command's stdout
+		if cmds[i+1].Stdin, err = cmd.StdoutPipe(); err != nil {
+			return nil, nil, err
+		}
+		// Connect each command's stderr to a buffer
+		cmd.Stderr = &stderr
+	}
+
+	// Connect the output and error for the last command
+	cmds[last].Stdout, cmds[last].Stderr = &output, &stderr
+
+	// Start each command
+	for _, cmd := range cmds {
+		if err := cmd.Start(); err != nil {
+			return output.Bytes(), stderr.Bytes(), err
+		}
+	}
+
+	// Wait for each command to complete
+	for _, cmd := range cmds {
+		if err := cmd.Wait(); err != nil {
+			return output.Bytes(), stderr.Bytes(), err
+		}
+	}
+
+	// Return the pipeline output and the collected standard error
+	return output.Bytes(), stderr.Bytes(), nil
+}