diff --git a/emitter.go b/emitter.go index afbe7db..1fe6821 100644 --- a/emitter.go +++ b/emitter.go @@ -8,41 +8,41 @@ import ( ) // Start initialize loop for sending data from inputs to outputs -func Start(stop chan int) { +func Start(plugins *InOutPlugins, stop chan int) { if Settings.middleware != "" { middleware := NewMiddleware(Settings.middleware) - for _, in := range Plugins.Inputs { + for _, in := range plugins.Inputs { middleware.ReadFrom(in) } // We are going only to read responses, so using same ReadFrom method - for _, out := range Plugins.Outputs { + for _, out := range plugins.Outputs { if r, ok := out.(io.Reader); ok { middleware.ReadFrom(r) } } go func() { - if err := CopyMulty(middleware, Plugins.Outputs...); err != nil { + if err := CopyMulty(middleware, plugins.Outputs...); err != nil { log.Println("Error during copy: ", err) close(stop) } }() } else { - for _, in := range Plugins.Inputs { + for _, in := range plugins.Inputs { go func(in io.Reader) { - if err := CopyMulty(in, Plugins.Outputs...); err != nil { + if err := CopyMulty(in, plugins.Outputs...); err != nil { log.Println("Error during copy: ", err) close(stop) } }(in) } - for _, out := range Plugins.Outputs { + for _, out := range plugins.Outputs { if r, ok := out.(io.Reader); ok { go func(r io.Reader) { - if err := CopyMulty(r, Plugins.Outputs...); err != nil { + if err := CopyMulty(r, plugins.Outputs...); err != nil { log.Println("Error during copy: ", err) close(stop) } @@ -54,7 +54,7 @@ func Start(stop chan int) { for { select { case <-stop: - finalize() + finalize(plugins) return case <-time.After(100 * time.Millisecond): } diff --git a/emitter_test.go b/emitter_test.go index 419b06b..5157e52 100644 --- a/emitter_test.go +++ b/emitter_test.go @@ -17,10 +17,12 @@ func TestEmitter(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < 1000; i++ { wg.Add(1) @@ -43,12 +45,14 @@ func TestEmitterFiltered(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } methods := HTTPMethods{[]byte("GET")} Settings.modifierConfig = HTTPModifierConfig{methods: methods} - go Start(quit) + go Start(plugins, quit) wg.Add(2) @@ -97,12 +101,14 @@ func TestEmitterRoundRobin(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output1, output2} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output1, output2}, + } Settings.splitOutput = true - go Start(quit) + go Start(plugins, quit) for i := 0; i < 1000; i++ { wg.Add(1) @@ -130,10 +136,12 @@ func BenchmarkEmitter(b *testing.B) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) b.ResetTimer() diff --git a/gor.go b/gor.go index 005527d..fbc39c7 100644 --- a/gor.go +++ b/gor.go @@ -50,6 +50,7 @@ func main() { } args := os.Args[1:] + var plugins *InOutPlugins if len(args) > 0 && args[0] == "file-server" { if len(args) != 2 { log.Fatal("You should specify port and IP (optional) for the file server. Example: `gor file-server :80`") @@ -61,12 +62,12 @@ func main() { log.Fatal(http.ListenAndServe(args[1], loggingMiddleware(http.FileServer(http.Dir(dir))))) } else { flag.Parse() - InitPlugins() + plugins = InitPlugins() } fmt.Println("Version:", VERSION) - if len(Plugins.Inputs) == 0 || len(Plugins.Outputs) == 0 { + if len(plugins.Inputs) == 0 || len(plugins.Outputs) == 0 { log.Fatal("Required at least 1 input and 1 output") } @@ -88,7 +89,7 @@ func main() { signal.Notify(c, os.Interrupt, syscall.SIGTERM) go func() { <-c - finalize() + finalize(plugins) os.Exit(1) }() @@ -101,11 +102,11 @@ func main() { }) } - Start(closeCh) + Start(plugins, closeCh) } -func finalize() { - for _, p := range Plugins.All { +func finalize(plugins *InOutPlugins) { + for _, p := range plugins.All { if cp, ok := p.(io.Closer); ok { cp.Close() } diff --git a/input_file_test.go b/input_file_test.go index 9029611..e5fe3a0 100644 --- a/input_file_test.go +++ b/input_file_test.go @@ -322,10 +322,12 @@ func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile { outputFile := NewFileOutput(f.Name(), &FileOutputConfig{flushInterval: time.Minute, append: true}) - Plugins.Inputs = requestGenerator.inputs - Plugins.Outputs = []io.Writer{output, outputFile} + plugins := &InOutPlugins{ + Inputs: requestGenerator.inputs, + Outputs: []io.Writer{output, outputFile}, + } - go Start(quit) + go Start(plugins, quit) requestGenerator.emit() requestGenerator.wg.Wait() @@ -350,11 +352,13 @@ func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } wg.Add(count) - go Start(quit) + go Start(plugins, quit) done := make(chan int, 1) go func() { diff --git a/input_http_test.go b/input_http_test.go index 0d09fd1..297cdb0 100644 --- a/input_http_test.go +++ b/input_http_test.go @@ -21,10 +21,12 @@ func TestHTTPInput(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) address := strings.Replace(input.listener.Addr().String(), "[::]", "127.0.0.1", -1) @@ -55,10 +57,12 @@ func TestInputHTTPLargePayload(t *testing.T) { } wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) wg.Add(1) address := strings.Replace(input.listener.Addr().String(), "[::]", "127.0.0.1", -1) diff --git a/input_raw_test.go b/input_raw_test.go index f4dcf66..5df2052 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -65,12 +65,14 @@ func TestRAWInputIPv4(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } client := NewHTTPClient("http://"+listener.Addr().String(), &HTTPClientConfig{}) - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { // request + response @@ -113,12 +115,14 @@ func TestRAWInputNoKeepAlive(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } client := NewHTTPClient("http://"+listener.Addr().String(), &HTTPClientConfig{}) - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { // request + response @@ -169,12 +173,14 @@ func TestRAWInputIPv6(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } client := NewHTTPClient("http://"+listener.Addr().String(), &HTTPClientConfig{}) - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { // request + response @@ -234,10 +240,12 @@ func TestInputRAW100Expect(t *testing.T) { httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{testOutput, httpOutput} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{testOutput, httpOutput}, + } - go Start(quit) + go Start(plugins, quit) // Origin + Response/Request Test Output + Request Http Output wg.Add(4) @@ -284,10 +292,12 @@ func TestInputRAWChunkedEncoding(t *testing.T) { httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: true}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{httpOutput} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{httpOutput}, + } - go Start(quit) + go Start(plugins, quit) wg.Add(2) @@ -350,10 +360,12 @@ func TestInputRAWLargePayload(t *testing.T) { httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: false}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{httpOutput} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{httpOutput}, + } - go Start(quit) + go Start(plugins, quit) wg.Add(2) curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--header", "Expect:", "--data-binary", "@/tmp/large") @@ -394,10 +406,12 @@ func BenchmarkRAWInput(b *testing.B) { httpOutput := NewLimiter(NewHTTPOutput(upstreamAddr, &HTTPOutputConfig{}), "10%") - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output, httpOutput} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output, httpOutput}, + } - go Start(quit) + go Start(plugins, quit) emitted := 0 fileContent, _ := ioutil.ReadFile("LICENSE.txt") diff --git a/input_tcp_test.go b/input_tcp_test.go index e5b5ba9..43ca7b7 100644 --- a/input_tcp_test.go +++ b/input_tcp_test.go @@ -27,10 +27,12 @@ func TestTCPInput(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) tcpAddr, err := net.ResolveTCPAddr("tcp", input.listener.Addr().String()) @@ -107,10 +109,12 @@ func TestTCPInputSecure(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) conf := &tls.Config{ InsecureSkipVerify: true, diff --git a/limiter_test.go b/limiter_test.go index f1757d8..9813fdb 100644 --- a/limiter_test.go +++ b/limiter_test.go @@ -18,10 +18,12 @@ func TestOutputLimiter(t *testing.T) { }), "10") wg.Add(10) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { input.EmitGET() @@ -42,10 +44,12 @@ func TestInputLimiter(t *testing.T) { }) wg.Add(10) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { input.(*Limiter).plugin.(*TestInput).EmitGET() @@ -66,10 +70,12 @@ func TestPercentLimiter1(t *testing.T) { wg.Done() }), "0%") - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { input.EmitGET() @@ -91,10 +97,12 @@ func TestPercentLimiter2(t *testing.T) { }), "100%") wg.Add(100) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { input.EmitGET() diff --git a/middleware_test.go b/middleware_test.go index 28f65dc..02c94a0 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -124,11 +124,13 @@ func TestEchoMiddleware(t *testing.T) { // And redirect to another output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: false}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } // Start Gor - go Start(quit) + go Start(plugins, quit) // Wait till middleware initialization time.Sleep(100 * time.Millisecond) @@ -186,11 +188,13 @@ func TestTokenMiddleware(t *testing.T) { // And redirect to another output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: true}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } // Start Gor - go Start(quit) + go Start(plugins, quit) // Wait for middleware to initialize // Give go compiller time to build programm diff --git a/output_file_test.go b/output_file_test.go index 604af29..10f831e 100644 --- a/output_file_test.go +++ b/output_file_test.go @@ -20,10 +20,12 @@ func TestFileOutput(t *testing.T) { input := NewTestInput() output := NewFileOutput("/tmp/test_requests.gor", &FileOutputConfig{flushInterval: time.Minute, append: true}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { wg.Add(2) @@ -44,10 +46,12 @@ func TestFileOutput(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input2} - Plugins.Outputs = []io.Writer{output2} + plugins2 := &InOutPlugins{ + Inputs: []io.Reader{input2}, + Outputs: []io.Writer{output2}, + } - go Start(quit) + go Start(plugins2, quit) wg.Wait() close(quit) diff --git a/output_http_test.go b/output_http_test.go index 43010f3..930f9be 100644 --- a/output_http_test.go +++ b/output_http_test.go @@ -48,10 +48,12 @@ func TestHTTPOutput(t *testing.T) { wg.Done() }) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{http_output, output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{http_output, output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < 1; i++ { // 2 http-output, 2 - test output request, 2 - test output http response @@ -88,10 +90,12 @@ func TestHTTPOutputKeepOriginalHost(t *testing.T) { output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: false, OriginalHost: true}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) wg.Add(1) input.EmitGET() @@ -115,10 +119,12 @@ func TestOutputHTTPSSL(t *testing.T) { input := NewTestInput() output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) wg.Add(2) @@ -142,10 +148,12 @@ func BenchmarkHTTPOutput(b *testing.B) { input := NewTestInput() output := NewHTTPOutput(server.URL, &HTTPOutputConfig{}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < b.N; i++ { wg.Add(1) diff --git a/output_tcp_test.go b/output_tcp_test.go index 64ef8ca..9cb0100 100644 --- a/output_tcp_test.go +++ b/output_tcp_test.go @@ -19,10 +19,12 @@ func TestTCPOutput(t *testing.T) { input := NewTestInput() output := NewTCPOutput(listener.Addr().String(), &TCPOutputConfig{}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) for i := 0; i < 100; i++ { wg.Add(1) @@ -71,10 +73,12 @@ func BenchmarkTCPOutput(b *testing.B) { input := NewTestInput() output := NewTCPOutput(listener.Addr().String(), &TCPOutputConfig{}) - Plugins.Inputs = []io.Reader{input} - Plugins.Outputs = []io.Writer{output} + plugins := &InOutPlugins{ + Inputs: []io.Reader{input}, + Outputs: []io.Writer{output}, + } - go Start(quit) + go Start(plugins, quit) b.ResetTimer() for i := 0; i < b.N; i++ { diff --git a/plugins.go b/plugins.go index 7cc2f7a..954584d 100644 --- a/plugins.go +++ b/plugins.go @@ -17,7 +17,7 @@ type InOutPlugins struct { var pluginMu sync.Mutex // Plugins holds all the plugin objects -var Plugins *InOutPlugins = new(InOutPlugins) +var plugins *InOutPlugins = new(InOutPlugins) // extractLimitOptions detects if plugin get called with limiter support // Returns address and limit @@ -67,18 +67,18 @@ func registerPlugin(constructor interface{}, options ...interface{}) { // Some of the output can be Readers as well because return responses if isR && !isW { - Plugins.Inputs = append(Plugins.Inputs, pluginWrapper.(io.Reader)) + plugins.Inputs = append(plugins.Inputs, pluginWrapper.(io.Reader)) } if isW { - Plugins.Outputs = append(Plugins.Outputs, pluginWrapper.(io.Writer)) + plugins.Outputs = append(plugins.Outputs, pluginWrapper.(io.Writer)) } - Plugins.All = append(Plugins.All, plugin) + plugins.All = append(plugins.All, plugin) } // InitPlugins specify and initialize all available plugins -func InitPlugins() { +func InitPlugins() *InOutPlugins { pluginMu.Lock() defer pluginMu.Unlock() @@ -149,4 +149,6 @@ func InitPlugins() { if Settings.inputKafkaConfig.host != "" && Settings.inputKafkaConfig.topic != "" { registerPlugin(NewKafkaInput, "", &Settings.inputKafkaConfig) } + + return plugins } diff --git a/plugins_test.go b/plugins_test.go index 6275037..6331ca0 100644 --- a/plugins_test.go +++ b/plugins_test.go @@ -1,42 +1,38 @@ package main import ( - "io" "testing" ) func TestPluginsRegistration(t *testing.T) { - Plugins.Inputs = []io.Reader{} - Plugins.Outputs = []io.Writer{} - Settings.inputDummy = MultiOption{"[]"} Settings.outputDummy = MultiOption{"[]"} Settings.outputHTTP = MultiOption{"www.example.com|10"} Settings.inputFile = MultiOption{"/dev/null"} - InitPlugins() + plugins := InitPlugins() - if len(Plugins.Inputs) != 2 { - t.Errorf("Should be 2 inputs %d", len(Plugins.Inputs)) + if len(plugins.Inputs) != 2 { + t.Errorf("Should be 2 inputs %d", len(plugins.Inputs)) } - if _, ok := Plugins.Inputs[0].(*DummyInput); !ok { + if _, ok := plugins.Inputs[0].(*DummyInput); !ok { t.Errorf("First input should be DummyInput") } - if _, ok := Plugins.Inputs[1].(*FileInput); !ok { + if _, ok := plugins.Inputs[1].(*FileInput); !ok { t.Errorf("Second input should be FileInput") } - if len(Plugins.Outputs) != 2 { - t.Errorf("Should be 2 output %d", len(Plugins.Outputs)) + if len(plugins.Outputs) != 2 { + t.Errorf("Should be 2 output %d", len(plugins.Outputs)) } - if _, ok := Plugins.Outputs[0].(*DummyOutput); !ok { + if _, ok := plugins.Outputs[0].(*DummyOutput); !ok { t.Errorf("First output should be DummyOutput") } - if l, ok := Plugins.Outputs[1].(*Limiter); ok { + if l, ok := plugins.Outputs[1].(*Limiter); ok { if _, ok := l.plugin.(*HTTPOutput); !ok { t.Errorf("HTTPOutput should be wrapped in limiter") } diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index 791b6d4..d09fcca 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -199,7 +199,9 @@ func (t *TCPMessage) checkSeqIntegrity() { var bEmptyLine = []byte("\r\n\r\n") var bBR = []byte("\r\n") -var bChunkEnd = []byte("\r\n0\r\n\r\n") + +// last-chunk always is 0\r\n\r\n\. More info https://tools.ietf.org/html/rfc2616#section-3.6.1 +var bChunkEnd = []byte("0\r\n\r\n") func (t *TCPMessage) updateHeadersPacket() { if len(t.packets) == 1 {