Fix tests

Tcp pool

.gitignore

@@ -0,0 +1,3 @@
+*.swp
+
+*.gor

.travis.yml

@@ -1,3 +1,3 @@
 language: go 
 go: 1.1
-script: sudo -E bash -c "source /etc/profile && gvm use go1.1 && export GOPATH=$HOME/gopath:$GOPATH && go test"
+script: sudo -E bash -c "source /etc/profile && gvm use go1.1 && export GOPATH=$HOME/gopath:$GOPATH && go get && go test -v"

CHANGELOG.md

@@ -1,5 +1,12 @@
+v0.7.0 - 31 Oct 2013
+* New modular architecture. Listener and Replay functionality merged.
+* Added option to equally split traffic between multiple outputs: --split-output true
+* Saving requests to file and replaying from it
+* Injecting custom headers to http requests
+* Advanced stats using ElasticSearch
+
 v0.3.5 - 15 Sep 2013
-* Significally improved test coverage
+* Significantly improved test coverage
 * Fixed bug with redirect replay https://github.com/buger/gor/pull/15
 * Added limit on listener side
 * Improved stability (catch and log panic, instead of exiting)
@@ -7,8 +14,8 @@ v0.3.5 - 15 Sep 2013
 
 v0.3.3 - 22 Jun 2013
 * Using TCP instead of UDP for communication between Listener and Replay
-* Significally improved performance
-* Fixed bugs causing locking and message droping (concurrency issues)
+* Significantly improved performance
+* Fixed bugs causing locking and message dropping (concurrency issues)
 * Rewrote concurrency model to use more channels
 
 v0.3 - 10 Jun 2013

ELASTICSEARCH.md

@@ -0,0 +1,65 @@
+gor & elasticsearch
+===================
+
+Prerequisites
+-------------
+
+- elasticsearch
+- kibana (Get it here: http://www.elasticsearch.org/overview/kibana/)
+- gor
+
+
+elasticsearch
+-------------
+
+The default elasticsearch configuration is just fine for most workloads. You won't need clustering, sharding or something like that.
+
+In this example we're installing it on our gor replay server which gives us the elasticsearch listener on _http://localhost:9200_
+
+
+kibana
+------
+
+Kibana (elasticsearch analytics web-ui) is just as simple. 
+Download it, extract it and serve it via a simple webserver.
+(Could be nginx or apache)
+
+You could also use a shell, ```cd``` into the kibana directory and start a little quick and dirty python webserver with:
+
+```
+python -m SimpleHTTPServer 8000
+```
+
+In this example we're also choosing the gor replay server as our kibana host. If you choose a different server you'll have to point kibana to your elasticsearch host.
+
+
+gor
+---
+
+Start your gor replay server with elasticsearch option:
+
+```
+./gor --input-raw :8000 --output-http http://staging.com  --output-http-elasticsearch localhost:9200/gor
+```
+
+
+(You don't have to create the index upfront. That will be done for you automatically)
+
+
+Now visit your kibana url, load the predefined dashboard from the gist https://gist.github.com/gottwald/b2c875037f24719a9616 and watch the data rush in.
+
+
+Troubleshooting
+---------------
+
+The replay process may complain about __too many open files__.
+That's because your typical linux shell has a small open files soft limit at 1024.
+You can easily raise that when you do this before starting your _gor replay_ process:
+
+```
+ulimit -n 64000
+```
+
+Please be aware, this is not a permanent setting. It's just valid for the following jobs you start from that shell.
+
+We reached the 1024 limit in our tests with a ubuntu box replaying about 9000 requests per minute. (We had very slow responses there, should be way more with fast responses)

Makefile

@@ -0,0 +1,11 @@
+all: build-x86 build-x64
+
+build-x64:
+	GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build
+	tar -czf gor_x64.tar.gz gor
+	rm gor
+
+build-x86:
+	GOOS=linux GOARCH=386 CGO_ENABLED=0 go build
+	tar -czf gor_x86.tar.gz gor
+	rm gor

Procfile

@@ -0,0 +1,4 @@
+web: python -m SimpleHTTPServer 8000
+replayed_web: python -m SimpleHTTPServer 8001
+listener: sudo -E go run ./bin/gor.go --input-raw :8000 --output-tcp :8002 --verbose
+replay: go run ./bin/gor.go --input-tcp :8002 --output-http localhost:8001 --verbose

README.md

@@ -1,100 +1,217 @@
+[![Stories in Ready](https://badge.waffle.io/buger/gor.png?label=ready)](https://waffle.io/buger/gor)  
 [![Build Status](https://travis-ci.org/buger/gor.png?branch=master)](https://travis-ci.org/buger/gor)
 
 ## About
 
-Gor is a simple http traffic replication tool written in Go. 
+Gor is a simple http traffic replication tool written in Go.
 Its main goal is to replay traffic from production servers to staging and dev environments.
 
 
-Now you can test your code on real user sessions in an automated and repeatable fashion.  
+Now you can test your code on real user sessions in an automated and repeatable fashion.
 **No more falling down in production!**
 
-Gor consists of 2 parts: listener and replay servers.
-
-The listener server catches http traffic from a given port in real-time
-and sends it to the replay server. 
-The replay server forwards traffic to a given address.
+Here is basic worlkflow: The listener server catches http traffic and sends it to the replay server or saves to file.The replay server forwards traffic to a given address.
 
 
 ![Diagram](http://i.imgur.com/9mqj2SK.png)
 
 
-## Basic example
+## Examples
 
+### Capture traffic from port
 ```bash
 # Run on servers where you want to catch traffic. You can run it on each `web` machine.
-sudo gor listen -p 80 -r replay.server.local:28020 
+sudo gor --input-raw :80 --output-tcp replay.local:28020
 
-# Replay server (replay.server.local). 
-gor replay -f http://staging.server -p 28020
+# Replay server (replay.local).
+gor --input-tcp replay.local:28020 --output-http http://staging.com
+```
+
+### Using 1 Gor instance for both listening and replaying
+It's recommended to use separate server for replaying traffic, but if you have enough CPU resources you can use single Gor instance.
+
+```
+sudo gor --input-raw :80 --output-http "http://staging.com"
 ```
 
 ## Advanced use
 
 ### Rate limiting
-Both replay and listener supports rate limiting. It can be useful if you want
+Both replay and listener support rate limiting. It can be useful if you want
 forward only part of production traffic and not overload your staging
 environment. You can specify your desired requests per second using the
 "|" operator after the server address:
 
+#### Limiting replay
 ```
 # staging.server will not get more than 10 requests per second
-gor replay -f "http://staging.server|10"
+gor --input-tcp :28020 --output-http "http://staging.com|10"
 ```
 
+#### Limiting listener
 ```
 # replay server will not get more than 10 requests per second
 # useful for high-load environments
-gor listen -p 8080 -r "replay.server.local:28020|10"
+gor --input-raw :80 --output-tcp "replay.local:28020|10"
 ```
 
 ### Forward to multiple addresses
 
-You can forward traffic to multiple endpoints. Just separate the addresses by comma.
+You can forward traffic to multiple endpoints. Just add multiple --output-* arguments.
 ```
-gor replay -f "http://staging.server|10,http://dev.server|5"
+gor --input-tcp :28020 --output-http "http://staging.com"  --output-http "http://dev.com"
 ```
 
+#### Splitting traffic
+By default it will send same traffic to all outputs, but you have options to equally split it:
+
+```
+gor --input-tcp :28020 --output-http "http://staging.com"  --output-http "http://dev.com" --split-output true
+```
+
+### Saving requests to file
+You can save requests to file, and replay them later:
+```
+# write to file
+gor --input-raw :80 --output-file requests.gor
+
+# read from file
+gor --input-file requests.gor --output-http "http://staging.com"
+```
+
+**Note:** Replay will preserve the original time differences between requests.
+
+### Injecting headers
+
+Additional headers can be injected/overwritten into requests during replay. This may be useful if the hostname that staging responds to differs from production, you need to identify requests generated by Gor, or enable feature flagged functionality in an application:
+
+```
+gor --input-raw :80 --output-http "http://staging.server" \
+    --output-http-header "Host: staging.server" \
+    --output-http-header "User-Agent: Replayed by Gor" -header " \
+    --output-http-header "Enable-Feature-X: true"
+```
+
+## Filtering HTTP methods
+
+Requests not matching a specified whitelist can be filtered out. For example to strip non-nullipotent requests:
+
+```
+gor --input-raw :80 --output-http "http://staging.server" \
+    --output-http-method GET \
+    --output-http-method OPTIONS
+```
+
+### Basic Auth
+
+If your development or staging environment is protected by Basic Authentication then those credentials can be injected in during the replay:
+
+```
+gor --input-raw :80 --output-http "http://user:pass@staging .com"
+```
+
+Note: This will overwrite any Authorization headers in the original request.
+
+## Stats 
+### ElasticSearch 
+For deep response analyze based on url, cookie, user-agent and etc. you can export response metadata to ElasticSearch. See [ELASTICSEARCH.md](ELASTICSEARCH.md) for more details.
+
+```
+gor --input-tcp :80 --output-http "http://staging.com" --output-http-elasticsearch "es_host:api_port/index_name"
+```
+
+
 ## Additional help
-```
-$ gor listen -h
-Usage of ./bin/gor-linux:
-  -i="any": By default it try to listen on all network interfaces.To get list of interfaces run `ifconfig`
-  -p=80: Specify the http server port whose traffic you want to capture
-  -r="localhost:28020": Address of replay server.
-```
 
-```
-$ gor replay -h
-Usage of ./bin/gor-linux:
-  -f="http://localhost:8080": http address to forward traffic.
-	You can limit requests per second by adding `|#{num}` after address.
-	If you have multiple addresses with different limits. For example: http://staging.example.com|100,http://dev.example.com|10
-  -ip="0.0.0.0": ip addresses to listen on
-  -p=28020: specify port number
-```
+Feel free to ask question directly by email or by creating github issue.
 
-## Pre-build binaries
+## Latest releases (including binaries)
 
-[Download binaries (linux 32/64, darwin)](https://drive.google.com/folderview?id=0B46uay48NwcfWFowc1E4a1BISVU&usp=sharing)
+https://github.com/buger/gor/releases
+
+## Command line reference
+`gor -h` output:
+```
+  -cpuprofile="": write cpu profile to file
+  -memprofile="": write memory profile to this file
+  
+  -input-dummy=[]: Used for testing outputs. Emits 'Get /' request every 1s
+
+  -input-file=[]: Read requests from file: 
+    gor --input-file ./requests.gor --output-http staging.com
+
+  -input-raw=[]: Capture traffic from given port (use RAW sockets and require *sudo* access):
+    # Capture traffic from 8080 port
+    gor --input-raw :8080 --output-http staging.com
+
+  -input-tcp=[]: Used for internal communication between Gor instances. Example: 
+    # Receive requests from other Gor instances on 28020 port, and redirect output to staging
+    gor --input-tcp :28020 --output-http staging.com
+
+  -output-dummy=[]: Used for testing inputs. Just prints data coming from inputs.
+
+  -output-file=[]: Write incoming requests to file: 
+    gor --input-raw :80 --output-file ./requests.gor
+
+  -output-http=[]: Forwards incoming requests to given http address.
+    # Redirect all incoming requests to staging.com address 
+    gor --input-raw :80 --output-http http://staging.com
+
+  -output-http-elasticsearch="": Send request and response stats to ElasticSearch:
+    gor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'
+
+  -output-http-header=[]: Inject additional headers to http reqest:
+    gor --input-raw :8080 --output-http staging.com --output-http-header 'User-Agent: Gor'
+    
+  -output-tcp=[]: Used for internal communication between Gor instances. Example: 
+    # Listen for requests on 80 port and forward them to other Gor instance on 28020 port
+    gor --input-raw :80 --output-tcp replay.local:28020
+    
+  -split-output=false: By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.
+```
 
 ## Building from source
 1. Setup standard Go environment http://golang.org/doc/code.html and ensure that $GOPATH environment variable properly set.
-2. `go get github.com/buger/gor`. 
+2. `go get github.com/buger/gor`.
 3. `cd $GOPATH/src/github.com/buger/gor`
-4. `go build gor.go` to get binary, or `go run gor.go` to build and run (useful for development)
+4. `go build ./bin/gor.go` to get binary, or `go run ./bin/gor.go` to build and run (useful for development)
 
 ## FAQ
 
 ### What OS are supported?
 For now only Linux based. *BSD (including MacOS is not supported yet, check https://github.com/buger/gor/issues/22 for details)
 
-### Why does the `gor listener` requires sudo or root access?
+### Why does the `--input-raw` requires sudo or root access?
 Listener works by sniffing traffic from a given port. It's accessible
 only by using sudo or root access.
 
-### Do you support all http request types?
-Yes. ~~Right now it supports only "GET" requests.~~
+### I'm getting 'too many open files' error
+Typical linux shell has a small open files soft limit at 1024. You can easily raise that when you do this before starting your gor replay process:
+  
+  ulimit -n 64000
+
+More about ulimit: http://blog.thecodingmachine.com/content/solving-too-many-open-files-exception-red5-or-any-other-application
+
+## Tuning
+
+To achieve the top most performance you should tune the source server system limits:
+
+    net.ipv4.tcp_max_tw_buckets = 65536
+    net.ipv4.tcp_tw_recycle = 1
+    net.ipv4.tcp_tw_reuse = 0
+    net.ipv4.tcp_max_syn_backlog = 131072
+    net.ipv4.tcp_syn_retries = 3
+    net.ipv4.tcp_synack_retries = 3
+    net.ipv4.tcp_retries1 = 3
+    net.ipv4.tcp_retries2 = 8
+    net.ipv4.tcp_rmem = 16384 174760 349520
+    net.ipv4.tcp_wmem = 16384 131072 262144
+    net.ipv4.tcp_mem = 262144 524288 1048576
+    net.ipv4.tcp_max_orphans = 65536
+    net.ipv4.tcp_fin_timeout = 10
+    net.ipv4.tcp_low_latency = 1
+    net.ipv4.tcp_syncookies = 0
+
 
 ## Contributing
 
@@ -104,10 +221,8 @@ Yes. ~~Right now it supports only "GET" requests.~~
 4. Push to the branch (git push origin my-new-feature)
 5. Create new Pull Request
 
-## TODO
+## Companies using Gor
 
-Use buffering for request throttling instead of simple rate limiting. 
-
-Better stats
-
-Optimize for load testing cases
+* http://granify.com
+* [GOV.UK](https://www.gov.uk) ([Government Digital Service](http://digital.cabinetoffice.gov.uk/))
+* To add your company drop me a line to github.com/buger or leonsbox@gmail.com

elasticsearch/elasticsearch.go

@@ -0,0 +1,153 @@
+package elasticsearch
+
+import (
+	"encoding/json"
+	"github.com/mattbaird/elastigo/api"
+	"github.com/mattbaird/elastigo/core"
+	"log"
+	"net/http"
+	"regexp"
+	"time"
+)
+
+type ESUriErorr struct{}
+
+func (e *ESUriErorr) Error() string {
+	return "Wrong ElasticSearch URL format. Expected to be: host:port/index_name"
+}
+
+type ESPlugin struct {
+	Active  bool
+	ApiPort string
+	Host    string
+	Index   string
+	indexor *core.BulkIndexer
+	done    chan bool
+}
+
+type ESRequestResponse struct {
+	ReqUrl               string         `json:"Req_URL"`
+	ReqMethod            string         `json:"Req_Method"`
+	ReqUserAgent         string         `json:"Req_User-Agent"`
+	ReqAcceptLanguage    string         `json:"Req_Accept-Language,omitempty"`
+	ReqAccept            string         `json:"Req_Accept,omitempty"`
+	ReqAcceptEncoding    string         `json:"Req_Accept-Encoding,omitempty"`
+	ReqIfModifiedSince   string         `json:"Req_If-Modified-Since,omitempty"`
+	ReqConnection        string         `json:"Req_Connection,omitempty"`
+	ReqCookies           []*http.Cookie `json:"Req_Cookies,omitempty"`
+	RespStatus           string         `json:"Resp_Status"`
+	RespStatusCode       int            `json:"Resp_Status-Code"`
+	RespProto            string         `json:"Resp_Proto,omitempty"`
+	RespContentLength    int64          `json:"Resp_Content-Length,omitempty"`
+	RespContentType      string         `json:"Resp_Content-Type,omitempty"`
+	RespTransferEncoding []string       `json:"Resp_Transfer-Encoding,omitempty"`
+	RespContentEncoding  string         `json:"Resp_Content-Encoding,omitempty"`
+	RespExpires          string         `json:"Resp_Expires,omitempty"`
+	RespCacheControl     string         `json:"Resp_Cache-Control,omitempty"`
+	RespVary             string         `json:"Resp_Vary,omitempty"`
+	RespSetCookie        string         `json:"Resp_Set-Cookie,omitempty"`
+	Rtt                  int64          `json:"RTT"`
+	Timestamp            time.Time
+}
+
+// Parse ElasticSearch URI
+//
+// Proper format is: host:port/index_name
+func parseURI(URI string) (err error, host string, port string, index string) {
+	rURI := regexp.MustCompile("(.+):([0-9]+)/(.+)")
+	match := rURI.FindAllStringSubmatch(URI, -1)
+
+	if len(match) == 0 {
+		err = new(ESUriErorr)
+	} else {
+		host = match[0][1]
+		port = match[0][2]
+		index = match[0][3]
+	}
+
+	return
+}
+
+func (p *ESPlugin) Init(URI string) {
+	var err error
+
+	err, p.Host, p.ApiPort, p.Index = parseURI(URI)
+
+	if err != nil {
+		log.Fatal("Can't initialize ElasticSearch plugin.", err)
+	}
+
+	api.Domain = p.Host
+	api.Port = p.ApiPort
+
+	p.indexor = core.NewBulkIndexerErrors(50, 60)
+	p.done = make(chan bool)
+	p.indexor.Run(p.done)
+
+	// Only start the ErrorHandler goroutine when in verbose mode
+	// no need to burn ressources otherwise
+	// go p.ErrorHandler()
+
+	log.Println("Initialized Elasticsearch Plugin")
+	return
+}
+
+func (p *ESPlugin) IndexerShutdown() {
+	p.done <- true
+	return
+}
+
+func (p *ESPlugin) ErrorHandler() {
+	for {
+		errBuf := <-p.indexor.ErrorChannel
+		log.Println(errBuf.Err)
+	}
+}
+
+func (p *ESPlugin) RttDurationToMs(d time.Duration) int64 {
+	sec := d / time.Second
+	nsec := d % time.Second
+	fl := float64(sec) + float64(nsec)*1e-6
+	return int64(fl)
+}
+
+func (p *ESPlugin) ResponseAnalyze(req *http.Request, resp *http.Response, start, stop time.Time) {
+	if resp == nil {
+		// nil http response - skipped elasticsearch export for this request
+		return
+	}
+	t := time.Now()
+	rtt := p.RttDurationToMs(stop.Sub(start))
+
+	esResp := ESRequestResponse{
+		ReqUrl:               req.URL.String(),
+		ReqMethod:            req.Method,
+		ReqUserAgent:         req.UserAgent(),
+		ReqAcceptLanguage:    req.Header.Get("Accept-Language"),
+		ReqAccept:            req.Header.Get("Accept"),
+		ReqAcceptEncoding:    req.Header.Get("Accept-Encoding"),
+		ReqIfModifiedSince:   req.Header.Get("If-Modified-Since"),
+		ReqConnection:        req.Header.Get("Connection"),
+		ReqCookies:           req.Cookies(),
+		RespStatus:           resp.Status,
+		RespStatusCode:       resp.StatusCode,
+		RespProto:            resp.Proto,
+		RespContentLength:    resp.ContentLength,
+		RespContentType:      resp.Header.Get("Content-Type"),
+		RespTransferEncoding: resp.TransferEncoding,
+		RespContentEncoding:  resp.Header.Get("Content-Encoding"),
+		RespExpires:          resp.Header.Get("Expires"),
+		RespCacheControl:     resp.Header.Get("Cache-Control"),
+		RespVary:             resp.Header.Get("Vary"),
+		RespSetCookie:        resp.Header.Get("Set-Cookie"),
+		Rtt:                  rtt,
+		Timestamp:            t,
+	}
+	j, err := json.Marshal(&esResp)
+	if err != nil {
+		log.Println(err)
+	} else {
+		p.indexor.Index(p.Index, "RequestResponse", "", "", &t, j)
+	}
+	return
+}

emitter.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"io"
+)
+
+func Start(stop chan int) {
+	for _, in := range Plugins.Inputs {
+		go CopyMulty(in, Plugins.Outputs...)
+	}
+
+	select {
+	case <-stop:
+		return
+	}
+}
+
+// Copy from 1 reader to multiple writers
+func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
+	buf := make([]byte, 32*1024)
+	wIndex := 0
+
+	for {
+		nr, er := src.Read(buf)
+		if nr > 0 && len(buf) < nr {
+			Debug("Sending", src, ": ", string(buf[0:nr]))
+
+			if Settings.splitOutput {
+				// Simple round robin
+				writers[wIndex].Write(buf[0:nr])
+
+				wIndex++
+
+				if wIndex >= len(writers) {
+					wIndex = 0
+				}
+			} else {
+				for _, dst := range writers {
+					dst.Write(buf[0:nr])
+				}
+			}
+
+		}
+		if er == io.EOF {
+			break
+		}
+		if er != nil {
+			err = er
+			break
+		}
+	}
+	return err
+}

emitter_test.go

@@ -0,0 +1,99 @@
+package main
+
+import (
+	"io"
+	"sync"
+	"sync/atomic"
+	"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)
+}
+
+func TestEmitterRoundRobin(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+
+	var counter1, counter2 int32
+
+	output1 := NewTestOutput(func(data []byte) {
+		atomic.AddInt32(&counter1, 1)
+		wg.Done()
+	})
+
+	output2 := NewTestOutput(func(data []byte) {
+		atomic.AddInt32(&counter2, 1)
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output1, output2}
+
+	Settings.splitOutput = true
+
+	go Start(quit)
+
+	for i := 0; i < 1000; i++ {
+		wg.Add(1)
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+
+	if counter1 == 0 || counter2 == 0 {
+		t.Errorf("Round robin should split traffic equally: %d vs %d", counter1, counter2)
+	}
+
+	Settings.splitOutput = false
+}
+
+func BenchmarkEmitter(b *testing.B) {
+	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)
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		input.EmitGET()
+	}
+
+	wg.Wait()
+	close(quit)
+}

gor.go

@@ -1,59 +1,56 @@
 // Gor is simple http traffic replication tool written in Go. Its main goal to replay traffic from production servers to staging and dev environments.
 // Now you can test your code on real user sessions in an automated and repeatable fashion.
-//
-// Gor consists of 2 parts: listener and replay servers.
-// Listener catch http traffic from given port in real-time and send it to replay server via UDP. Replay server forwards traffic to given address.
 package main
 
 import (
 	"flag"
 	"fmt"
-	"github.com/buger/gor/listener"
-	"github.com/buger/gor/replay"
 	"log"
 	"os"
+	"runtime/debug"
 	"runtime/pprof"
 	"time"
 )
 
-const (
-	VERSION = "0.3.5"
+var (
+	mode       string
+	cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
+	memprofile = flag.String("memprofile", "", "write memory profile to this file")
 )
 
-var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
-var memprofile = flag.String("memprofile", "", "write memory profile to this file")
-
 func main() {
+	// Don't exit on panic
 	defer func() {
 		if r := recover(); r != nil {
-			var ok bool
-			_, ok = r.(error)
-			if !ok {
-				fmt.Errorf("pkg: %v", r)
+			if _, ok := r.(error); !ok {
+				fmt.Printf("PANIC: pkg: %v %s \n", r, debug.Stack())
 			}
 		}
 	}()
 
 	fmt.Println("Version:", VERSION)
 
-	mode := "unknown"
-
-	if len(os.Args) > 1 {
-		mode = os.Args[1]
-	}
-
-	if mode != "listen" && mode != "replay" {
-		fmt.Println("Usage: \n\tgor listen -h\n\tgor replay -h")
-		return
-	}
-
-	// Remove mode attr
-	os.Args = append(os.Args[:1], os.Args[2:]...)
-
 	flag.Parse()
+	InitPlugins()
+
+	if len(Plugins.Inputs) == 0 || len(Plugins.Outputs) == 0 {
+		log.Fatal("Required at least 1 input and 1 output")
+	}
+
+	if *memprofile != "" {
+		profileMEM(*memprofile)
+	}
 
 	if *cpuprofile != "" {
-		f, err := os.Create(*cpuprofile)
+		profileCPU(*cpuprofile)
+	}
+
+	Start(nil)
+}
+
+func profileCPU(cpuprofile string) {
+	if cpuprofile != "" {
+		f, err := os.Create(cpuprofile)
 		if err != nil {
 			log.Fatal(err)
 		}
@@ -65,9 +62,11 @@ func main() {
 			log.Println("Stop profiling after 60 seconds")
 		})
 	}
+}
 
-	if *memprofile != "" {
-		f, err := os.Create(*memprofile)
+func profileMEM(memprofile string) {
+	if memprofile != "" {
+		f, err := os.Create(memprofile)
 		if err != nil {
 			log.Fatal(err)
 		}
@@ -76,11 +75,4 @@ func main() {
 			f.Close()
 		})
 	}
-
-	switch mode {
-	case "listen":
-		listener.Run()
-	case "replay":
-		replay.Run()
-	}
 }

input_dummy.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"time"
+)
+
+type DummyInput struct {
+	data chan []byte
+}
+
+func NewDummyInput(options string) (di *DummyInput) {
+	di = new(DummyInput)
+	di.data = make(chan []byte)
+
+	go di.emit()
+
+	return
+}
+
+func (i *DummyInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *DummyInput) emit() {
+	ticker := time.NewTicker(time.Second)
+
+	for {
+		select {
+		case <-ticker.C:
+			i.data <- []byte("GET / HTTP/1.1\r\n\r\n")
+		}
+	}
+}
+
+func (i *DummyInput) String() string {
+	return "Dummy Input"
+}

input_file.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"encoding/gob"
+	"log"
+	"os"
+	"time"
+)
+
+type FileInput struct {
+	data    chan []byte
+	path    string
+	decoder *gob.Decoder
+}
+
+func NewFileInput(path string) (i *FileInput) {
+	i = new(FileInput)
+	i.data = make(chan []byte)
+	i.path = path
+	i.Init(path)
+
+	go i.emit()
+
+	return
+}
+
+func (i *FileInput) Init(path string) {
+	file, err := os.Open(path)
+
+	if err != nil {
+		log.Fatal(i, "Cannot open file %q. Error: %s", path, err)
+	}
+
+	i.decoder = gob.NewDecoder(file)
+}
+
+func (i *FileInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *FileInput) String() string {
+	return "File input: " + i.path
+}
+
+func (i *FileInput) emit() {
+	var lastTime int64
+
+	for {
+		raw := new(RawRequest)
+		err := i.decoder.Decode(raw)
+
+		if err != nil {
+			return
+		}
+
+		if lastTime != 0 {
+			time.Sleep(time.Duration(raw.Timestamp - lastTime))
+			lastTime = raw.Timestamp
+		}
+
+		i.data <- raw.Request
+	}
+}

input_raw.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	raw "github.com/buger/gor/raw_socket_listener"
+	"log"
+	"net"
+	"strings"
+)
+
+type RAWInput struct {
+	data    chan []byte
+	address string
+}
+
+func NewRAWInput(address string) (i *RAWInput) {
+	i = new(RAWInput)
+	i.data = make(chan []byte)
+	i.address = address
+
+	go i.listen(address)
+
+	return
+}
+
+func (i *RAWInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *RAWInput) listen(address string) {
+	address = strings.Replace(address, "[::]", "127.0.0.1", -1)
+
+	host, port, err := net.SplitHostPort(address)
+
+	if err != nil {
+		log.Fatal("input-raw: error while parsing address", err)
+	}
+
+	listener := raw.NewListener(host, port)
+
+	for {
+		// Receiving TCPMessage object
+		m := listener.Receive()
+
+		i.data <- m.Bytes()
+	}
+}
+
+func (i *RAWInput) String() string {
+	return "RAW Socket input: " + i.address
+}

input_raw_test.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+)
+
+func TestRAWInput(t *testing.T) {
+
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	listener := startHTTP(func(req *http.Request) {})
+
+	input := NewRAWInput(listener.Addr().String())
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	address := strings.Replace(listener.Addr().String(), "[::]", "127.0.0.1", -1)
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		res, _ := http.Get("http://" + address)
+		res.Body.Close()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}

input_tcp.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"log"
+	"net"
+)
+
+// Can be tested using nc tool:
+//    echo "asdad" | nc 127.0.0.1 27017
+//
+type TCPInput struct {
+	data     chan []byte
+	address  string
+	listener net.Listener
+}
+
+func NewTCPInput(address string) (i *TCPInput) {
+	i = new(TCPInput)
+	i.data = make(chan []byte)
+	i.address = address
+
+	i.listen(address)
+
+	return
+}
+
+func (i *TCPInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *TCPInput) listen(address string) {
+	listener, err := net.Listen("tcp", address)
+	i.listener = listener
+
+	if err != nil {
+		log.Fatal("Can't start:", err)
+	}
+
+	go func() {
+		for {
+			conn, err := listener.Accept()
+
+			if err != nil {
+				log.Println("Error while Accept()", err)
+				continue
+			}
+
+			go i.handleConnection(conn)
+		}
+	}()
+}
+
+func scanBytes(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	if atEOF && len(data) == 0 {
+		return 0, nil, nil
+	}
+
+	// Search for ¶ symbol
+	if i := bytes.IndexByte(data, 194); i >= 0 {
+		if len(data) > i+1 && data[i+1] == 182 {
+			// We have a full newline-terminated line.
+			return i + 2, data[0:i], nil
+		}
+	}
+	// If we're at EOF, we have a final, non-terminated line. Return it.
+	if atEOF {
+		return len(data), data, nil
+	}
+	// Request more data.
+	return 0, nil, nil
+}
+
+func (i *TCPInput) handleConnection(conn net.Conn) {
+	defer conn.Close()
+
+	scanner := bufio.NewScanner(conn)
+
+	scanner.Split(scanBytes)
+
+	for scanner.Scan() {
+		i.data <- scanner.Bytes()
+	}
+}
+
+func (i *TCPInput) String() string {
+	return "TCP input: " + i.address
+}

input_tcp_test.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"io"
+	"log"
+	"net"
+	"sync"
+	"testing"
+)
+
+func TestTCPInput(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTCPInput(":0")
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	tcpAddr, err := net.ResolveTCPAddr("tcp", input.listener.Addr().String())
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	conn, err := net.DialTCP("tcp", nil, tcpAddr)
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	msg := []byte("GET / HTTP/1.1\r\n\r\n")
+
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		conn.Write(msg)
+		conn.Write([]byte("¶"))
+	}
+
+	wg.Wait()
+
+	close(quit)
+}
+
+func BenchmarkTCPInput(b *testing.B) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTCPInput(":0")
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	tcpAddr, err := net.ResolveTCPAddr("tcp", input.listener.Addr().String())
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	var connections []net.Conn
+
+	// Creating simple pool of workers, same as output_tcp have
+	dataChan := make(chan []byte, 1000)
+
+	for i := 0; i < 10; i++ {
+		conn, _ := net.DialTCP("tcp", nil, tcpAddr)
+		connections = append(connections, conn)
+
+		go func(conn net.Conn) {
+			for {
+				data := <-dataChan
+
+				conn.Write(data)
+				conn.Write([]byte("¶"))
+			}
+		}(conn)
+	}
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	msg := []byte("GET / HTTP/1.1\r\n\r\n")
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		dataChan <- msg
+	}
+
+	wg.Wait()
+
+	for _, conn := range connections {
+		conn.Close()
+	}
+
+	close(quit)
+}

integration_test.go

@@ -1,200 +0,0 @@
-package main
-
-import (
-	"testing"
-
-	"github.com/buger/gor/listener"
-	"github.com/buger/gor/replay"
-
-	"time"
-
-	"fmt"
-	"net/http"
-	"strconv"
-
-	"sync/atomic"
-)
-
-func isEqual(t *testing.T, a interface{}, b interface{}) {
-	if a != b {
-		t.Error("Original and Replayed request not match\n", a, "!=", b)
-	}
-}
-
-var envs int
-
-type Env struct {
-	Verbose bool
-
-	ListenHandler http.HandlerFunc
-	ReplayHandler http.HandlerFunc
-
-	ReplayLimit   int
-	ListenerLimit int
-}
-
-func (e *Env) start() (p int) {
-	p = 50000 + envs*10
-
-	go e.startHTTP(p, http.HandlerFunc(e.ListenHandler))
-	go e.startHTTP(p+2, http.HandlerFunc(e.ReplayHandler))
-	go e.startListener(p, p+1)
-	go e.startReplay(p+1, p+2)
-
-	// Time to start http and gor instances
-	time.Sleep(time.Millisecond * 100)
-
-	envs++
-
-	return
-}
-
-func (e *Env) startListener(port int, replayPort int) {
-	listener.Settings.Verbose = e.Verbose
-	listener.Settings.Address = "127.0.0.1"
-	listener.Settings.ReplayAddress = "127.0.0.1:" + strconv.Itoa(replayPort)
-	listener.Settings.Port = port
-
-	if e.ListenerLimit != 0 {
-		listener.Settings.ReplayAddress += "|" + strconv.Itoa(e.ListenerLimit)
-	}
-
-	listener.Run()
-}
-
-func (e *Env) startReplay(port int, forwardPort int) {
-	replay.Settings.Verbose = e.Verbose
-	replay.Settings.Host = "127.0.0.1"
-	replay.Settings.ForwardAddress = "127.0.0.1:" + strconv.Itoa(forwardPort)
-	replay.Settings.Port = port
-
-	if e.ReplayLimit != 0 {
-		replay.Settings.ForwardAddress += "|" + strconv.Itoa(e.ReplayLimit)
-	}
-
-	replay.Run()
-}
-
-func (e *Env) startHTTP(port int, handler http.Handler) {
-	err := http.ListenAndServe(":"+strconv.Itoa(port), handler)
-
-	if err != nil {
-		fmt.Println("Error while starting http server:", err)
-	}
-}
-
-func getRequest(port int) *http.Request {
-	req, _ := http.NewRequest("GET", "http://localhost:"+strconv.Itoa(port)+"/test", nil)
-	ck1 := new(http.Cookie)
-	ck1.Name = "test"
-	ck1.Value = "value"
-
-	req.AddCookie(ck1)
-
-	return req
-}
-
-func TestReplay(t *testing.T) {
-	var request *http.Request
-	received := make(chan int)
-
-	listenHandler := func(w http.ResponseWriter, r *http.Request) {
-		http.Error(w, "OK", http.StatusAccepted)
-	}
-
-	replayHandler := func(w http.ResponseWriter, r *http.Request) {
-		isEqual(t, r.URL.Path, request.URL.Path)
-
-		if len(r.Cookies()) > 0 {
-			isEqual(t, r.Cookies()[0].Value, request.Cookies()[0].Value)
-		} else {
-			t.Error("Cookies should not be blank")
-		}
-
-		http.Error(w, "OK", http.StatusAccepted)
-
-		if t.Failed() {
-			fmt.Println("\nReplayed:", r, "\nOriginal:", request)
-		}
-
-		received <- 1
-	}
-
-	env := &Env{
-		Verbose:       true,
-		ListenHandler: listenHandler,
-		ReplayHandler: replayHandler,
-	}
-	p := env.start()
-
-	request = getRequest(p)
-
-	_, err := http.DefaultClient.Do(request)
-
-	if err != nil {
-		t.Error("Can't make request", err)
-	}
-
-	select {
-	case <-received:
-	case <-time.After(time.Second):
-		t.Error("Timeout error")
-	}
-}
-
-func rateLimitEnv(replayLimit int, listenerLimit int, connCount int) int32 {
-	var processed int32
-
-	listenHandler := func(w http.ResponseWriter, r *http.Request) {
-		http.Error(w, "OK", http.StatusAccepted)
-	}
-
-	replayHandler := func(w http.ResponseWriter, r *http.Request) {
-		atomic.AddInt32(&processed, 1)
-		http.Error(w, "OK", http.StatusAccepted)
-	}
-
-	env := &Env{
-		ListenHandler: listenHandler,
-		ReplayHandler: replayHandler,
-		ReplayLimit:   replayLimit,
-		ListenerLimit: listenerLimit,
-	}
-
-	p := env.start()
-	req := getRequest(p)
-
-	for i := 0; i < connCount; i++ {
-		go func() {
-			http.DefaultClient.Do(req)
-		}()
-	}
-
-	time.Sleep(time.Millisecond * 500)
-
-	return processed
-}
-
-func TestWithoutReplayRateLimit(t *testing.T) {
-	processed := rateLimitEnv(0, 0, 10)
-
-	if processed != 10 {
-		t.Error("It should forward all requests without rate-limiting", processed)
-	}
-}
-
-func TestReplayRateLimit(t *testing.T) {
-	processed := rateLimitEnv(5, 0, 10)
-
-	if processed != 5 {
-		t.Error("It should forward only 5 requests with rate-limiting", processed)
-	}
-}
-
-func TestListenerRateLimit(t *testing.T) {
-	processed := rateLimitEnv(0, 3, 100)
-
-	if processed != 3 {
-		t.Error("It should forward only 3 requests with rate-limiting", processed)
-	}
-}

limiter.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"time"
+)
+
+type Limiter struct {
+	writer io.Writer
+	limit  int
+
+	currentRPS  int
+	currentTime int64
+}
+
+func NewLimiter(writer io.Writer, limit int) (l *Limiter) {
+	l = new(Limiter)
+	l.limit = limit
+	l.writer = writer
+	l.currentTime = time.Now().UnixNano()
+
+	return
+}
+
+func (l *Limiter) Write(data []byte) (n int, err error) {
+	if (time.Now().UnixNano() - l.currentTime) > time.Second.Nanoseconds() {
+		l.currentTime = time.Now().UnixNano()
+		l.currentRPS = 0
+	}
+
+	if l.currentRPS >= l.limit {
+		return 0, nil
+	}
+
+	n, err = l.writer.Write(data)
+
+	l.currentRPS++
+
+	return
+}
+
+func (l *Limiter) String() string {
+	return fmt.Sprintf("Limiting %s to: %d", l.writer, l.limit)
+}

limiter_test.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"io"
+	"sync"
+	"testing"
+)
+
+func TestLimiter(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+	output := NewLimiter(NewTestOutput(func(data []byte) {
+		wg.Done()
+	}), 10)
+	wg.Add(10)
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}

listener/listener.go

@@ -1,106 +0,0 @@
-// Listener capture TCP traffic using RAW SOCKETS.
-// Note: it requires sudo or root access.
-//
-// Rigt now it suport only HTTP
-package listener
-
-import (
-	"bufio"
-	"bytes"
-	"fmt"
-	"log"
-	"net"
-	"net/http"
-	"os"
-	"strconv"
-	"time"
-)
-
-// Enable debug logging only if "--verbose" flag passed
-func Debug(v ...interface{}) {
-	if Settings.Verbose {
-		log.Println(v...)
-	}
-}
-
-func ReplayServer() (conn net.Conn, err error) {
-	// Connection to reaplay server
-	conn, err = net.Dial("tcp", Settings.ReplayServer())
-
-	if err != nil {
-		log.Println("Connection error ", err, Settings.ReplayAddress)
-	}
-
-	return
-}
-
-// Because its sub-program, Run acts as `main`
-func Run() {
-	if os.Getuid() != 0 {
-		fmt.Println("Please start the listener as root or sudo!")
-		fmt.Println("This is required since listener sniff traffic on given port.")
-		os.Exit(1)
-	}
-
-	fmt.Println("Listening for HTTP traffic on", Settings.Address+":"+strconv.Itoa(Settings.Port))
-	fmt.Println("Forwarding requests to replay server:", Settings.ReplayServer(), "Limit:", Settings.ReplayLimit)
-
-	// Sniffing traffic from given address
-	listener := RAWTCPListen(Settings.Address, Settings.Port)
-
-	currentTime := time.Now().UnixNano()
-	currentRPS := 0
-
-	for {
-		// Receiving TCPMessage object
-		m := listener.Receive()
-
-		if Settings.ReplayLimit != 0 {
-			if (time.Now().UnixNano() - currentTime) > time.Second.Nanoseconds() {
-				currentTime = time.Now().UnixNano()
-				currentRPS = 0
-			}
-
-			if currentRPS >= Settings.ReplayLimit {
-				continue
-			}
-
-			currentRPS++
-		}
-
-		go sendMessage(m)
-	}
-}
-
-func sendMessage(m *TCPMessage) {
-	conn, err := ReplayServer()
-
-	if err != nil {
-		log.Println("Failed to send message. Replay server not respond.")
-		return
-	} else {
-		defer conn.Close()
-	}
-
-	// For debugging purpose
-	// Usually request parsing happens in replay part
-	if Settings.Verbose {
-		buf := bytes.NewBuffer(m.Bytes())
-		reader := bufio.NewReader(buf)
-
-		request, err := http.ReadRequest(reader)
-
-		if err != nil {
-			Debug("Error while parsing request:", err, string(m.Bytes()))
-		} else {
-			request.ParseMultipartForm(32 << 20)
-			Debug("Forwarding request:", request)
-		}
-	}
-
-	_, err = conn.Write(m.Bytes())
-
-	if err != nil {
-		log.Println("Error while sending requests", err)
-	}
-}

listener/listener_test.go

@@ -1,45 +0,0 @@
-package listener
-
-import (
-	"bytes"
-	"fmt"
-	"net"
-	"testing"
-)
-
-func getTCPMessage() (msg *TCPMessage) {
-	packet := &TCPPacket{Data: []byte("GET /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\n")}
-
-	return &TCPMessage{packets: []*TCPPacket{packet}}
-}
-
-func mockReplayServer() (listener net.Listener) {
-	listener, _ = net.Listen("tcp", "127.0.0.1:0")
-
-	Settings.ReplayAddress = listener.Addr().String()
-
-	fmt.Println(listener.Addr().String())
-
-	return
-}
-
-func TestSendMessage(t *testing.T) {
-	Settings.Verbose = true
-
-	listener := mockReplayServer()
-
-	msg := getTCPMessage()
-
-	sendMessage(msg)
-
-	conn, _ := listener.Accept()
-	defer conn.Close()
-
-	buf := make([]byte, 1024)
-	n, _ := conn.Read(buf)
-	buf = buf[0:n]
-
-	if bytes.Compare(buf, msg.Bytes()) != 0 {
-		t.Errorf("Original and reveived requests does not match")
-	}
-}

listener/raw_tcp_listener.go

@@ -1,159 +0,0 @@
-package listener
-
-import (
-	"encoding/binary"
-	"log"
-	"net"
-)
-
-// Capture traffic from socket using RAW_SOCKET's
-// http://en.wikipedia.org/wiki/Raw_socket
-//
-// RAW_SOCKET allow you listen for traffic on any port (e.g. sniffing) because they operate on IP level.
-// Ports is TCP feature, same as flow control, reliable transmission and etc.
-// Since we can't use default TCP libraries RAWTCPLitener implements own TCP layer
-// TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go
-type RAWTCPListener struct {
-	messages []*TCPMessage // buffer of TCPMessages waiting to be send
-
-	c_packets  chan *TCPPacket
-	c_messages chan *TCPMessage // Messages ready to be send to client
-
-	c_del_message chan *TCPMessage // Used for notifications about completed or expired messages
-
-	addr string // IP to listen
-	port int    // Port to listen
-}
-
-func RAWTCPListen(addr string, port int) (listener *RAWTCPListener) {
-	listener = &RAWTCPListener{}
-
-	listener.c_packets = make(chan *TCPPacket)
-	listener.c_messages = make(chan *TCPMessage)
-	listener.c_del_message = make(chan *TCPMessage)
-
-	listener.addr = addr
-	listener.port = port
-
-	go listener.listen()
-	go listener.readRAWSocket()
-
-	return
-}
-
-func (t *RAWTCPListener) listen() {
-	for {
-		select {
-		// If message ready for deletion it means that its also complete or expired by timeout
-		case message := <-t.c_del_message:
-			t.c_messages <- message
-			t.deleteMessage(message)
-
-		// We need to use channgels to process each packet to avoid data races
-		case packet := <-t.c_packets:
-			t.processTCPPacket(packet)
-		}
-	}
-}
-
-// Deleting messages that came from t.c_del_message channel
-func (t *RAWTCPListener) deleteMessage(message *TCPMessage) bool {
-	var idx int = -1
-
-	// Searching for given message in messages buffer
-	for i, m := range t.messages {
-		if m.Ack == message.Ack {
-			idx = i
-			break
-		}
-	}
-
-	if idx == -1 {
-		return false
-	}
-
-	// Delete element from array
-	// Note: that this version for arrays that consist of pointers
-	// https://code.google.com/p/go-wiki/wiki/SliceTricks
-	copy(t.messages[idx:], t.messages[idx+1:])
-	t.messages[len(t.messages)-1] = nil // Ensure that value will be garbage-collected.
-	t.messages = t.messages[:len(t.messages)-1]
-
-	return true
-}
-
-func (t *RAWTCPListener) readRAWSocket() {
-	conn, e := net.ListenPacket("ip4:tcp", t.addr)
-	defer conn.Close()
-
-	if e != nil {
-		log.Fatal(e)
-	}
-
-	buf := make([]byte, 4096*2)
-
-	for {
-		// Note: ReadFrom receive messages without IP header
-		n, _, err := conn.ReadFrom(buf)
-
-		if err != nil {
-			Debug("Error:", err)
-		}
-
-		if n > 0 {
-			// To avoid full packet parsing every time, we manually parsing values needed for packet filtering
-			// http://en.wikipedia.org/wiki/Transmission_Control_Protocol
-			dest_port := binary.BigEndian.Uint16(buf[2:4])
-
-			// Because RAW_SOCKET can't be bound to port, we have to control it by ourself
-			if int(dest_port) == t.port {
-				// Check TCPPacket code for more description
-				flags := binary.BigEndian.Uint16(buf[12:14]) & 0x1FF
-				f_psh := (flags & TCP_PSH) != 0
-
-				// We need only packets with data inside
-				// TCP PSH flag indicate that packet have data inside
-				if f_psh {
-					// We should create new buffer because go slices is pointers. So buffer data shoud be immutable.
-					new_buf := make([]byte, n)
-					copy(new_buf, buf[:n])
-
-					// To avoid socket locking processing packet in new goroutine
-					go func(buf []byte) {
-						packet := NewTCPPacket(new_buf)
-						t.c_packets <- packet
-					}(new_buf)
-				}
-			}
-		}
-	}
-}
-
-// Trying to add packet to existing message or creating new message
-//
-// For TCP message unique id is Acknowledgment number (see tcp_packet.go)
-func (t *RAWTCPListener) processTCPPacket(packet *TCPPacket) {
-	var message *TCPMessage
-
-	// Searching for message with same Ack
-	for _, msg := range t.messages {
-		if msg.Ack == packet.Ack {
-			message = msg
-			break
-		}
-	}
-
-	if message == nil {
-		// We sending c_del_message channel, so message object can communicate with Listener and notify it if message completed
-		message = NewTCPMessage(packet.Ack, t.c_del_message)
-
-		t.messages = append(t.messages, message)
-	}
-
-	// Adding packet to message
-	message.c_packets <- packet
-}
-
-func (t *RAWTCPListener) Receive() *TCPMessage {
-	return <-t.c_messages
-}

listener/settings.go

@@ -1,52 +0,0 @@
-package listener
-
-import (
-	"flag"
-	"os"
-	"strconv"
-	"strings"
-)
-
-const (
-	defaultPort    = 80
-	defaultAddress = "0.0.0.0"
-
-	defaultReplayAddress = "localhost:28020"
-)
-
-type ListenerSettings struct {
-	Port    int
-	Address string
-
-	ReplayAddress string
-
-	ReplayLimit int
-
-	Verbose bool
-}
-
-var Settings ListenerSettings = ListenerSettings{}
-
-func (s *ListenerSettings) ReplayServer() string {
-	host_info := strings.Split(s.ReplayAddress, "|")
-
-	if len(host_info) > 1 {
-		s.ReplayLimit, _ = strconv.Atoi(host_info[1])
-	}
-
-	return host_info[0]
-}
-
-func init() {
-	if len(os.Args) < 2 || os.Args[1] != "listen" {
-		return
-	}
-
-	flag.IntVar(&Settings.Port, "p", defaultPort, "Specify the http server port whose traffic you want to capture")
-
-	flag.StringVar(&Settings.Address, "ip", defaultAddress, "Specifi IP address to listen")
-
-	flag.StringVar(&Settings.ReplayAddress, "r", defaultReplayAddress, "Address of replay server.")
-
-	flag.BoolVar(&Settings.Verbose, "verbose", false, "Log requests")
-}

output_dummy.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"fmt"
+)
+
+type DummyOutput struct {
+}
+
+func NewDummyOutput(options string) (di *DummyOutput) {
+	di = new(DummyOutput)
+
+	return
+}
+
+func (i *DummyOutput) Write(data []byte) (int, error) {
+	fmt.Println("Writing message: ", data)
+
+	return len(data), nil
+}
+
+func (i *DummyOutput) String() string {
+	return "Dummy Output"
+}

output_file.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"encoding/gob"
+	"log"
+	"os"
+	"time"
+)
+
+type RawRequest struct {
+	Timestamp int64
+	Request   []byte
+}
+
+type FileOutput struct {
+	path    string
+	encoder *gob.Encoder
+	file    *os.File
+}
+
+func NewFileOutput(path string) (o *FileOutput) {
+	o = new(FileOutput)
+	o.path = path
+	o.Init(path)
+
+	return
+}
+
+func (o *FileOutput) Init(path string) {
+	var err error
+
+	o.file, err = os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
+
+	if err != nil {
+		log.Fatal(o, "Cannot open file %q. Error: %s", path, err)
+	}
+
+	o.encoder = gob.NewEncoder(o.file)
+}
+
+func (o *FileOutput) Write(data []byte) (n int, err error) {
+	raw := RawRequest{time.Now().UnixNano(), data}
+
+	o.encoder.Encode(raw)
+
+	return len(data), nil
+}
+
+func (o *FileOutput) String() string {
+	return "File output: " + o.path
+}

output_file_test.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"io"
+	"sync"
+	"testing"
+)
+
+func TestFileOutput(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+	output := NewFileOutput("/tmp/test_requests.gor")
+
+	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()
+	}
+	close(quit)
+
+	quit = make(chan int)
+
+	input2 := NewFileInput("/tmp/test_requests.gor")
+	output2 := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input2}
+	Plugins.Outputs = []io.Writer{output2}
+
+	go Start(quit)
+
+	wg.Wait()
+	close(quit)
+}

output_http.go

@@ -0,0 +1,151 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	es "github.com/buger/gor/elasticsearch"
+	"io"
+	"log"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type RedirectNotAllowed struct{}
+
+func (e *RedirectNotAllowed) Error() string {
+	return "Redirects not allowed"
+}
+
+// customCheckRedirect disables redirects https://github.com/buger/gor/pull/15
+func customCheckRedirect(req *http.Request, via []*http.Request) error {
+	if len(via) >= 0 {
+		return new(RedirectNotAllowed)
+	}
+	return nil
+}
+
+// ParseRequest in []byte returns a http request or an error
+func ParseRequest(data []byte) (request *http.Request, err error) {
+	buf := bytes.NewBuffer(data)
+	reader := bufio.NewReader(buf)
+
+	request, err = http.ReadRequest(reader)
+
+	return
+}
+
+type HTTPOutput struct {
+	address string
+	limit   int
+	buf     chan []byte
+
+	headers HTTPHeaders
+	methods HTTPMethods
+
+	elasticSearch *es.ESPlugin
+}
+
+func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, elasticSearchAddr string) io.Writer {
+	o := new(HTTPOutput)
+
+	optionsArr := strings.Split(options, "|")
+	address := optionsArr[0]
+
+	if !strings.HasPrefix(address, "http") {
+		address = "http://" + address
+	}
+
+	o.address = address
+	o.headers = headers
+	o.methods = methods
+
+	o.buf = make(chan []byte, 100)
+
+	if elasticSearchAddr != "" {
+		o.elasticSearch = new(es.ESPlugin)
+		o.elasticSearch.Init(elasticSearchAddr)
+	}
+
+	if len(optionsArr) > 1 {
+		o.limit, _ = strconv.Atoi(optionsArr[1])
+	}
+
+	for i := 0; i < 10; i++ {
+		go o.worker(i)
+	}
+
+	if o.limit > 0 {
+		return NewLimiter(o, o.limit)
+	} else {
+		return o
+	}
+}
+
+func (o *HTTPOutput) worker(n int) {
+	client := &http.Client{
+		CheckRedirect: customCheckRedirect,
+	}
+
+	for {
+		data := <-o.buf
+		o.sendRequest(client, data)
+	}
+}
+
+func (o *HTTPOutput) Write(data []byte) (n int, err error) {
+	buf := make([]byte, len(data))
+	copy(buf, data)
+
+	o.buf <- buf
+
+	return len(data), nil
+}
+
+func (o *HTTPOutput) sendRequest(client *http.Client, data []byte) {
+	request, err := ParseRequest(data)
+
+	if err != nil {
+		log.Println("Can not parse request", string(data), err)
+		return
+	}
+
+	if len(o.methods) > 0 && !o.methods.Contains(request.Method) {
+		return
+	}
+
+	// Change HOST of original request
+	URL := o.address + request.URL.Path + "?" + request.URL.RawQuery
+
+	request.RequestURI = ""
+	request.URL, _ = url.ParseRequestURI(URL)
+
+	for _, header := range o.headers {
+		request.Header.Set(header.Name, header.Value)
+	}
+
+	start := time.Now()
+	resp, err := client.Do(request)
+	stop := time.Now()
+
+	// We should not count Redirect as errors
+	if _, ok := err.(*RedirectNotAllowed); ok {
+		err = nil
+	}
+
+	if err == nil {
+		defer resp.Body.Close()
+	} else {
+		log.Println("Request error:", err)
+	}
+
+	if o.elasticSearch != nil {
+		o.elasticSearch.ResponseAnalyze(request, resp, start, stop)
+	}
+}
+
+func (o *HTTPOutput) String() string {
+	return "HTTP output: " + o.address
+}

output_http_test.go

@@ -0,0 +1,93 @@
+package main
+
+import (
+	"io"
+	"net"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+)
+
+func startHTTP(cb func(*http.Request)) net.Listener {
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		go cb(r)
+	})
+
+	listener, _ := net.Listen("tcp", ":0")
+
+	go http.Serve(listener, handler)
+
+	return listener
+}
+
+func TestHTTPOutput(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+
+	headers := HTTPHeaders{HTTPHeader{"User-Agent", "Gor"}}
+	methods := HTTPMethods{"GET", "PUT", "POST"}
+
+	listener := startHTTP(func(req *http.Request) {
+		if req.Header.Get("User-Agent") != "Gor" {
+			t.Error("Wrong header")
+		}
+
+		if req.Method == "OPTIONS" {
+			t.Error("Wrong method")
+		}
+
+		wg.Done()
+	})
+
+	output := NewHTTPOutput(listener.Addr().String(), headers, methods, "")
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		wg.Add(2)
+		input.EmitPOST()
+		input.EmitOPTIONS()
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}
+
+func BenchmarkHTTPOutput(b *testing.B) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+
+	headers := HTTPHeaders{HTTPHeader{"User-Agent", "Gor"}}
+	methods := HTTPMethods{"GET", "PUT", "POST"}
+
+	listener := startHTTP(func(req *http.Request) {
+		time.Sleep(50 * time.Millisecond)
+		wg.Done()
+	})
+
+	output := NewHTTPOutput(listener.Addr().String(), headers, methods, "")
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		input.EmitPOST()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}

output_tcp.go

@@ -0,0 +1,69 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"strconv"
+	"strings"
+)
+
+type TCPOutput struct {
+	address string
+	limit   int
+	buf     chan []byte
+}
+
+func NewTCPOutput(options string) io.Writer {
+	o := new(TCPOutput)
+
+	optionsArr := strings.Split(options, "|")
+	o.address = optionsArr[0]
+
+	o.buf = make(chan []byte, 100)
+
+	if len(optionsArr) > 1 {
+		o.limit, _ = strconv.Atoi(optionsArr[1])
+	}
+
+	for i := 0; i < 10; i++ {
+		go o.worker()
+	}
+
+	if o.limit > 0 {
+		return NewLimiter(o, o.limit)
+	} else {
+		return o
+	}
+}
+
+func (o *TCPOutput) worker() {
+	conn, _ := o.connect(o.address)
+	defer conn.Close()
+
+	for {
+		conn.Write(<-o.buf)
+		conn.Write([]byte("¶"))
+	}
+}
+
+func (o *TCPOutput) Write(data []byte) (n int, err error) {
+	o.buf <- data
+
+	return len(data), nil
+}
+
+func (o *TCPOutput) connect(address string) (conn net.Conn, err error) {
+	conn, err = net.Dial("tcp", address)
+
+	if err != nil {
+		log.Println("Connection error ", err, o.address)
+	}
+
+	return
+}
+
+func (o *TCPOutput) String() string {
+	return fmt.Sprintf("TCP output %s, limit: %d", o.address, o.limit)
+}

output_tcp_test.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"bufio"
+	"io"
+	"log"
+	"net"
+	"sync"
+	"testing"
+)
+
+func TestTCPOutput(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	listener := startTCP(func(data []byte) {
+		wg.Done()
+	})
+	input := NewTestInput()
+	output := NewTCPOutput(listener.Addr().String())
+
+	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(cb func([]byte)) net.Listener {
+	listener, err := net.Listen("tcp", ":0")
+
+	if err != nil {
+		log.Fatal("Can't start:", err)
+	}
+
+	go func() {
+		for {
+			conn, _ := listener.Accept()
+
+			go func() {
+				scanner := bufio.NewScanner(conn)
+
+				scanner.Split(scanBytes)
+
+				for scanner.Scan() {
+					cb(scanner.Bytes())
+				}
+
+				conn.Close()
+			}()
+		}
+	}()
+
+	return listener
+}
+
+func BenchmarkTCPOutput(b *testing.B) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	listener := startTCP(func(data []byte) {
+		wg.Done()
+	})
+	input := NewTestInput()
+	output := NewTCPOutput(listener.Addr().String())
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		wg.Add(1)
+		input.EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}

plugins.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"io"
+)
+
+type InOutPlugins struct {
+	Inputs  []io.Reader
+	Outputs []io.Writer
+}
+
+var Plugins *InOutPlugins = new(InOutPlugins)
+
+func InitPlugins() {
+	for _, options := range Settings.inputDummy {
+		Plugins.Inputs = append(Plugins.Inputs, NewDummyInput(options))
+	}
+
+	for _, options := range Settings.outputDummy {
+		Plugins.Outputs = append(Plugins.Outputs, NewDummyOutput(options))
+	}
+
+	for _, options := range Settings.inputRAW {
+		Plugins.Inputs = append(Plugins.Inputs, NewRAWInput(options))
+	}
+
+	for _, options := range Settings.inputTCP {
+		Plugins.Inputs = append(Plugins.Inputs, NewTCPInput(options))
+	}
+
+	for _, options := range Settings.outputTCP {
+		Plugins.Outputs = append(Plugins.Outputs, NewTCPOutput(options))
+	}
+
+	for _, options := range Settings.inputFile {
+		Plugins.Inputs = append(Plugins.Inputs, NewFileInput(options))
+	}
+
+	for _, options := range Settings.outputFile {
+		Plugins.Outputs = append(Plugins.Outputs, NewFileOutput(options))
+	}
+
+	for _, options := range Settings.outputHTTP {
+		Plugins.Outputs = append(Plugins.Outputs, NewHTTPOutput(options, Settings.outputHTTPHeaders, Settings.outputHTTPMethods, Settings.outputHTTPElasticSearch))
+	}
+}

raw_socket_listener/listener.go

@@ -0,0 +1,141 @@
+package raw_socket
+
+import (
+	"encoding/binary"
+	"log"
+	"net"
+	"strconv"
+)
+
+// Capture traffic from socket using RAW_SOCKET's
+// http://en.wikipedia.org/wiki/Raw_socket
+//
+// RAW_SOCKET allow you listen for traffic on any port (e.g. sniffing) because they operate on IP level.
+// Ports is TCP feature, same as flow control, reliable transmission and etc.
+// Since we can't use default TCP libraries RAWTCPLitener implements own TCP layer
+// TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go
+type Listener struct {
+	messages map[string]*TCPMessage // buffer of TCPMessages waiting to be send
+
+	c_packets  chan *TCPPacket
+	c_messages chan *TCPMessage // Messages ready to be send to client
+
+	c_del_message chan *TCPMessage // Used for notifications about completed or expired messages
+
+	addr string // IP to listen
+	port int    // Port to listen
+}
+
+// RAWTCPListen creates a listener to capture traffic from RAW_SOCKET
+func NewListener(addr string, port string) (rawListener *Listener) {
+	rawListener = &Listener{}
+
+	rawListener.c_packets = make(chan *TCPPacket, 100)
+	rawListener.c_messages = make(chan *TCPMessage, 100)
+	rawListener.c_del_message = make(chan *TCPMessage, 100)
+	rawListener.messages = make(map[string]*TCPMessage)
+
+	rawListener.addr = addr
+	rawListener.port, _ = strconv.Atoi(port)
+
+	go rawListener.listen()
+	go rawListener.readRAWSocket()
+
+	return
+}
+
+func (t *Listener) listen() {
+	for {
+		select {
+		// If message ready for deletion it means that its also complete or expired by timeout
+		case message := <-t.c_del_message:
+			t.c_messages <- message
+			delete(t.messages, message.ID)
+
+		// We need to use channels to process each packet to avoid data races
+		case packet := <-t.c_packets:
+			t.processTCPPacket(packet)
+		}
+	}
+}
+
+func (t *Listener) readRAWSocket() {
+	conn, e := net.ListenPacket("ip4:tcp", t.addr)
+	defer conn.Close()
+
+	if e != nil {
+		log.Fatal(e)
+	}
+
+	buf := make([]byte, 4096*2)
+
+	for {
+		// Note: ReadFrom receive messages without IP header
+		n, addr, err := conn.ReadFrom(buf)
+
+		if err != nil {
+			log.Println("Error:", err)
+			continue
+		}
+
+		if n > 0 {
+			t.parsePacket(addr, buf[:n])
+		}
+	}
+}
+
+func (t *Listener) parsePacket(addr net.Addr, buf []byte) {
+	if t.isIncomingDataPacket(buf) {
+		new_buf := make([]byte, len(buf))
+		copy(new_buf, buf)
+
+		t.c_packets <- ParseTCPPacket(addr, new_buf)
+	}
+}
+
+func (t *Listener) isIncomingDataPacket(buf []byte) bool {
+	// To avoid full packet parsing every time, we manually parsing values needed for packet filtering
+	// http://en.wikipedia.org/wiki/Transmission_Control_Protocol
+	dest_port := binary.BigEndian.Uint16(buf[2:4])
+
+	// Because RAW_SOCKET can't be bound to port, we have to control it by ourself
+	if int(dest_port) == t.port {
+		// Get the 'data offset' (size of the TCP header in 32-bit words)
+		dataOffset := (buf[12] & 0xF0) >> 4
+
+		// We need only packets with data inside
+		// Check that the buffer is larger than the size of the TCP header
+		if len(buf) > int(dataOffset*4) {
+			// We should create new buffer because go slices is pointers. So buffer data shoud be immutable.
+			return true
+		}
+	}
+
+	return false
+}
+
+// Trying to add packet to existing message or creating new message
+//
+// For TCP message unique id is Acknowledgment number (see tcp_packet.go)
+func (t *Listener) processTCPPacket(packet *TCPPacket) {
+	defer func() { recover() }()
+
+	var message *TCPMessage
+	m_id := packet.Addr.String() + strconv.Itoa(int(packet.Ack))
+
+	message, ok := t.messages[m_id]
+
+	if !ok {
+		// We sending c_del_message channel, so message object can communicate with Listener and notify it if message completed
+		message = NewTCPMessage(m_id, t.c_del_message)
+		t.messages[m_id] = message
+	}
+
+	// Adding packet to message
+	message.c_packets <- packet
+}
+
+// Receive TCP messages from the listener channel
+func (t *Listener) Receive() *TCPMessage {
+	return <-t.c_messages
+}

raw_socket_listener/tcp_message.go

@@ -1,6 +1,7 @@
-package listener
+package raw_socket
 
 import (
+	"log"
 	"sort"
 	"time"
 )
@@ -10,28 +11,25 @@ const MSG_EXPIRE = 200 * time.Millisecond
 // TCPMessage ensure that all TCP packets for given request is received, and processed in right sequence
 // Its needed because all TCP message can be fragmented or re-transmitted
 //
-// Each TCP Packet have 2 ids: acknowledgement - message_id, and sequence - packet_id
+// Each TCP Packet have 2 ids: acknowledgment - message_id, and sequence - packet_id
 // Message can be compiled from unique packets with same message_id which sorted by sequence
-// Message is received if we did't receive any packets for 200ms
+// Message is received if we didn't receive any packets for 200ms
 type TCPMessage struct {
-	Ack     uint32 // Message ID
+	ID      string // Message ID
 	packets []*TCPPacket
 
 	timer *time.Timer // Used for expire check
 
-	expired bool
-
 	c_packets chan *TCPPacket
-	c_closing chan int
 
 	c_del_message chan *TCPMessage
 }
 
-func NewTCPMessage(Ack uint32, c_del chan *TCPMessage) (msg *TCPMessage) {
-	msg = &TCPMessage{Ack: Ack}
+// NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted
+func NewTCPMessage(ID string, c_del chan *TCPMessage) (msg *TCPMessage) {
+	msg = &TCPMessage{ID: ID}
 
 	msg.c_packets = make(chan *TCPPacket)
-	msg.c_closing = make(chan int)
 	msg.c_del_message = c_del // used for notifying that message completed or expired
 
 	// Every time we receive packet we reset this timer
@@ -45,21 +43,24 @@ func NewTCPMessage(Ack uint32, c_del chan *TCPMessage) (msg *TCPMessage) {
 func (t *TCPMessage) listen() {
 	for {
 		select {
-		case <-t.c_closing:
-			close(t.c_packets)
-			return // Stop loop if message completed/expired
-		case packet := <-t.c_packets:
-			t.AddPacket(packet)
+		case packet, more := <-t.c_packets:
+			if more {
+				t.AddPacket(packet)
+			} else {
+				// Stop loop if channel closed
+				return
+			}
 		}
 	}
 }
 
+// Timeout notifies message to stop listening, close channel and message ready to be sent
 func (t *TCPMessage) Timeout() {
-	t.c_closing <- 1     // Notify to stop listen loop and close channel
+	close(t.c_packets)   // Notify to stop listen loop and close channel
 	t.c_del_message <- t // Notify RAWListener that message is ready to be send to replay server
 }
 
-// Sort packets in right orders and return message content
+// Bytes sorts packets in right orders and return message content
 func (t *TCPMessage) Bytes() (output []byte) {
 	mk := make([]int, len(t.packets))
 
@@ -78,14 +79,9 @@ func (t *TCPMessage) Bytes() (output []byte) {
 	return
 }
 
-// Add packet to the message and ensure packet uniquiness
+// AddPacket to the message and ensure packet uniqueness
 // TCP allows that packet can be re-send multiple times
 func (t *TCPMessage) AddPacket(packet *TCPPacket) {
-	if t.expired {
-		Debug("Adding packet to expired message")
-		return
-	}
-
 	packetFound := false
 
 	for _, pkt := range t.packets {
@@ -96,7 +92,7 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) {
 	}
 
 	if packetFound {
-		Debug("Received packet with same sequence")
+		log.Println("Received packet with same sequence")
 	} else {
 		t.packets = append(t.packets, packet)
 	}

raw_socket_listener/tcp_packet.go

@@ -1,7 +1,8 @@
-package listener
+package raw_socket
 
 import (
 	"encoding/binary"
+	"net"
 	"strconv"
 	"strings"
 )
@@ -34,45 +35,45 @@ type TCPPacket struct {
 	Urgent     uint16
 
 	Data []byte
+
+	Addr net.Addr
 }
 
-func NewTCPPacket(b []byte) (p *TCPPacket) {
+func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) {
 	p = &TCPPacket{Data: b}
-	p.ParseFast()
+	p.ParseBasic()
+	p.Addr = addr
 
 	return p
 }
 
-// Inspired by: https://github.com/miekg/pcap/blob/master/packet.go
+// Parse TCP Packet, inspired by: https://github.com/miekg/pcap/blob/master/packet.go
 func (t *TCPPacket) Parse() {
+	t.ParseBasic()
 	t.SrcPort = binary.BigEndian.Uint16(t.Data[0:2])
 	t.DestPort = binary.BigEndian.Uint16(t.Data[2:4])
-	t.Seq = binary.BigEndian.Uint32(t.Data[4:8])
-	t.Ack = binary.BigEndian.Uint32(t.Data[8:12])
-	t.DataOffset = (t.Data[12] & 0xF0) >> 4
 	t.Flags = binary.BigEndian.Uint16(t.Data[12:14]) & 0x1FF
 	t.Window = binary.BigEndian.Uint16(t.Data[14:16])
 	t.Checksum = binary.BigEndian.Uint16(t.Data[16:18])
 	t.Urgent = binary.BigEndian.Uint16(t.Data[18:20])
-
-	t.Data = t.Data[t.DataOffset*4:]
 }
 
-// Parse only needed set of fields
-func (t *TCPPacket) ParseFast() {
+// ParseBasic set of fields
+func (t *TCPPacket) ParseBasic() {
 	t.Seq = binary.BigEndian.Uint32(t.Data[4:8])
 	t.Ack = binary.BigEndian.Uint32(t.Data[8:12])
-
 	t.DataOffset = (t.Data[12] & 0xF0) >> 4
+
 	t.Data = t.Data[t.DataOffset*4:]
 }
 
+// String output for a TCP Packet
 func (t *TCPPacket) String() string {
 	return strings.Join([]string{
 		"Source port: " + strconv.Itoa(int(t.SrcPort)),
 		"Dest port:" + strconv.Itoa(int(t.DestPort)),
 		"Sequence:" + strconv.Itoa(int(t.Seq)),
-		"Acknowledgement:" + strconv.Itoa(int(t.Ack)),
+		"Acknowledgment:" + strconv.Itoa(int(t.Ack)),
 		"Header len:" + strconv.Itoa(int(t.DataOffset)),
 
 		"Flag ns:" + strconv.FormatBool(t.Flags&TCP_NS != 0),

replay/replay.go

@@ -1,120 +0,0 @@
-// Replay server receive requests objects from Listeners and forward it to given address.
-// Basic usage:
-//
-//     gor replay -f http://staging.server
-//
-//
-// Rate limiting
-//
-// It can be useful if you want forward only part of production traffic, not to overload staging environment. You can specify desired request per second using "|" operator after server address:
-//
-//     # staging.server not get more than 10 requests per second
-//     gor replay -f "http://staging.server|10"
-//
-//
-// Forward to multiple addresses
-//
-// Just separate addresses by coma:
-//    gor replay -f "http://staging.server|10,http://dev.server|20"
-//
-//
-//  For more help run:
-//
-//     gor replay -h
-//
-package replay
-
-import (
-	"bufio"
-	"bytes"
-	"io"
-	"log"
-	"net"
-	"net/http"
-)
-
-const bufSize = 4096
-
-// Enable debug logging only if "--verbose" flag passed
-func Debug(v ...interface{}) {
-	if Settings.Verbose {
-		log.Println(v...)
-	}
-}
-
-func ParseRequest(data []byte) (request *http.Request, err error) {
-	buf := bytes.NewBuffer(data)
-	reader := bufio.NewReader(buf)
-
-	request, err = http.ReadRequest(reader)
-	return
-}
-
-// Because its sub-program, Run acts as `main`
-// Replay server listen to UDP traffic from Listeners
-// Each request processed by RequestFactory
-func Run() {
-	listener, err := net.Listen("tcp", Settings.Address())
-
-	log.Println("Starting replay server at:", Settings.Address())
-
-	if err != nil {
-		log.Fatal("Can't start:", err)
-	}
-
-	for _, host := range Settings.ForwardedHosts() {
-		log.Println("Forwarding requests to:", host.Url, "limit:", host.Limit)
-	}
-
-	requestFactory := NewRequestFactory()
-
-	for {
-		conn, err := listener.Accept()
-
-		if err != nil {
-			log.Println("Error while Accept()", err)
-			continue
-		}
-
-		go handleConnection(conn, requestFactory)
-	}
-
-}
-
-func handleConnection(conn net.Conn, rf *RequestFactory) error {
-	defer conn.Close()
-
-	var read = true
-	var response []byte
-	var buf []byte
-
-	buf = make([]byte, bufSize)
-
-	for read {
-		n, err := conn.Read(buf)
-
-		switch err {
-		case io.EOF:
-			read = false
-		case nil:
-			response = append(response, buf[0:n]...)
-			if n < bufSize {
-				read = false
-			}
-		default:
-			read = false
-		}
-	}
-
-	go func() {
-		if request, err := ParseRequest(response); err != nil {
-			Debug("Error while parsing request", err, response)
-		} else {
-			Debug("Adding request", request)
-
-			rf.Add(request)
-		}
-	}()
-
-	return nil
-}

replay/request_factory.go

@@ -1,102 +0,0 @@
-package replay
-
-import (
-	"errors"
-	"net/http"
-	"net/url"
-)
-
-type HttpResponse struct {
-	host *ForwardHost
-	req  *http.Request
-	resp *http.Response
-	err  error
-}
-
-// Class for processing requests
-//
-// Basic workflow:
-//
-// 1. When request added via Add() it get pushed to `responses` chan
-// 2. handleRequest() listen for `responses` chan and decide where request should be forwarded, and apply rate-limit if needed
-// 3. sendRequest() forwards request and returns response info to `responses` chan
-// 4. handleRequest() listen for `response` channel and updates stats
-type RequestFactory struct {
-	c_responses chan *HttpResponse
-	c_requests  chan *http.Request
-}
-
-// RequestFactory contstuctor
-// One created, it starts listening for incoming requests: requests channel
-func NewRequestFactory() (factory *RequestFactory) {
-	factory = &RequestFactory{}
-	factory.c_responses = make(chan *HttpResponse)
-	factory.c_requests = make(chan *http.Request)
-
-	go factory.handleRequests()
-
-	return
-}
-
-// Disable redirects https://github.com/buger/gor/pull/15
-func customCheckRedirect(req *http.Request, via []*http.Request) error {
-	if len(via) >= 0 {
-		return errors.New("stopped after 2 redirects")
-	}
-	return nil
-}
-
-// Forward http request to given host
-func (f *RequestFactory) sendRequest(host *ForwardHost, request *http.Request) {
-	client := &http.Client{
-		CheckRedirect: customCheckRedirect,
-	}
-
-	// Change HOST of original request
-	URL := host.Url + request.URL.Path + "?" + request.URL.RawQuery
-
-	request.RequestURI = ""
-	request.URL, _ = url.ParseRequestURI(URL)
-
-	Debug("Sending request:", host.Url, request)
-
-	resp, err := client.Do(request)
-
-	if err == nil {
-		defer resp.Body.Close()
-	} else {
-		Debug("Request error:", err)
-	}
-
-	f.c_responses <- &HttpResponse{host, request, resp, err}
-}
-
-// Handle incoming requests, and they responses
-func (f *RequestFactory) handleRequests() {
-	hosts := Settings.ForwardedHosts()
-
-	for {
-		select {
-		case req := <-f.c_requests:
-			for _, host := range hosts {
-				// Ensure that we have actual stats for given timestamp
-				host.Stat.Touch()
-
-				if host.Limit == 0 || host.Stat.Count < host.Limit {
-					// Increment Stat.Count
-					host.Stat.IncReq()
-
-					go f.sendRequest(host, req)
-				}
-			}
-		case resp := <-f.c_responses:
-			// Increment returned http code stats, and elapsed time
-			resp.host.Stat.IncResp(resp)
-		}
-	}
-}
-
-// Add request to channel for further processing
-func (f *RequestFactory) Add(request *http.Request) {
-	f.c_requests <- request
-}

replay/request_stats.go

@@ -1,63 +0,0 @@
-package replay
-
-import (
-	"time"
-)
-
-// Stats stores in context of current timestamp
-type RequestStat struct {
-	timestamp int64
-
-	Codes map[int]int // { 200: 10, 404:2, 500:1 }
-
-	Count  int // All requests including errors
-	Errors int // Rquests with errors (timeout or host not reachable). Not include 50x errors.
-
-	host *ForwardHost
-}
-
-// Ensure that current stats is actual (for current timestamp)
-func (s *RequestStat) Touch() {
-	if s.timestamp != time.Now().Unix() {
-		s.reset()
-	}
-}
-
-// Called on request start
-func (s *RequestStat) IncReq() {
-	s.Count++
-}
-
-// Called after response
-func (s *RequestStat) IncResp(resp *HttpResponse) {
-	s.Touch()
-
-	if resp.err != nil {
-		s.Errors++
-		return
-	}
-
-	s.Codes[resp.resp.StatusCode]++
-}
-
-// Updated stats timestamp to current time and reset to zero all stats values
-// TODO: Further on reset it should write stats to file
-func (s *RequestStat) reset() {
-	if s.timestamp != 0 {        
-	   Debug("Host:", s.host.Url, "Requests:", s.Count, "Errors:", s.Errors, "Status codes:", s.Codes)
-	}
-
-	s.timestamp = time.Now().Unix()
-
-	s.Codes = make(map[int]int)
-	s.Count = 0
-	s.Errors = 0
-}
-
-// RequestStat constructor
-func NewRequestStats(host *ForwardHost) (stat *RequestStat) {
-	stat = &RequestStat{host: host}
-	stat.reset()
-
-	return
-}

replay/settings.go

@@ -1,79 +0,0 @@
-package replay
-
-import (
-	"flag"
-	"os"
-	"strconv"
-	"strings"
-)
-
-type ForwardHost struct {
-	Url   string
-	Limit int
-
-	Stat *RequestStat
-}
-
-type ReplaySettings struct {
-	Port int
-	Host string
-
-	ForwardAddress string
-
-	Verbose bool
-}
-
-// ForwardedHosts implements forwardAddress syntax support for multiple hosts (coma separated), and rate limiting by specifing "|maxRps" after host name.
-//
-//    -f "host1,http://host2|10,host3"
-//
-func (r *ReplaySettings) ForwardedHosts() (hosts []*ForwardHost) {
-	hosts = make([]*ForwardHost, 0, 10)
-
-	for _, address := range strings.Split(r.ForwardAddress, ",") {
-		host_info := strings.Split(address, "|")
-
-		if strings.Index(host_info[0], "http") == -1 {
-			host_info[0] = "http://" + host_info[0]
-		}
-
-		host := &ForwardHost{Url: host_info[0]}
-		host.Stat = NewRequestStats(host)
-
-		if len(host_info) > 1 {
-			host.Limit, _ = strconv.Atoi(host_info[1])
-		}
-
-		hosts = append(hosts, host)
-	}
-
-	return
-}
-
-// Helper to return address with port, e.g.: 127.0.0.1:28020
-func (r *ReplaySettings) Address() string {
-	return r.Host + ":" + strconv.Itoa(r.Port)
-}
-
-var Settings ReplaySettings = ReplaySettings{}
-
-func init() {
-	if len(os.Args) < 2 || os.Args[1] != "replay" {
-		return
-	}
-
-	const (
-		defaultPort = 28020
-		defaultHost = "0.0.0.0"
-
-		defaultAddress = "http://localhost:8080"
-	)
-
-	flag.IntVar(&Settings.Port, "p", defaultPort, "specify port number")
-
-	flag.StringVar(&Settings.Host, "ip", defaultHost, "ip addresses to listen on")
-
-	flag.StringVar(&Settings.ForwardAddress, "f", defaultAddress, "http address to forward traffic.\n\tYou can limit requests per second by adding `|num` after address.\n\tIf you have multiple addresses with different limits. For example: http://staging.example.com|100,http://dev.example.com|10")
-
-	flag.BoolVar(&Settings.Verbose, "verbose", false, "Log requests")
-}

settings.go

@@ -0,0 +1,72 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"log"
+	"os"
+)
+
+const (
+	VERSION = "0.8"
+)
+
+type AppSettings struct {
+	verbose bool
+
+	splitOutput bool
+
+	inputDummy  MultiOption
+	outputDummy MultiOption
+
+	inputTCP  MultiOption
+	outputTCP MultiOption
+
+	inputFile  MultiOption
+	outputFile MultiOption
+
+	inputRAW MultiOption
+
+	outputHTTP              MultiOption
+	outputHTTPHeaders       HTTPHeaders
+	outputHTTPMethods       HTTPMethods
+	outputHTTPElasticSearch string
+}
+
+var Settings AppSettings = AppSettings{}
+
+func usage() {
+	fmt.Printf("Gor is a simple http traffic replication tool written in Go. Its main goal is to replay traffic from production servers to staging and dev environments.\nProject page: https://github.com/buger/gor\nAuthor: <Leonid Bugaev> leonsbox@gmail.com\nCurrent Version: %s\n\n", VERSION)
+	flag.PrintDefaults()
+	os.Exit(2)
+}
+
+func init() {
+	flag.Usage = usage
+
+	flag.BoolVar(&Settings.verbose, "verbose", false, "Turn on verbose/debug output")
+
+	flag.BoolVar(&Settings.splitOutput, "split-output", false, "By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.")
+
+	flag.Var(&Settings.inputDummy, "input-dummy", "Used for testing outputs. Emits 'Get /' request every 1s")
+	flag.Var(&Settings.outputDummy, "output-dummy", "Used for testing inputs. Just prints data coming from inputs.")
+
+	flag.Var(&Settings.inputTCP, "input-tcp", "Used for internal communication between Gor instances. Example: \n\t# Receive requests from other Gor instances on 28020 port, and redirect output to staging\n\tgor --input-tcp :28020 --output-http staging.com")
+	flag.Var(&Settings.outputTCP, "output-tcp", "Used for internal communication between Gor instances. Example: \n\t# Listen for requests on 80 port and forward them to other Gor instance on 28020 port\n\tgor --input-raw :80 --output-tcp replay.local:28020")
+
+	flag.Var(&Settings.inputFile, "input-file", "Read requests from file: \n\tgor --input-file ./requests.gor --output-http staging.com")
+	flag.Var(&Settings.outputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor")
+
+	flag.Var(&Settings.inputRAW, "input-raw", "Capture traffic from given port (use RAW sockets and require *sudo* access):\n\t# Capture traffic from 8080 port\n\tgor --input-raw :8080 --output-http staging.com")
+
+	flag.Var(&Settings.outputHTTP, "output-http", "Forwards incoming requests to given http address.\n\t# Redirect all incoming requests to staging.com address \n\tgor --input-raw :80 --output-http http://staging.com")
+	flag.Var(&Settings.outputHTTPHeaders, "output-http-header", "Inject additional headers to http reqest:\n\tgor --input-raw :8080 --output-http staging.com --output-http-header 'User-Agent: Gor'")
+	flag.Var(&Settings.outputHTTPMethods, "output-http-method", "Whitelist of HTTP methods to replay. Anything else will be dropped:\n\tgor --input-raw :8080 --output-http staging.com --output-http-method GET --output-http-method OPTIONS")
+	flag.StringVar(&Settings.outputHTTPElasticSearch, "output-http-elasticsearch", "", "Send request and response stats to ElasticSearch:\n\tgor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'")
+}
+
+func Debug(args ...interface{}) {
+	if Settings.verbose {
+		log.Println(args...)
+	}
+}

settings_headers.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+type HTTPHeaders []HTTPHeader
+type HTTPHeader struct {
+	Name  string
+	Value string
+}
+
+func (h *HTTPHeaders) String() string {
+	return fmt.Sprint(*h)
+}
+
+func (h *HTTPHeaders) Set(value string) error {
+	v := strings.SplitN(value, ":", 2)
+	if len(v) != 2 {
+		return errors.New("Expected `Key: Value`")
+	}
+
+	header := HTTPHeader{
+		strings.TrimSpace(v[0]),
+		strings.TrimSpace(v[1]),
+	}
+
+	*h = append(*h, header)
+	return nil
+}

settings_methods.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"fmt"
+	"strings"
+)
+
+type HTTPMethods []string
+
+func (h *HTTPMethods) String() string {
+	return fmt.Sprint(*h)
+}
+
+func (h *HTTPMethods) Set(value string) error {
+	*h = append(*h, strings.ToUpper(value))
+	return nil
+}
+
+func (h *HTTPMethods) Contains(value string) bool {
+	for _, method := range *h {
+		if value == method {
+			return true
+		}
+	}
+	return false
+}

settings_methods_test.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"testing"
+)
+
+func TestHTTPMethods(t *testing.T) {
+	methods := HTTPMethods{}
+
+	methods.Set("lower")
+	methods.Set("UPPER")
+
+	if !methods.Contains("LOWER") {
+		t.Error("Does not contain LOWER")
+	}
+
+	if !methods.Contains("UPPER") {
+		t.Error("Does not contain UPPER")
+	}
+
+	if methods.Contains("ABSENT") {
+		t.Error("Does contain ABSENT")
+	}
+}

settings_option.go

@@ -0,0 +1,16 @@
+package main
+
+import (
+	"fmt"
+)
+
+type MultiOption []string
+
+func (h *MultiOption) String() string {
+	return fmt.Sprint(*h)
+}
+
+func (h *MultiOption) Set(value string) error {
+	*h = append(*h, value)
+	return nil
+}

test_input.go

@@ -0,0 +1,35 @@
+package main
+
+type TestInput struct {
+	data chan []byte
+}
+
+func NewTestInput() (i *TestInput) {
+	i = new(TestInput)
+	i.data = make(chan []byte, 100)
+
+	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) EmitOPTIONS() {
+	i.data <- []byte("OPTIONS / HTTP/1.1\nHost: www.w3.org\r\n\r\n")
+}
+
+func (i *TestInput) String() string {
+	return "Test Input"
+}

test_output.go

@@ -0,0 +1,24 @@
+package main
+
+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"
+}