Merge branch 'master' into pr/308

When Gor deals with multiple input files, it sorts them and replay their requests file after file.
This behavior is not convenient and should be changed. We should respect timestamps order when replaying requests.

input_file.go

@@ -9,31 +9,40 @@ import (
 	"log"
 	"os"
 	"path/filepath"
-	"sort"
-	"strconv"
 	"strings"
+	"strconv"
 	"time"
 )
 
 // FileInput can read requests generated by FileOutput
 type FileInput struct {
-	data          chan []byte
+	data             chan []byte
-	path          string
+	exit             chan bool
-	currentFile   *os.File
+	path             string
-	currentReader *bufio.Reader
+	fileInputReaders []*fileInputReader
-	speedFactor   float64
+	speedFactor      float64
-	loop          bool
+	loop             bool
+}
+
+
+type fileInputReader struct {
+	reader    *bufio.Reader
+	meta      [][]byte
+	data      []byte
+	file      *os.File
+	timestamp int64
 }
 
 // NewFileInput constructor for FileInput. Accepts file path as argument.
 func NewFileInput(path string, loop bool) (i *FileInput) {
 	i = new(FileInput)
 	i.data = make(chan []byte)
+	i.exit = make(chan bool)
 	i.path = path
 	i.speedFactor = 1
 	i.loop = loop
 
-	if err := i.updateFile(); err != nil {
+	if err := i.init(); err != nil {
 		return
 	}
 
@@ -48,9 +57,7 @@ func (_ *NextFileNotFound) Error() string {
 	return "There is no new files"
 }
 
-// path can be a pattern
+func (i *FileInput) init() (err error) {
-// It sort paths lexicographically and tries to choose next one
-func (i *FileInput) updateFile() (err error) {
 	var matches []string
 
 	if matches, err = filepath.Glob(i.path); err != nil {
@@ -63,40 +70,24 @@ func (i *FileInput) updateFile() (err error) {
 		return errors.New("No matching files")
 	}
 
-	sort.Sort(sortByFileIndex(matches))
+	i.fileInputReaders = make([]*fileInputReader, len(matches))
 
-	// Just pick first file, if there is many, and we are just started
+	for idx, p := range matches {
-	if i.currentFile == nil {
+		file, _ := os.Open(p)
-		if i.currentFile, err = os.Open(matches[0]); err != nil {
+		fileInputReader := &fileInputReader{}
-			log.Println("Can't read file ", matches[0], err)
+		fileInputReader.file = file
-			return
+		if strings.HasSuffix(p, ".gz") {
-		}
+			gzReader, err := gzip.NewReader(file)
-	} else {
+			if err != nil {
-		found := false
+				log.Fatal(err)
-		for idx, p := range matches {
-			if p == i.currentFile.Name() && idx != len(matches)-1 {
-				if i.currentFile, err = os.Open(matches[idx+1]); err != nil {
-					log.Println("Can't read file ", matches[idx+1], err)
-					return
-				}
-
-				found = true
 			}
+			fileInputReader.reader = bufio.NewReader(gzReader)
+		} else {
+			fileInputReader.reader = bufio.NewReader(file)
 		}
 
-		if !found {
+		fileInputReader.readNextInput()
-			return new(NextFileNotFound)
+		i.fileInputReaders[idx] = fileInputReader
-		}
-	}
-
-	if strings.HasSuffix(i.currentFile.Name(), ".gz") {
-		gzReader, err := gzip.NewReader(i.currentFile)
-		if err != nil {
-			log.Fatal(err)
-		}
-		i.currentReader = bufio.NewReader(gzReader)
-	} else {
-		i.currentReader = bufio.NewReader(i.currentFile)
 	}
 
 	return nil
@@ -113,83 +104,123 @@ func (i *FileInput) String() string {
 	return "File input: " + i.path
 }
 
-func (i *FileInput) emit() {
+func (f *fileInputReader) readNextInput() {
-	var lastTime int64
+	nextInput := f.nextInput()
+	f.parseNextInput(nextInput)
+}
 
+func (f *fileInputReader) parseNextInput(input []byte) {
+	if (input != nil) {
+		f.meta = payloadMeta(input)
+		f.timestamp, _ = strconv.ParseInt(string(f.meta[2]), 10, 64)
+		f.data = input
+	}
+}
+
+func (f *fileInputReader) nextInput() []byte {
 	payloadSeparatorAsBytes := []byte(payloadSeparator)
-
 	var buffer bytes.Buffer
 
-	if i.currentReader == nil {
-		return
-	}
-
 	for {
-		line, err := i.currentReader.ReadBytes('\n')
+		line, err := f.reader.ReadBytes('\n')
 
 		if err != nil {
 			if err != io.EOF {
 				log.Fatal(err)
 			}
 
-			// If our path pattern match multiple files, try to find them
 			if err == io.EOF {
-				if e := i.updateFile(); e != nil {
+				f.file.Close()
-					if _, ok := e.(*NextFileNotFound); ok && i.loop {
+				f.file = nil
-						// Start from the first file
+				return nil
-						i.Close()
-						i.currentFile = nil
-						i.currentReader = nil
-						lastTime = 0
-						i.updateFile()
-
-						continue
-					} else {
-						break
-					}
-				}
-
-				continue
 			}
 		}
 
 		if bytes.Equal(payloadSeparatorAsBytes[1:], line) {
 			asBytes := buffer.Bytes()
-			buffer.Reset()
-
-			meta := payloadMeta(asBytes)
-
-			if len(meta) > 2 && meta[0][0] == RequestPayload {
-				ts, _ := strconv.ParseInt(string(meta[2]), 10, 64)
-
-				if lastTime != 0 {
-					timeDiff := ts - lastTime
-
-					if i.speedFactor != 1 {
-						timeDiff = int64(float64(timeDiff) / i.speedFactor)
-					}
-
-					time.Sleep(time.Duration(timeDiff))
-				}
-
-				lastTime = ts
-			}
 
 			// Bytes() returns only pointer, so to remove data-race copy the data to an array
-			newBuf := make([]byte, len(asBytes)-1)
+			newBuf := make([]byte, len(asBytes) - 1)
 			copy(newBuf, asBytes)
-
+			return newBuf
-			i.data <- newBuf
-		} else {
-			buffer.Write(line)
 		}
 
+		buffer.Write(line)
+	}
+}
+
+func (i *FileInput) nextInputReader() *fileInputReader {
+	var nextFileInputReader *fileInputReader
+	for _, fileInputReader := range i.fileInputReaders {
+		if fileInputReader.file == nil {
+			continue
+		}
+
+		if fileInputReader.meta[0][0] == ResponsePayload {
+			return fileInputReader
+		}
+
+		if nextFileInputReader == nil || nextFileInputReader.timestamp > fileInputReader.timestamp {
+			nextFileInputReader = fileInputReader
+			continue
+		}
+	}
+
+	return nextFileInputReader;
+}
+
+func (i *FileInput) emit() {
+	var lastTime int64 = -1
+
+	for {
+		fileInputReader := i.nextInputReader()
+
+		if fileInputReader == nil {
+			if i.loop {
+				i.init()
+				lastTime = -1
+				continue
+			} else {
+				break;
+			}
+		}
+
+		if fileInputReader.meta[0][0] == RequestPayload {
+			lastTime = i.simulateRequestDelay(fileInputReader, lastTime)
+		}
+
+		select {
+		case <-i.exit:
+			for _, fileInputReader := range i.fileInputReaders {
+				if fileInputReader.file != nil {
+					fileInputReader.file.Close()
+				}
+			}
+			break
+		case i.data <- fileInputReader.data:
+			fileInputReader.readNextInput()
+		}
 	}
 
 	log.Printf("FileInput: end of file '%s'\n", i.path)
 }
 
+func (i*FileInput) simulateRequestDelay(fileInputReader *fileInputReader, lastTime int64) int64 {
+	if lastTime != -1 {
+		timeDiff := fileInputReader.timestamp - lastTime
+
+		if i.speedFactor != 1 {
+			timeDiff = int64(float64(timeDiff) / i.speedFactor)
+		}
+
+		time.Sleep(time.Duration(timeDiff))
+	}
+
+	return fileInputReader.timestamp
+}
+
 func (i *FileInput) Close() error {
-	i.currentFile.Close()
+	i.exit <- true
 	return nil
 }
+

input_file_test.go

@@ -97,35 +97,109 @@ func TestInputFileWithGETAndPOST(t *testing.T) {
 
 }
 
-func TestInputFileMultipleFiles(t *testing.T) {
+func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) {
 	rnd := rand.Int63()
 
 	file1, _ := os.OpenFile(fmt.Sprintf("/tmp/%d_0", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
 	file1.Write([]byte("1 1 1\ntest1"))
 	file1.Write([]byte(payloadSeparator))
-	file1.Write([]byte("1 1 2\ntest2"))
+	file1.Write([]byte("1 1 3\ntest2"))
 	file1.Write([]byte(payloadSeparator))
 	file1.Close()
 
 	file2, _ := os.OpenFile(fmt.Sprintf("/tmp/%d_1", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
-	file2.Write([]byte("1 1 3\ntest3"))
+	file2.Write([]byte("1 1 2\ntest3"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Write([]byte("1 1 4\ntest4"))
 	file2.Write([]byte(payloadSeparator))
 	file2.Close()
 
 	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
 	buf := make([]byte, 1000)
-	n, _ := input.Read(buf)
+
-	if buf[10] != '1' {
+	for i := '1'; i <= '4'; i++ {
-		t.Error("Shound emit requests in right order", string(buf[:n]))
+		n, _ := input.Read(buf)
-	}
+		if buf[4] != byte(i) {
-	input.Read(buf)
+			t.Error("Should emit requests in right order", string(buf[:n]))
-	if buf[10] != '2' {
+		}
-		t.Error("Shound emit requests in right order", string(buf[:n]))
 	}
 
-	input.Read(buf)
+	os.Remove(file1.Name())
-	if buf[10] != '3' {
+	os.Remove(file2.Name())
-		t.Error("Shound emit requests from second file", string(buf[:n]))
+}
+
+func TestInputFileRequestsWithLatency(t *testing.T) {
+	rnd := rand.Int63()
+
+	file, _ := os.OpenFile(fmt.Sprintf("/tmp/%d", rnd), os.O_WRONLY | os.O_CREATE | os.O_TRUNC, 0660)
+	defer file.Close()
+
+	file.Write([]byte("1 1 100000000\nrequest1"))
+	file.Write([]byte(payloadSeparator))
+	file.Write([]byte("1 2 150000000\nrequest2"))
+	file.Write([]byte(payloadSeparator))
+	file.Write([]byte("1 3 250000000\nrequest3"))
+	file.Write([]byte(payloadSeparator))
+
+	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false)
+	buf := make([]byte, 1000)
+
+	start := time.Now().UnixNano()
+	for i := 0; i < 3; i++ {
+		input.Read(buf)
+	}
+	end := time.Now().UnixNano()
+
+	var expectedLatency int64 = 250000000 - 100000000
+	realLatency := end - start
+	if realLatency < expectedLatency {
+		t.Errorf("Should emit requests respecting latency. Expected: %v, real: %v", expectedLatency, realLatency)
+	}
+
+	if realLatency > expectedLatency + 10000000 {
+		t.Errorf("Should emit requests respecting latency. Expected: %v, real: %v", expectedLatency, realLatency)
+
+	}
+}
+
+func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) {
+	rnd := rand.Int63()
+
+	file1, _ := os.OpenFile(fmt.Sprintf("/tmp/%d_0", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
+	file1.Write([]byte("1 1 1\nrequest1"))
+	file1.Write([]byte(payloadSeparator))
+	file1.Write([]byte("2 1 1\nresponse1"))
+	file1.Write([]byte(payloadSeparator))
+	file1.Write([]byte("1 2 3\nrequest2"))
+	file1.Write([]byte(payloadSeparator))
+	file1.Write([]byte("2 2 3\nresponse2"))
+	file1.Write([]byte(payloadSeparator))
+	file1.Close()
+
+	file2, _ := os.OpenFile(fmt.Sprintf("/tmp/%d_1", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
+	file2.Write([]byte("1 3 2\nrequest3"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Write([]byte("2 3 2\nresponse3"))
+ 	file2.Write([]byte(payloadSeparator))
+	file2.Write([]byte("1 4 4\nrequest4"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Write([]byte("2 4 4\nresponse4"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Close()
+
+	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
+	buf := make([]byte, 1000)
+
+	for i := '1'; i <= '4'; i++ {
+		n, _ := input.Read(buf)
+		if buf[0] != '1' && buf[4] != byte(i) {
+			t.Error("Shound emit requests in right order", string(buf[:n]))
+		}
+
+		n, _ = input.Read(buf)
+		if buf[0] != '2' && buf[4] != byte(i) {
+			t.Error("Shound emit responses in right order", string(buf[:n]))
+		}
 	}
 
 	os.Remove(file1.Name())
@@ -149,8 +223,8 @@ func TestInputFileLoop(t *testing.T) {
 	for i := 0; i < 1000; i++ {
 		input.Read(buf)
 	}
-	input.Close()
 
+	input.Close()
 	os.Remove(file.Name())
 }