diff --git a/.gitignore b/.gitignore index 3503fea..b9d19c7 100644 --- a/.gitignore +++ b/.gitignore @@ -11,4 +11,5 @@ *.test .idea +*.iml gor diff --git a/input_file.go b/input_file.go index 16d9765..755933b 100644 --- a/input_file.go +++ b/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) } diff --git a/input_file_test.go b/input_file_test.go new file mode 100644 index 0000000..cbe79ba --- /dev/null +++ b/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 +} diff --git a/test_input.go b/test_input.go index c699dff..cd146c5 100644 --- a/test_input.go +++ b/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