refactor

7 files changed, 243 insertions, 72 deletions

diff --git a/integrationtests/commandutils.go b/integrationtests/commandutils.go
new file mode 100644
index 0000000..d2f567f
--- /dev/null
+++ b/integrationtests/commandutils.go
@@ -0,0 +1,112 @@
+package integrationtests
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"sync"
+	"syscall"
+	"time"
+)
+
+// The exit code and the Go error of the command terminated.
+type exitPromise func() (int, error)
+
+func runCommand(ctx context.Context, stdoutFile, cmdStr string,
+	args ...string) (int, error) {
+
+	stdinCh, _, exit, err := startCommand(ctx, cmdStr, args...)
+	if err != nil {
+		return -1, err
+	}
+
+	fd, err := os.Create(stdoutFile)
+	if err != nil {
+		return -2, err
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	defer wg.Wait()
+
+	go func() {
+		defer fd.Close()
+		defer wg.Done()
+		for line := range stdinCh {
+			fd.WriteString(line)
+			fd.WriteString("\n")
+		}
+	}()
+
+	return exit()
+}
+
+func runCommandRetry(ctx context.Context, retries int, stdoutFile, cmd string,
+	args ...string) (exitCode int, err error) {
+
+	for i := 0; i < retries; i++ {
+		time.Sleep(time.Second)
+		if exitCode, err = runCommand(ctx, stdoutFile, cmd, args...); exitCode == 0 {
+			return
+		}
+	}
+	return
+}
+
+func startCommand(ctx context.Context, cmdStr string,
+	args ...string) (<-chan string, <-chan string, exitPromise, error) {
+
+	stdoutCh := make(chan string)
+	stderrCh := make(chan string)
+
+	if _, err := os.Stat(cmdStr); err != nil {
+		return stdoutCh, stderrCh, nil,
+			fmt.Errorf("no such executable '%s', please compile first: %v", cmdStr, err)
+	}
+
+	cmd := exec.CommandContext(ctx, cmdStr, args...)
+
+	cmdStdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return stdoutCh, stderrCh, nil, err
+	}
+	cmdStderr, err := cmd.StderrPipe()
+	err = cmd.Start()
+	if err != nil {
+		return stdoutCh, stderrCh, nil, err
+	}
+
+	go func() {
+		defer close(stdoutCh)
+		scanner := bufio.NewScanner(cmdStdout)
+		scanner.Split(bufio.ScanLines)
+		for scanner.Scan() {
+			stdoutCh <- scanner.Text()
+		}
+	}()
+	go func() {
+		close(stderrCh)
+		scanner := bufio.NewScanner(cmdStderr)
+		scanner.Split(bufio.ScanLines)
+		for scanner.Scan() {
+			stderrCh <- scanner.Text()
+		}
+	}()
+
+	return stdoutCh, stderrCh, func() (int, error) {
+		err := cmd.Wait()
+		return exitCodeFromError(err), err
+	}, nil
+}
+
+func exitCodeFromError(err error) int {
+	if err != nil {
+		if exitError, ok := err.(*exec.ExitError); ok {
+			ws := exitError.Sys().(syscall.WaitStatus)
+			return ws.ExitStatus()
+		}
+	}
+	return 0
+}
diff --git a/integrationtests/dcat_test.go b/integrationtests/dcat_test.go
index 342ebd0..e172bfa 100644
--- a/integrationtests/dcat_test.go
+++ b/integrationtests/dcat_test.go
@@ -1,6 +1,7 @@
 package integrationtests
 
 import (
+	"context"
 	"os"
 	"testing"
 )
@@ -8,15 +9,19 @@ import (
 func TestDCat(t *testing.T) {
 	testdataFile := "dcat.txt.expected"
 	stdoutFile := "dcat.out"
-	args := []string{"-spartan", testdataFile}
 
-	if _, err := runCommand(t, "../dcat", args, stdoutFile); err != nil {
+	_, err := runCommand(context.TODO(), stdoutFile,
+		"../dcat", "--spartan", testdataFile)
+
+	if err != nil {
 		t.Error(err)
 		return
 	}
+
 	if err := compareFiles(t, stdoutFile, testdataFile); err != nil {
 		t.Error(err)
 		return
 	}
+
 	os.Remove(stdoutFile)
 }
diff --git a/integrationtests/dgrep_test.go b/integrationtests/dgrep_test.go
index 4d54a2d..15519b3 100644
--- a/integrationtests/dgrep_test.go
+++ b/integrationtests/dgrep_test.go
@@ -1,6 +1,7 @@
 package integrationtests
 
 import (
+	"context"
 	"os"
 	"testing"
 )
@@ -9,16 +10,20 @@ func TestDGrep(t *testing.T) {
 	inFile := "mapr_testdata.log"
 	stdoutFile := "dgrep.stdout.tmp"
 	expectedStdoutFile := "dgrep.txt.expected"
-	args := []string{"-spartan", "--grep", "20211002-071947", inFile}
 
-	if _, err := runCommand(t, "../dgrep", args, stdoutFile); err != nil {
+	_, err := runCommand(context.TODO(), stdoutFile,
+		"../dgrep", "--spartan", "--grep", "20211002-071947", inFile)
+
+	if err != nil {
 		t.Error(err)
 		return
 	}
+
 	if err := compareFiles(t, stdoutFile, expectedStdoutFile); err != nil {
 		t.Error(err)
 		return
 	}
+
 	os.Remove(stdoutFile)
 }
 
@@ -26,15 +31,19 @@ func TestDGrep2(t *testing.T) {
 	inFile := "mapr_testdata.log"
 	stdoutFile := "dgrep2.stdout.tmp"
 	expectedStdoutFile := "dgrep2.txt.expected"
-	args := []string{"-spartan", "--grep", "20211002-071947", "--invert", inFile}
 
-	if _, err := runCommand(t, "../dgrep", args, stdoutFile); err != nil {
+	_, err := runCommand(context.TODO(), stdoutFile,
+		"../dgrep", "-spartan", "--grep", "20211002-071947", "--invert", inFile)
+
+	if err != nil {
 		t.Error(err)
 		return
 	}
+
 	if err := compareFiles(t, stdoutFile, expectedStdoutFile); err != nil {
 		t.Error(err)
 		return
 	}
+
 	os.Remove(stdoutFile)
 }
diff --git a/integrationtests/dmap_test.go b/integrationtests/dmap_test.go
index f5c78e0..966944a 100644
--- a/integrationtests/dmap_test.go
+++ b/integrationtests/dmap_test.go
@@ -1,6 +1,7 @@
 package integrationtests
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"testing"
@@ -17,12 +18,15 @@ func TestDMap(t *testing.T) {
 	query := fmt.Sprintf("from STATS select count($line),last($time),"+
 		"avg($goroutines),min(concurrentConnections),max(lifetimeConnections) "+
 		"group by $hostname outfile %s", csvFile)
-	args := []string{"-query", query, inFile}
 
-	if _, err := runCommand(t, "../dmap", args, stdoutFile); err != nil {
+	_, err := runCommand(context.TODO(), stdoutFile,
+		"../dmap", "--query", query, inFile)
+
+	if err != nil {
 		t.Error(err)
 		return
 	}
+
 	if err := compareFiles(t, csvFile, expectedCsvFile); err != nil {
 		t.Error(err)
 		return
@@ -49,11 +53,13 @@ func TestDMap2(t *testing.T) {
 		"avg($goroutines),min($goroutines) group by $time order by count($time) "+
 		"outfile %s", csvFile)
 
-	args := []string{"-query", query, inFile}
-	if _, err := runCommand(t, "../dmap", args, stdoutFile); err != nil {
+	_, err := runCommand(context.TODO(), stdoutFile,
+		"../dmap", "--query", query, inFile)
+	if err != nil {
 		t.Error(err)
 		return
 	}
+
 	if err := compareFilesContents(t, csvFile, expectedCsvFile); err != nil {
 		t.Error(err)
 		return
@@ -86,7 +92,7 @@ func TestDMap3(t *testing.T) {
 		args = append(args, inFile)
 	}
 
-	if _, err := runCommand(t, "../dmap", args, stdoutFile); err != nil {
+	if _, err := runCommand(context.TODO(), stdoutFile, "../dmap", args...); err != nil {
 		t.Error(err)
 		return
 	}
diff --git a/integrationtests/dtail_test.go b/integrationtests/dtail_test.go
index 36eadc0..267cd26 100644
--- a/integrationtests/dtail_test.go
+++ b/integrationtests/dtail_test.go
@@ -1,16 +1,102 @@
 package integrationtests
 
 import (
+	"context"
 	"os"
 	"testing"
 )
 
+func TestDTailWithServer(t *testing.T) {
+	followFile := "dtail.follow.tmp"
+	//serverStdoutFile := "dtail.dserver.stdout.tmp"
+	//greetings := []string{"world", "sol system", "milky way", "universe", "multiverse"}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	serverCh, _, _, err := startCommand(ctx,
+		"../dserver",
+		"--logger", "stdout",
+		"--logLevel", "info",
+		"--port", "4242",
+		"--relaxedAuth",
+	)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	clientCh, _, _, err := startCommand(ctx,
+		"../dtail",
+		"--logger", "stdout",
+		"--logLevel", "devel",
+		"--servers", "localhost:4242",
+		"--files", followFile,
+		"--grep", "Hello",
+		"--trustAllHosts",
+		"--noColor",
+	)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	for {
+		select {
+		case line := <-serverCh:
+			t.Log("server:", line)
+		case line := <-clientCh:
+			t.Log("client:", line)
+		case <-ctx.Done():
+			t.Log("Done reading client and server pipes")
+		}
+	}
+
+	/*
+		// Start dtail client, connect to the server and follow followFile.
+
+		//clientStdoutFile := "dtail.stdout.tmp"
+		/*
+
+			t.Log(clientArgs)
+			// TODO: Pipe with dtail command to read stdin stream.
+			//		runCommandContextRetry(ctx, t, "../dtail", clientArgs, clientStdoutFile)
+
+			// Write greetings to followFile
+			fd, err := os.Create(followFile)
+			if err != nil {
+				t.Error(err)
+			}
+			defer fd.Close()
+
+			go func() {
+				var circular int
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case <-time.After(time.Second):
+						fd.WriteString(time.Now().String())
+						fd.WriteString(fmt.Sprintf(" - Hello %s!\n", greetings[circular]))
+						circular = (circular + 1) % len(greetings)
+					}
+				}
+			}()
+	*/
+
+	/*
+		os.Remove(serverStdoutFile)
+		os.Remove(clientStdoutFile)
+		os.Remove(followFile)
+	*/
+}
+
 func TestDTailColorTable(t *testing.T) {
 	stdoutFile := "dtailcolortable.stdout.tmp"
 	expectedStdoutFile := "dtailcolortable.expected"
-	args := []string{"-colorTable"}
 
-	if _, err := runCommand(t, "../dtail", args, stdoutFile); err != nil {
+	_, err := runCommand(context.TODO(), stdoutFile, "../dtail", "--colorTable")
+	if err != nil {
 		t.Error(err)
 		return
 	}
diff --git a/integrationtests/dtailhealthcheck_test.go b/integrationtests/dtailhealthcheck_test.go
index d562239..a99bfdc 100644
--- a/integrationtests/dtailhealthcheck_test.go
+++ b/integrationtests/dtailhealthcheck_test.go
@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"os"
 	"testing"
-	"time"
 )
 
 func TestDTailHealthCheck(t *testing.T) {
@@ -13,7 +12,7 @@ func TestDTailHealthCheck(t *testing.T) {
 	expectedStdoutFile := "dtailhealthcheck.expected"
 
 	t.Log("Serverless check, is supposed to exit with warning state.")
-	exitCode, err := runCommand(t, "../dtailhealthcheck", []string{}, stdoutFile)
+	exitCode, err := runCommand(context.TODO(), stdoutFile, "../dtailhealthcheck")
 	if exitCode != 1 {
 		t.Error(fmt.Sprintf("Expected exit code '1' but got '%d': %v", exitCode, err))
 		return
@@ -23,17 +22,17 @@ func TestDTailHealthCheck(t *testing.T) {
 		t.Error(err)
 		return
 	}
-
 	os.Remove(stdoutFile)
 }
 
 func TestDTailHealthCheck2(t *testing.T) {
 	stdoutFile := "dtailhealthcheck2.stdout.tmp"
 	expectedStdoutFile := "dtailhealthcheck2.expected"
-	args := []string{"--server", "example:1"}
 
 	t.Log("Negative test, is supposed to exit with a critical state.")
-	exitCode, err := runCommand(t, "../dtailhealthcheck", args, stdoutFile)
+	exitCode, err := runCommand(context.TODO(), stdoutFile,
+		"../dtailhealthcheck", "--server", "example:1")
+
 	if exitCode != 2 {
 		t.Error(fmt.Sprintf("Expected exit code '2' but got '%d': %v", exitCode, err))
 		return
@@ -49,27 +48,20 @@ func TestDTailHealthCheck2(t *testing.T) {
 
 func TestDTailHealthCheck3(t *testing.T) {
 	stdoutFile := "dtailhealthcheck3.stdout.tmp"
-	serverStdoutFile := "dtailhealthcheck3.dserver.stdout.tmp"
 	expectedStdoutFile := "dtailhealthcheck3.expected"
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	go func() {
-		args := []string{"--logger", "stdout", "--logLevel", "trace", "--port", "4242"}
-		runCommandContext(ctx, t, "../dserver", args, serverStdoutFile)
-	}()
+	startCommand(ctx,
+		"../dserver",
+		"--logger", "stdout",
+		"--logLevel", "trace",
+		"--port", "4242",
+	)
 
-	var err error
-	args := []string{"--server", "localhost:4242"}
-	for i := 0; i < 30; i++ {
-		t.Log("Waiting for dserver to start", i)
-		time.Sleep(time.Second)
-		var exitCode int
-		if exitCode, err = runCommand(t, "../dtailhealthcheck", args, stdoutFile); exitCode == 0 {
-			break
-		}
-	}
+	_, err := runCommandRetry(ctx, 10, stdoutFile,
+		"../dtailhealthcheck", "--server", "localhost:4242")
 	if err != nil {
 		t.Error(err)
 		return
@@ -80,6 +72,5 @@ func TestDTailHealthCheck3(t *testing.T) {
 		return
 	}
 
-	os.Remove(serverStdoutFile)
 	os.Remove(stdoutFile)
 }
diff --git a/integrationtests/commons.go b/integrationtests/fileutils.go
index 2fdbfc3..d771607 100644
--- a/integrationtests/commons.go
+++ b/integrationtests/fileutils.go
@@ -2,53 +2,15 @@ package integrationtests
 
 import (
 	"bufio"
-	"context"
 	"crypto/sha256"
 	"encoding/base64"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"os/exec"
-	"strings"
-	"syscall"
 	"testing"
 )
 
-func runCommand(t *testing.T, cmd string, args []string, stdoutFile string) (int, error) {
-	return runCommandContext(context.TODO(), t, cmd, args, stdoutFile)
-}
-
-func runCommandContext(ctx context.Context, t *testing.T, cmd string, args []string,
-	stdoutFile string) (int, error) {
-
-	if _, err := os.Stat(cmd); err != nil {
-		return -1, fmt.Errorf("No such binary %s, please compile first (%v)", cmd, err)
-	}
-
-	t.Log("Running command:", cmd, strings.Join(args, " "))
-	bytes, cmdErr := exec.CommandContext(ctx, cmd, args...).Output()
-
-	t.Log("Writing stdout to file", stdoutFile)
-	fd, err := os.Create(stdoutFile)
-	if err != nil {
-		return -1, err
-	}
-	defer fd.Close()
-	fd.Write(bytes)
-
-	return exitCodeFromError(cmdErr), err
-}
-
-func exitCodeFromError(err error) int {
-	if err != nil {
-		if exitError, ok := err.(*exec.ExitError); ok {
-			ws := exitError.Sys().(syscall.WaitStatus)
-			return ws.ExitStatus()
-		}
-	}
-	return 0
-}
-
 // Checks whether both files have the same lines (order doesn't matter)
 func compareFilesContents(t *testing.T, fileA, fileB string) error {
 	mapFile := func(file string) (map[string]int, error) {