Generalize consumeSlow and add stop support

Signed-off-by: Tõnis Tiigi <tonistiigi@gmail.com> (github: tonistiigi)

integration-cli/docker_cli_run_test.go

@@ -4,7 +4,6 @@ import (
 	"bufio"
 	"bytes"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"net"
 	"os"
@@ -2462,7 +2461,7 @@ func TestRunSlowStdoutConsumer(t *testing.T) {
 	if err := c.Start(); err != nil {
 		t.Fatal(err)
 	}
-	n, err := consumeSlow(stdout, 10000, 5*time.Millisecond)
+	n, err := consumeWithSpeed(stdout, 10000, 5*time.Millisecond, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

integration-cli/utils.go

@@ -254,18 +254,25 @@ func makeRandomString(n int) string {
 	return string(b)
 }
 
-func consumeSlow(reader io.Reader, chunkSize int, interval time.Duration) (n int, err error) {
+// Reads chunkSize bytes from reader after every interval.
+// Returns total read bytes.
+func consumeWithSpeed(reader io.Reader, chunkSize int, interval time.Duration, stop chan bool) (n int, err error) {
 	buffer := make([]byte, chunkSize)
 	for {
-		var readBytes int
+		select {
-		readBytes, err = reader.Read(buffer)
+		case <-stop:
-		n += readBytes
-		if err != nil {
-			if err == io.EOF {
-				err = nil
-			}
 			return
+		default:
+			var readBytes int
+			readBytes, err = reader.Read(buffer)
+			n += readBytes
+			if err != nil {
+				if err == io.EOF {
+					err = nil
+				}
+				return
+			}
+			time.Sleep(interval)
 		}
-		time.Sleep(interval)
 	}
 }