From ce76453fa63454d9e5ce039fede572e883f46bcc Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Tue, 29 Oct 2013 11:42:15 +0100 Subject: [PATCH] Added tests for all plugins --- emitter.go | 11 +++---- emitter_test.go | 31 +++++++++++++++++++ gor/gor.go | 2 +- input_raw_test.go | 44 ++++++++++++++++++++++++++ input_tcp.go | 20 ++++++------ input_tcp_test.go | 47 ++++++++++++++++++++++++++++ output_http_test.go | 43 ++++++++++++++++++++++++++ output_tcp_test.go | 75 +++++++++++++++++++++++++++++++++++++++++++++ settings.go | 11 +++++++ test_input.go | 31 +++++++++++++++++++ test_output.go | 24 +++++++++++++++ 11 files changed, 323 insertions(+), 16 deletions(-) create mode 100644 emitter_test.go create mode 100644 input_raw_test.go create mode 100644 input_tcp_test.go create mode 100644 output_http_test.go create mode 100644 output_tcp_test.go create mode 100644 test_input.go create mode 100644 test_output.go diff --git a/emitter.go b/emitter.go index 67d60c8..4fcb8fa 100644 --- a/emitter.go +++ b/emitter.go @@ -2,17 +2,16 @@ package gor import ( "io" - "log" - "time" ) -func Start() { +func Start(stop chan int) { for _, in := range Plugins.Inputs { CopyMulty(in, Plugins.Outputs...) } - for { - time.Sleep(time.Second) + select { + case <-stop: + return } } @@ -23,7 +22,7 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) { for { nr, er := src.Read(buf) if nr > 0 { - log.Println("Sending", src, ": ", string(buf[0:nr])) + Debug("Sending", src, ": ", string(buf[0:nr])) for _, dst := range writers { dst.Write(buf[0:nr]) diff --git a/emitter_test.go b/emitter_test.go new file mode 100644 index 0000000..6a603e4 --- /dev/null +++ b/emitter_test.go @@ -0,0 +1,31 @@ +package gor + +import ( + "io" + "sync" + "testing" +) + +func TestEmitter(t *testing.T) { + wg := new(sync.WaitGroup) + quit := make(chan int) + + input := NewTestInput() + output := NewTestOutput(func(data []byte) { + wg.Done() + }) + + Plugins.Inputs = []io.Reader{input} + Plugins.Outputs = []io.Writer{output} + + go Start(quit) + + for i := 0; i < 1000; i++ { + wg.Add(1) + input.EmitGET() + } + + wg.Wait() + + close(quit) +} diff --git a/gor/gor.go b/gor/gor.go index 97739fc..8898d4f 100644 --- a/gor/gor.go +++ b/gor/gor.go @@ -54,7 +54,7 @@ func main() { profileCPU(*cpuprofile) } - gor.Start() + gor.Start(nil) } func profileCPU(cpuprofile string) { diff --git a/input_raw_test.go b/input_raw_test.go new file mode 100644 index 0000000..93a3f0a --- /dev/null +++ b/input_raw_test.go @@ -0,0 +1,44 @@ +package gor + +import ( + "io" + "net/http" + "sync" + "testing" +) + +func TestRAWInput(t *testing.T) { + startHTTP := func(addr string, cb func(*http.Request)) { + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cb(r) + }) + + go http.ListenAndServe(addr, handler) + } + + wg := new(sync.WaitGroup) + quit := make(chan int) + + input := NewRAWInput("127.0.0.1:50004") + output := NewTestOutput(func(data []byte) { + wg.Done() + }) + + startHTTP("127.0.0.1:50004", func(req *http.Request) { + wg.Done() + }) + + Plugins.Inputs = []io.Reader{input} + Plugins.Outputs = []io.Writer{output} + + go Start(quit) + + for i := 0; i < 100; i++ { + wg.Add(1) + http.Get("http://127.0.0.1:50004") + } + + wg.Wait() + + close(quit) +} diff --git a/input_tcp.go b/input_tcp.go index bfccdce..510ffeb 100644 --- a/input_tcp.go +++ b/input_tcp.go @@ -19,7 +19,7 @@ func NewTCPInput(address string) (i *TCPInput) { i.data = make(chan []byte) i.address = address - go i.listen(address) + i.listen(address) return } @@ -38,16 +38,18 @@ func (i *TCPInput) listen(address string) { log.Fatal("Can't start:", err) } - for { - conn, err := listener.Accept() + go func() { + for { + conn, err := listener.Accept() - if err != nil { - log.Println("Error while Accept()", err) - continue + if err != nil { + log.Println("Error while Accept()", err) + continue + } + + go i.handleConnection(conn) } - - go i.handleConnection(conn) - } + }() } func (i *TCPInput) handleConnection(conn net.Conn) { diff --git a/input_tcp_test.go b/input_tcp_test.go new file mode 100644 index 0000000..293a0f5 --- /dev/null +++ b/input_tcp_test.go @@ -0,0 +1,47 @@ +package gor + +import ( + "io" + "log" + "net" + "sync" + "testing" +) + +func TestTCPInput(t *testing.T) { + wg := new(sync.WaitGroup) + quit := make(chan int) + + input := NewTCPInput("127.0.0.1:50001") + output := NewTestOutput(func(data []byte) { + wg.Done() + }) + + Plugins.Inputs = []io.Reader{input} + Plugins.Outputs = []io.Writer{output} + + go Start(quit) + + for i := 0; i < 100; i++ { + wg.Add(1) + sendTCP("127.0.0.1:50001", []byte("GET / HTTP/1.1\r\n\r\n")) + } + + wg.Wait() + + close(quit) +} + +func sendTCP(addr string, data []byte) { + tcpAddr, err := net.ResolveTCPAddr("tcp", addr) + if err != nil { + log.Fatal(err) + } + + conn, err := net.DialTCP("tcp", nil, tcpAddr) + if err != nil { + log.Fatal(err) + } + + conn.Write(data) +} diff --git a/output_http_test.go b/output_http_test.go new file mode 100644 index 0000000..f78d055 --- /dev/null +++ b/output_http_test.go @@ -0,0 +1,43 @@ +package gor + +import ( + "io" + "net/http" + "sync" + "testing" +) + +func TestHTTPOutput(t *testing.T) { + startHTTP := func(addr string, cb func(*http.Request)) { + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cb(r) + }) + + go http.ListenAndServe(addr, handler) + } + + wg := new(sync.WaitGroup) + quit := make(chan int) + + input := NewTestInput() + output := NewHTTPOutput("127.0.0.1:50003") + + startHTTP("127.0.0.1:50003", func(req *http.Request) { + wg.Done() + }) + + Plugins.Inputs = []io.Reader{input} + Plugins.Outputs = []io.Writer{output} + + go Start(quit) + + for i := 0; i < 100; i++ { + wg.Add(2) + input.EmitGET() + input.EmitPOST() + } + + wg.Wait() + + close(quit) +} diff --git a/output_tcp_test.go b/output_tcp_test.go new file mode 100644 index 0000000..beffe87 --- /dev/null +++ b/output_tcp_test.go @@ -0,0 +1,75 @@ +package gor + +import ( + "io" + "log" + "net" + "sync" + "testing" +) + +func TestTCPOutput(t *testing.T) { + wg := new(sync.WaitGroup) + quit := make(chan int) + + input := NewTestInput() + output := NewTCPOutput(":50002") + + startTCP(":50002", func(data []byte) { + wg.Done() + }) + + Plugins.Inputs = []io.Reader{input} + Plugins.Outputs = []io.Writer{output} + + go Start(quit) + + for i := 0; i < 100; i++ { + wg.Add(1) + input.EmitGET() + } + + wg.Wait() + + close(quit) +} + +func startTCP(addr string, cb func([]byte)) { + listener, err := net.Listen("tcp", addr) + + if err != nil { + log.Fatal("Can't start:", err) + } + + go func() { + for { + conn, _ := listener.Accept() + + var read = true + var response []byte + var buf []byte + + buf = make([]byte, 4094) + + for read { + n, err := conn.Read(buf) + + switch err { + case io.EOF: + read = false + case nil: + response = append(response, buf[:n]...) + if n < 4096 { + read = false + } + default: + read = false + } + } + + cb(response) + + conn.Close() + } + }() +} diff --git a/settings.go b/settings.go index bbc0d7c..a16a99a 100644 --- a/settings.go +++ b/settings.go @@ -2,9 +2,12 @@ package gor import ( "flag" + "log" ) type AppSettings struct { + verbose bool + inputDummy MultiOption outputDummy MultiOption @@ -22,6 +25,8 @@ type AppSettings struct { var Setttings AppSettings = AppSettings{} func init() { + flag.BoolVar(&Setttings.verbose, "verbose", false, "") + flag.Var(&Setttings.inputDummy, "input-dummy", "") flag.Var(&Setttings.outputDummy, "output-dummy", "") @@ -35,3 +40,9 @@ func init() { flag.Var(&Setttings.outputHTTP, "output-http", "") } + +func Debug(args ...interface{}) { + if Setttings.verbose { + log.Println(args...) + } +} diff --git a/test_input.go b/test_input.go new file mode 100644 index 0000000..d8c6cdb --- /dev/null +++ b/test_input.go @@ -0,0 +1,31 @@ +package gor + +type TestInput struct { + data chan []byte +} + +func NewTestInput() (i *TestInput) { + i = new(TestInput) + i.data = make(chan []byte) + + return +} + +func (i *TestInput) Read(data []byte) (int, error) { + buf := <-i.data + copy(data, buf) + + return len(buf), nil +} + +func (i *TestInput) EmitGET() { + i.data <- []byte("GET / HTTP/1.1\r\n\r\n") +} + +func (i *TestInput) EmitPOST() { + i.data <- []byte("POST /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\na=1&b=2\r\n\r\n") +} + +func (i *TestInput) String() string { + return "Test Input" +} diff --git a/test_output.go b/test_output.go new file mode 100644 index 0000000..e350d64 --- /dev/null +++ b/test_output.go @@ -0,0 +1,24 @@ +package gor + +type writeCallback func(data []byte) + +type TestOutput struct { + cb writeCallback +} + +func NewTestOutput(cb writeCallback) (i *TestOutput) { + i = new(TestOutput) + i.cb = cb + + return +} + +func (i *TestOutput) Write(data []byte) (int, error) { + i.cb(data) + + return len(data), nil +} + +func (i *TestOutput) String() string { + return "Test Input" +}