Replace the use of bufio.Scanner with bufio.Reader so that requests over 64Kb in size can be replayed from a capture file.

.gitignore

@@ -11,4 +11,5 @@
 
 *.test
 .idea
+*.iml
 gor

input_file.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"bufio"
+	"bytes"
+	"io"
 	"log"
 	"os"
 	"strconv"
@@ -53,36 +55,53 @@ func (i *FileInput) String() string {
 func (i *FileInput) emit() {
 	var lastTime int64
 
-	// reader := bufio.NewReader(conn)
-	scanner := bufio.NewScanner(i.file)
-	scanner.Split(payloadScanner)
+	payloadSeparatorAsBytes := []byte(payloadSeparator)
+	reader := bufio.NewReader(i.file)
+	var buffer bytes.Buffer
 
-	for scanner.Scan() {
-		buf := scanner.Bytes()
-		meta := payloadMeta(buf)
+	for {
+		line, err := reader.ReadBytes('\n')
 
-		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))
+		if err != nil {
+			if err != io.EOF {
+				log.Fatal(err)
 			}
+			break
 
-			lastTime = ts
 		}
 
-		// scanner returs only pointer, so to remove data-race we have to allocate new array
-		newBuf := make([]byte, len(buf))
-		copy(newBuf, buf)
+		if bytes.Equal(payloadSeparatorAsBytes[1:], line) {
+			asBytes := buffer.Bytes()[:buffer.Len()-1]
+			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))
+			copy(newBuf, asBytes)
+
+			i.data <- newBuf
+		} else {
+			buffer.Write(line)
+		}
 
-		i.data <- newBuf
 	}
 
-	log.Println("FileInput: end of file")
+	log.Printf("FileInput: end of file '%s'\n", i.path)
 }

input_file_test.go

@@ -0,0 +1,220 @@
+package main
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"io/ioutil"
+	"os"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+)
+
+func TestInputFileWithGET(t *testing.T) {
+
+	input := NewTestInput()
+	rg := NewRequestGenerator([]io.Reader{input}, func() { input.EmitGET() }, 1)
+	readPayloads := [][]byte{}
+
+	// Given a capture file with a GET request
+	expectedCaptureFile := CreateCaptureFile(rg)
+	defer expectedCaptureFile.TearDown()
+
+	// When the request is read from the capture file
+	err := ReadFromCaptureFile(expectedCaptureFile.file, 1, func(data []byte) {
+		readPayloads = append(readPayloads, Duplicate(data))
+	})
+
+	// The read request should match the original request
+	if err != nil {
+		t.Error(err)
+	} else {
+		if !expectedCaptureFile.PayloadsEqual(readPayloads) {
+			t.Error("Request read back from file should match")
+		}
+	}
+
+}
+
+func TestInputFileWithPayloadLargerThan64Kb(t *testing.T) {
+
+	input := NewTestInput()
+	rg := NewRequestGenerator([]io.Reader{input}, func() { input.EmitSizedPOST(64 * 1024) }, 1)
+	readPayloads := [][]byte{}
+
+	// Given a capture file with a request over 64Kb
+	expectedCaptureFile := CreateCaptureFile(rg)
+	defer expectedCaptureFile.TearDown()
+
+	// When the request is read from the capture file
+	err := ReadFromCaptureFile(expectedCaptureFile.file, 1, func(data []byte) {
+		readPayloads = append(readPayloads, Duplicate(data))
+	})
+
+	// The read request should match the original request
+	if err != nil {
+		t.Error(err)
+	} else {
+		if !expectedCaptureFile.PayloadsEqual(readPayloads) {
+			t.Error("Request read back from file should match")
+		}
+	}
+
+}
+
+func TestInputFileWithGETAndPOST(t *testing.T) {
+
+	input := NewTestInput()
+	rg := NewRequestGenerator([]io.Reader{input}, func() {
+		input.EmitGET()
+		input.EmitPOST()
+	}, 2)
+	readPayloads := [][]byte{}
+
+	// Given a capture file with a GET request
+	expectedCaptureFile := CreateCaptureFile(rg)
+	defer expectedCaptureFile.TearDown()
+
+	// When the requests are read from the capture file
+	err := ReadFromCaptureFile(expectedCaptureFile.file, 2, func(data []byte) {
+		readPayloads = append(readPayloads, Duplicate(data))
+	})
+
+	// The read requests should match the original request
+	if err != nil {
+		t.Error(err)
+	} else {
+		if !expectedCaptureFile.PayloadsEqual(readPayloads) {
+			t.Error("Request read back from file should match")
+		}
+	}
+
+}
+
+type CaptureFile struct {
+	data [][]byte
+	file *os.File
+}
+
+func NewExpectedCaptureFile(data [][]byte, file *os.File) *CaptureFile {
+	ecf := new(CaptureFile)
+	ecf.file = file
+	ecf.data = data
+	return ecf
+}
+
+func (expectedCaptureFile *CaptureFile) TearDown() {
+	if expectedCaptureFile.file != nil {
+		syscall.Unlink(expectedCaptureFile.file.Name())
+	}
+}
+
+type RequestGenerator struct {
+	inputs []io.Reader
+	emit   func()
+	wg     *sync.WaitGroup
+}
+
+func NewRequestGenerator(inputs []io.Reader, emit func(), count int) (rg *RequestGenerator) {
+	rg = new(RequestGenerator)
+	rg.inputs = inputs
+	rg.emit = emit
+	rg.wg = new(sync.WaitGroup)
+	rg.wg.Add(count)
+	return
+}
+
+func (expectedCaptureFile *CaptureFile) PayloadsEqual(other [][]byte) bool {
+
+	if len(expectedCaptureFile.data) != len(other) {
+		return false
+	}
+
+	for i, payload := range other {
+
+		if !bytes.Equal(expectedCaptureFile.data[i], payload) {
+			return false
+		}
+
+	}
+
+	return true
+
+}
+
+func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile {
+
+	f, err := ioutil.TempFile("", "testmainconf")
+	if err != nil {
+		panic(err)
+	}
+
+	quit := make(chan int)
+
+	readPayloads := [][]byte{}
+	output := NewTestOutput(func(data []byte) {
+
+		readPayloads = append(readPayloads, Duplicate(data))
+
+		requestGenerator.wg.Done()
+	})
+
+	outputFile := NewFileOutput(f.Name())
+
+	Plugins.Inputs = requestGenerator.inputs
+	Plugins.Outputs = []io.Writer{output, outputFile}
+
+	go Start(quit)
+
+	requestGenerator.emit()
+	requestGenerator.wg.Wait()
+
+	close(quit)
+
+	return NewExpectedCaptureFile(readPayloads, f)
+
+}
+
+func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback) (err error) {
+
+	quit := make(chan int)
+	wg := new(sync.WaitGroup)
+
+	input := NewFileInput(captureFile.Name())
+	output := NewTestOutput(func(data []byte) {
+		callback(data)
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	wg.Add(count)
+	go Start(quit)
+
+	done := make(chan int, 1)
+	go func() {
+		wg.Wait()
+		done <- 1
+	}()
+
+	select {
+	case <-done:
+		break
+	case <-time.After(2 * time.Second):
+		err = errors.New("Timed out")
+	}
+	close(quit)
+
+	return
+
+}
+
+func Duplicate(data []byte) (duplicate []byte) {
+	duplicate = make([]byte, len(data))
+	copy(duplicate, data)
+
+	return
+}

test_input.go

@@ -53,6 +53,18 @@ func (i *TestInput) EmitLargePOST() {
 	rs := base64.URLEncoding.EncodeToString(rb)
 
 	i.data <- []byte("POST / HTTP/1.1\nHost: www.w3.org\nContent-Length:5242880\r\n\r\n" + rs)
+	Debug("Sent large POST")
+}
+
+// EmitSizedPOST emit a POST with a payload set to a supplied size
+func (i *TestInput) EmitSizedPOST(payloadSize int) {
+	rb := make([]byte, payloadSize)
+	rand.Read(rb)
+
+	rs := base64.URLEncoding.EncodeToString(rb)
+
+	i.data <- []byte("POST / HTTP/1.1\nHost: www.w3.org\nContent-Length:5242880\r\n\r\n" + rs)
+	Debug("Sent large POST")
 }
 
 // EmitOPTIONS emits OPTIONS request, similar to GET