It should be 0.9.2 version

Limiter improvements

Makefile

@@ -1,4 +1,4 @@
-SOURCE = emitter.go gor.go gor_stat.go input_dummy.go input_file.go input_raw.go input_tcp.go limiter.go output_dummy.go output_file.go output_http.go output_tcp.go plugins.go settings.go settings_header_filters.go settings_header_hash_filters.go settings_headers.go settings_methods.go settings_option.go settings_url_regexp.go test_input.go elasticsearch.go settings_url_map.go
+SOURCE = emitter.go gor.go gor_stat.go input_dummy.go input_file.go input_raw.go input_tcp.go limiter.go output_dummy.go output_file.go input_http.go output_http.go output_tcp.go plugins.go settings.go settings_header_filters.go settings_header_hash_filters.go settings_headers.go settings_methods.go settings_option.go settings_url_regexp.go test_input.go elasticsearch.go settings_url_map.go
 
 release: release-x86 release-x64
 
@@ -12,10 +12,17 @@ dbuild:
 	docker build -t gor .
 
 dtest:
-	docker run -v `pwd`:/gopath/src/gor -t -i --env GORACE="halt_on_error=1" gor go test -race -v
+	docker run -v `pwd`:/gopath/src/gor -t -i --env GORACE="halt_on_error=1" gor go test $(ARGS) -race -v
 
+dfmt:
+	docker run -v `pwd`:/gopath/src/gor -t -i gor go fmt
+
+dbench:
+	docker run -v `pwd`:/gopath/src/gor -t -i gor go test -v -run NOT_EXISTING -bench HTTP
+
+# Used mainly for debugging, because docker container do not have access to parent machine ports
 drun:
-	docker run -v `pwd`:/gopath/src/gor -t -i gor go run $(SOURCE) --input-dummy=0 --output-dummy=0 --verbose
+	docker run -v `pwd`:/gopath/src/gor -t -i gor go run $(SOURCE) --input-dummy=0 --input-http=:9000 --output-http="http://localhost:9000"  --verbose
 
 dbash: 
 	docker run -v `pwd`:/gopath/src/gor -t -i gor /bin/bash

README.md

@@ -1,357 +1,386 @@
-[![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.
-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.
-**No more falling down in production!**
-
-Here is basic workflow: 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)
-
-
-## 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 --input-raw :80 --output-tcp replay.local: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 support rate limiting. It can be useful if you want
-forward only part of production traffic and not overload your staging
-environment. 
-Current throttling works like this: If for current second it reached specified requests limit - disregard the rest, on next second counter reseted.
-
-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 --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 --input-raw :80 --output-tcp "replay.local:28020|10"
-```
-
-#### Match on regexp of url
-```
-# only forward requests being sent to the api... domains
-gor --input-raw :8080 --output-http staging.com --output-http-url-regexp ^www.
-```
-
-#### Filter based on regexp of header
-```
-# only forward requests with an api version of 1.0x
-gor --input-raw :8080 --output-http staging.com --output-http-header-filter api-version:^1\.0\d
-```
-
-#### Filter based on hash of header
-```
-# send 1/32 of all users consistently to staging
-gor --input-raw :8080 --output-http staging.com --output-http-header-hash-filter user-id:1/32
-```
-
-### Forward to multiple addresses
-
-You can forward traffic to multiple endpoints. Just add multiple --output-* arguments.
-```
-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 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 "User-Agent: Replayed by Gor" \
-    --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.
-
-#### Rewrite the target urls based on a mapping
-```
-# rewrite url to match the following
-gor --input-raw :8080 --output-http staging.com --output-http-rewrite-url /xml_test/interface.php:/api/service.do
-```
-
-
-## 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
-
-Feel free to ask question directly by email or by creating github issue.
-
-## Latest releases (including binaries)
-
-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-http-header-filter=[]: A regexp to match a specific header against. Requests with non-matching headers will be dropped:
-    gor --input-raw :8080 --output-http staging.com --output-http-header-filter api-version:^v1
-
-  -output-http-header-hash-filter=[]: Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific header. The fraction must have a denominator that is a power of two:
-    gor --input-raw :8080 --output-http staging.com --output-http-header-hash-filter user-id:1/4
-
-  -output-http-url-regexp=: A regexp to match requests against. Anything else will be dropped:
-    gor --input-raw :8080 --output-http staging.com --output-http-url-regexp ^www.
-    
-  -output-http-workers=-1: Number of http output workers desired. Use default -1 for dynamic worker scaling.  Gor will add http workers if its work queue starts getting too full and kill them .
-    
-  -output-http-stats=false: If set to `true` it gives out queuing stats for the HTTP output every 5 seconds in the form latest,mean,max,count,count/second.
-    
-  -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
-    
-  -output-tcp-stats=false: If set to `true` it gives out queuing stats for the TCP output every 5 seconds in the form latest,mean,max,count,count/second.
-    
-  -split-output=false: By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.
-
-  -output-http-rewrite-url=[]: Rewrites the url in the request based on a mapping
-    gor --input-raw :8080 --output-http staging.com --output-http-rewrite-url /xml_test/interface.php:/api/service.do
-```
-
-## 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`.
-3. `cd $GOPATH/src/github.com/buger/gor`
-4. `go build` to get binary, or `go test` to run tests
-
-## Development
-Project contains Docker environment.
-
-1. Build container: `make dbuild`
-2. Run tests: `make dtest`
-3. Bash access to container: `make dbash`. Inside container you have python to run simple web server `python -m SimpleHTTPServer 8080` and `curl` to make http requests. 
-
-
-## Questions and support 
-
-All bug-reports and suggestions should go though Github Issues or our [Google Group](https://groups.google.com/forum/#!forum/gor-users). Or you can just send email to gor-users@googlegroups.com
-
-If you have some private questions you can send direct mail to leonsbox@gmail.com
-
-## 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 `--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.
-
-### 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
-
-### What do the stats commands do?
-Gor can report stats on the output-tcp and output-http request queues. Stats are reported to the console every 5 seconds in the form `latest,mean,max,count,count/second` by using the `-output-http-stats` and `-output-tcp-stats` options.
-
-Examples:
-
-```
-2014/04/23 21:17:50 output_tcp:latest,mean,max,count,count/second
-2014/04/23 21:17:50 output_tcp:0,0,0,0,0
-2014/04/23 21:17:55 output_tcp:1,1,2,68,13
-2014/04/23 21:18:00 output_tcp:1,1,2,92,18
-2014/04/23 21:18:05 output_tcp:1,1,2,119,23
-2014/04/23 21:18:10 output_tcp:1,0,1,95,19
-2014/04/23 21:18:15 output_tcp:1,1,2,92,18
-2014/04/23 21:18:20 output_tcp:1,1,2,108,21
-2014/04/23 21:18:25 output_tcp:1,1,2,117,23
-2014/04/23 21:18:30 output_tcp:1,1,2,113,22
-2014/04/23 21:18:35 output_tcp:21,20,21,132,26
-2014/04/23 21:18:40 output_tcp:100,99,100,99,19
-```
-
-```
-Version: 0.8
-2014/04/23 21:19:46 output_http:latest,mean,max,count,count/second
-2014/04/23 21:19:46 output_http:0,0,0,0,0
-2014/04/23 21:19:51 output_http:0,0,0,0,0
-2014/04/23 21:19:56 output_http:0,0,0,0,0
-2014/04/23 21:20:01 output_http:1,0,1,50,10
-2014/04/23 21:20:06 output_http:1,1,4,72,14
-2014/04/23 21:20:11 output_http:1,0,1,179,35
-2014/04/23 21:20:16 output_http:1,0,1,148,29
-2014/04/23 21:20:21 output_http:1,1,2,91,18
-2014/04/23 21:20:26 output_http:1,1,2,150,30
-2014/04/23 21:18:15 output_http:100,99,100,70,14
-2014/04/23 21:18:21 output_http:100,99,100,55,11
-2014/04/23 21:18:28 output_http:100,99,100,55,11
-2014/04/23 21:18:34 output_http:100,99,100,57,11
-2014/04/23 21:18:41 output_http:100,99,100,61,12
-2014/04/23 21:18:48 output_http:100,99,100,56,11
-2014/04/23 21:18:56 output_http:100,99,100,58,11
-2014/04/23 21:19:01 output_http:100,99,100,31,6
-2014/04/23 21:19:08 output_http:100,99,100,61,12
-2014/04/23 21:19:15 output_http:100,99,100,64,12
-2014/04/23 21:19:21 output_http:100,99,100,70,14
-2014/04/23 21:19:28 output_http:100,99,100,61,12
-2014/04/23 21:19:35 output_http:100,99,100,56,11
-```
-
-### How can I tell if I have bottlenecks?
-Key areas that sometimes experience bottlenecks are the output-tcp and output-http functions which have internal queues for requests. Each queue has an upper limit of 100. Enable stats reporting to see if any queues are experiencing bottleneck behavior.
- 
-#### output-http bottlenecks
-When running a Gor replay the output-http feature may bottleneck if:
-
-  * the replay has inadequate bandwidth. If the replay is receiving or sending more messages than its network adapter can handle the output-http-stats  may report that the output-http queue is filling up. See if there is a way to upgrade the replay's bandwidth.
-  * with `--output-http-workers` set to anything other than `-1` the `-output-http` target is unable to respond to messages in a timely manner. The http output workers which take messages off the output-http queue, process the request, and ensure that the request did not result in an error may not be able to keep up with the number of incoming requests. If the replay is not using dynamic worker scaling (`--output-http-workers=-1`)  The optimal number of output-http-workers can be determined with the formula `output-workers = (Average number of requests per second)/(Average target response time per second)`.
-
-#### output-tcp bottlenecks
-When using the Gor listener the output-tcp feature may bottleneck if:
-
-  * the replay is unable to accept and process more requests than the listener is able generate. Prior to troubleshooting the output-tcp bottleneck, ensure that the replay target is not experiencing any bottlenecks. 
-  * the replay target has inadequate bandwidth to handle all its incoming requests.  If a replay target's incoming bandwidth is maxed out the output-tcp-stats may report that the output-tcp queue is filling up. See if there is a way to upgrade the replay's bandwidth.
-
-### The CPU average across my load-balanced targets is higher than the source
-If you are replaying traffic from multiple listeners to a load-balanced target and you use sticky sessions, you may observe that the target servers have a higher CPU load than the listener servers. This may be because the sticky session cookie of the original load balancer is not honored by the target load balancer thus resulting in requests that would normally hit the same target server hitting different servers on the backend thus reducing some caching benefits gained via the load balancing.  Try running just one listener against one replay target and see if the CPU utilization comparison is more accurate.
- 
-### How does dynamic http worker scaling work?
-By using the Gor setting `--output-http-workers=-1` Gor will create more http output workers when the http output queue length is greater than 10.  The number of workers created (N) is equal to the queue length at the time which it is checked and found to have a length greater than 10. The queue length is checked every time a message is written to the http output queue.  No more workers will be spawned until that request to spawn N workers is satisfied.  If a dynamic worker cannot process a message at that time, it will sleep for 100 milliseconds. If a dynamic worker cannot process a message for 2 seconds it dies.  
-
-## 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
-
-1. Fork it
-2. Create your feature branch (git checkout -b my-new-feature)
-3. Commit your changes (git commit -am 'Added some feature')
-4. Push to the branch (git push origin my-new-feature)
-5. Create new Pull Request
-
-## Companies using Gor
-
-* [Granify](http://granify.com)
-* [GOV.UK](https://www.gov.uk) ([Government Digital Service](http://digital.cabinetoffice.gov.uk/))
-* [theguardian.com](http://theguardian.com)
-* [TomTom](http://www.tomtom.com/)
-* To add your company drop me a line to github.com/buger or leonsbox@gmail.com
+[![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.
+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.
+**No more falling down in production!**
+
+Here is basic workflow: 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)
+
+
+## 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 --input-raw :80 --output-tcp replay.local: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"
+```
+
+### Guarante of replay and HTTP input
+Due to how traffic interception works, there is chance of missing requests. If you want guarantee that requests will be replayed you can use http input, but it will require changes in your app as well. 
+
+```
+sudo gor --input-http :28019 --output-http "http://staging.com"
+```
+
+Then in your application you should send copy (e.g. like reverse proxy) all incoming requests to Gor http input. 
+
+
+## Advanced use
+
+### Rate limiting
+Every input and output support rate limiting. It can be useful if you want
+forward only part of production traffic and not overload your staging
+environment. 
+
+There are 2 limiting algorithms: absolute or percentage based. 
+
+Absolute: If for current second it reached specified requests limit - disregard the rest, on next second counter reseted.
+
+Percentage: For input-file it will slowdown or speedup request execution, for the rest it will use random generator to decide if request pass or not based on weight you specified. 
+
+You can specify your desired limit using the
+"|" operator after the server address:
+
+#### Limiting replay using absolute number
+```
+# staging.server will not get more than 10 requests per second
+gor --input-tcp :28020 --output-http "http://staging.com|10"
+```
+
+#### Limiting listener using percentage based limiter
+```
+# replay server will not get more than 10% of requests 
+# useful for high-load environments
+gor --input-raw :80 --output-tcp "replay.local:28020|10%"
+```
+
+### Load testing
+
+Currently it supported only by `input-file` and only when using percentage based limiter. Unlike default limiter for `input-file` instead of dropping requests it will slowdown or speedup request emitting.
+
+```
+# Replay from file on 2x speed 
+gor --input-file "requests.gor|200%" --output-http "staging.com"
+```
+
+
+### Filtering 
+
+#### Match on regexp of url
+```
+# only forward requests being sent to the api... domains
+gor --input-raw :8080 --output-http staging.com --output-http-url-regexp ^www.
+```
+
+#### Filter based on regexp of header
+```
+# only forward requests with an api version of 1.0x
+gor --input-raw :8080 --output-http staging.com --output-http-header-filter api-version:^1\.0\d
+```
+
+#### Filter based on hash of header
+```
+# send 1/32 of all users consistently to staging
+gor --input-raw :8080 --output-http staging.com --output-http-header-hash-filter user-id:1/32
+```
+
+### Forward to multiple addresses
+
+You can forward traffic to multiple endpoints. Just add multiple --output-* arguments.
+```
+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 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 "User-Agent: Replayed by Gor" \
+    --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.
+
+#### Rewrite the target urls based on a mapping
+```
+# rewrite url to match the following
+gor --input-raw :8080 --output-http staging.com --output-http-rewrite-url /xml_test/interface.php:/api/service.do
+```
+
+
+## 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
+
+Feel free to ask question directly by email or by creating github issue.
+
+## Latest releases (including binaries)
+
+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-http-header-filter=[]: A regexp to match a specific header against. Requests with non-matching headers will be dropped:
+    gor --input-raw :8080 --output-http staging.com --output-http-header-filter api-version:^v1
+
+  -output-http-header-hash-filter=[]: Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific header. The fraction must have a denominator that is a power of two:
+    gor --input-raw :8080 --output-http staging.com --output-http-header-hash-filter user-id:1/4
+
+  -output-http-url-regexp=: A regexp to match requests against. Anything else will be dropped:
+    gor --input-raw :8080 --output-http staging.com --output-http-url-regexp ^www.
+    
+  -output-http-workers=-1: Number of http output workers desired. Use default -1 for dynamic worker scaling.  Gor will add http workers if its work queue starts getting too full and kill them .
+    
+  -output-http-stats=false: If set to `true` it gives out queuing stats for the HTTP output every 5 seconds in the form latest,mean,max,count,count/second.
+    
+  -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
+    
+  -output-tcp-stats=false: If set to `true` it gives out queuing stats for the TCP output every 5 seconds in the form latest,mean,max,count,count/second.
+    
+  -split-output=false: By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.
+
+  -output-http-rewrite-url=[]: Rewrites the url in the request based on a mapping
+    gor --input-raw :8080 --output-http staging.com --output-http-rewrite-url /xml_test/interface.php:/api/service.do
+```
+
+## 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`.
+3. `cd $GOPATH/src/github.com/buger/gor`
+4. `go build` to get binary, or `go test` to run tests
+
+## Development
+Project contains Docker environment.
+
+1. Build container: `make dbuild`
+2. Run all tests: `make dtest`. Run specific test: `make dtest ARGS=-test.run=**regexp**`
+3. Bash access to container: `make dbash`. Inside container you have python to run simple web server `python -m SimpleHTTPServer 8080` and `curl` to make http requests. 
+
+
+## Questions and support 
+
+All bug-reports and suggestions should go though Github Issues or our [Google Group](https://groups.google.com/forum/#!forum/gor-users). Or you can just send email to gor-users@googlegroups.com
+
+If you have some private questions you can send direct mail to leonsbox@gmail.com
+
+## 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 `--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.
+
+### 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
+
+### What do the stats commands do?
+Gor can report stats on the output-tcp and output-http request queues. Stats are reported to the console every 5 seconds in the form `latest,mean,max,count,count/second` by using the `-output-http-stats` and `-output-tcp-stats` options.
+
+Examples:
+
+```
+2014/04/23 21:17:50 output_tcp:latest,mean,max,count,count/second
+2014/04/23 21:17:50 output_tcp:0,0,0,0,0
+2014/04/23 21:17:55 output_tcp:1,1,2,68,13
+2014/04/23 21:18:00 output_tcp:1,1,2,92,18
+2014/04/23 21:18:05 output_tcp:1,1,2,119,23
+2014/04/23 21:18:10 output_tcp:1,0,1,95,19
+2014/04/23 21:18:15 output_tcp:1,1,2,92,18
+2014/04/23 21:18:20 output_tcp:1,1,2,108,21
+2014/04/23 21:18:25 output_tcp:1,1,2,117,23
+2014/04/23 21:18:30 output_tcp:1,1,2,113,22
+2014/04/23 21:18:35 output_tcp:21,20,21,132,26
+2014/04/23 21:18:40 output_tcp:100,99,100,99,19
+```
+
+```
+Version: 0.8
+2014/04/23 21:19:46 output_http:latest,mean,max,count,count/second
+2014/04/23 21:19:46 output_http:0,0,0,0,0
+2014/04/23 21:19:51 output_http:0,0,0,0,0
+2014/04/23 21:19:56 output_http:0,0,0,0,0
+2014/04/23 21:20:01 output_http:1,0,1,50,10
+2014/04/23 21:20:06 output_http:1,1,4,72,14
+2014/04/23 21:20:11 output_http:1,0,1,179,35
+2014/04/23 21:20:16 output_http:1,0,1,148,29
+2014/04/23 21:20:21 output_http:1,1,2,91,18
+2014/04/23 21:20:26 output_http:1,1,2,150,30
+2014/04/23 21:18:15 output_http:100,99,100,70,14
+2014/04/23 21:18:21 output_http:100,99,100,55,11
+2014/04/23 21:18:28 output_http:100,99,100,55,11
+2014/04/23 21:18:34 output_http:100,99,100,57,11
+2014/04/23 21:18:41 output_http:100,99,100,61,12
+2014/04/23 21:18:48 output_http:100,99,100,56,11
+2014/04/23 21:18:56 output_http:100,99,100,58,11
+2014/04/23 21:19:01 output_http:100,99,100,31,6
+2014/04/23 21:19:08 output_http:100,99,100,61,12
+2014/04/23 21:19:15 output_http:100,99,100,64,12
+2014/04/23 21:19:21 output_http:100,99,100,70,14
+2014/04/23 21:19:28 output_http:100,99,100,61,12
+2014/04/23 21:19:35 output_http:100,99,100,56,11
+```
+
+### How can I tell if I have bottlenecks?
+Key areas that sometimes experience bottlenecks are the output-tcp and output-http functions which have internal queues for requests. Each queue has an upper limit of 100. Enable stats reporting to see if any queues are experiencing bottleneck behavior.
+ 
+#### output-http bottlenecks
+When running a Gor replay the output-http feature may bottleneck if:
+
+  * the replay has inadequate bandwidth. If the replay is receiving or sending more messages than its network adapter can handle the output-http-stats  may report that the output-http queue is filling up. See if there is a way to upgrade the replay's bandwidth.
+  * with `--output-http-workers` set to anything other than `-1` the `-output-http` target is unable to respond to messages in a timely manner. The http output workers which take messages off the output-http queue, process the request, and ensure that the request did not result in an error may not be able to keep up with the number of incoming requests. If the replay is not using dynamic worker scaling (`--output-http-workers=-1`)  The optimal number of output-http-workers can be determined with the formula `output-workers = (Average number of requests per second)/(Average target response time per second)`.
+
+#### output-tcp bottlenecks
+When using the Gor listener the output-tcp feature may bottleneck if:
+
+  * the replay is unable to accept and process more requests than the listener is able generate. Prior to troubleshooting the output-tcp bottleneck, ensure that the replay target is not experiencing any bottlenecks. 
+  * the replay target has inadequate bandwidth to handle all its incoming requests.  If a replay target's incoming bandwidth is maxed out the output-tcp-stats may report that the output-tcp queue is filling up. See if there is a way to upgrade the replay's bandwidth.
+
+### The CPU average across my load-balanced targets is higher than the source
+If you are replaying traffic from multiple listeners to a load-balanced target and you use sticky sessions, you may observe that the target servers have a higher CPU load than the listener servers. This may be because the sticky session cookie of the original load balancer is not honored by the target load balancer thus resulting in requests that would normally hit the same target server hitting different servers on the backend thus reducing some caching benefits gained via the load balancing.  Try running just one listener against one replay target and see if the CPU utilization comparison is more accurate.
+ 
+### How does dynamic http worker scaling work?
+By using the Gor setting `--output-http-workers=-1` Gor will create more http output workers when the http output queue length is greater than 10.  The number of workers created (N) is equal to the queue length at the time which it is checked and found to have a length greater than 10. The queue length is checked every time a message is written to the http output queue.  No more workers will be spawned until that request to spawn N workers is satisfied.  If a dynamic worker cannot process a message at that time, it will sleep for 100 milliseconds. If a dynamic worker cannot process a message for 2 seconds it dies.  
+
+## 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
+
+1. Fork it
+2. Create your feature branch (git checkout -b my-new-feature)
+3. Commit your changes (git commit -am 'Added some feature')
+4. Push to the branch (git push origin my-new-feature)
+5. Create new Pull Request
+
+## Companies using Gor
+
+* [Granify](http://granify.com)
+* [GOV.UK](https://www.gov.uk) ([Government Digital Service](http://digital.cabinetoffice.gov.uk/))
+* [theguardian.com](http://theguardian.com)
+* [TomTom](http://www.tomtom.com/)
+* [3SCALE](http://www.3scale.net/)
+* [Optionlab](http://www.opinionlab.com)
+* To add your company drop me a line to github.com/buger or leonsbox@gmail.com

emitter.go

@@ -14,19 +14,19 @@ func Start(stop chan int) {
 		select {
 		case <-stop:
 			return
-		case <- time.After(1 * time.Second):
+		case <-time.After(1 * time.Second):
 		}
 	}
 }
 
 // Copy from 1 reader to multiple writers
 func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
-	buf := make([]byte, 32*1024)
+	buf := make([]byte, 5*1024*1024)
 	wIndex := 0
 
 	for {
 		nr, er := src.Read(buf)
-		if nr > 0 && len(buf) > nr{
+		if nr > 0 && len(buf) > nr {
 			Debug("Sending", src, ": ", string(buf[0:nr]))
 
 			if Settings.splitOutput {

gor_stat.go

@@ -1,9 +1,9 @@
 package main
 
 import (
-	"time"
 	"log"
 	"strconv"
+	"time"
 )
 
 const (
@@ -11,11 +11,11 @@ const (
 )
 
 type GorStat struct {
-	statName	string
-	latest		int
-	mean	 	int
-	max			int
-	count		int
+	statName string
+	latest   int
+	mean     int
+	max      int
+	count    int
 }
 
 func NewGorStat(statName string) (s *GorStat) {
@@ -54,14 +54,13 @@ func (s *GorStat) Reset() {
 }
 
 func (s *GorStat) String() string {
-	return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count / rate)
+	return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count/rate)
 }
 
 func (s *GorStat) reportStats() {
 	for {
-			log.Println(s)
-			s.Reset()
-			time.Sleep(rate * time.Second)
+		log.Println(s)
+		s.Reset()
+		time.Sleep(rate * time.Second)
 	}
 }
-

input_file.go

@@ -8,15 +8,17 @@ import (
 )
 
 type FileInput struct {
-	data    chan []byte
-	path    string
-	decoder *gob.Decoder
+	data        chan []byte
+	path        string
+	decoder     *gob.Decoder
+	speedFactor float64
 }
 
 func NewFileInput(path string) (i *FileInput) {
 	i = new(FileInput)
 	i.data = make(chan []byte)
 	i.path = path
+	i.speedFactor = 1
 	i.Init(path)
 
 	go i.emit()
@@ -57,8 +59,16 @@ func (i *FileInput) emit() {
 		}
 
 		if lastTime != 0 {
-			time.Sleep(time.Duration(raw.Timestamp - lastTime))
+			timeDiff := raw.Timestamp - lastTime
+
+			// We can speedup or slowdown execution based on speedFactor
+			if i.speedFactor != 1 {
+				timeDiff = int64(float64(raw.Timestamp-lastTime) / i.speedFactor)
+			}
+
+			time.Sleep(time.Duration(timeDiff))
 		}
+
 		lastTime = raw.Timestamp
 
 		i.data <- raw.Request

input_http.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"log"
+	"net"
+	"net/http"
+	"net/http/httputil"
+)
+
+type HTTPInput struct {
+	data     chan []byte
+	address  string
+	listener net.Listener
+}
+
+func NewHTTPInput(address string) (i *HTTPInput) {
+	i = new(HTTPInput)
+	i.data = make(chan []byte)
+	i.address = address
+
+	i.listen(address)
+
+	return
+}
+
+func (i *HTTPInput) Read(data []byte) (int, error) {
+	buf := <-i.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (i *HTTPInput) handler(w http.ResponseWriter, r *http.Request) {
+	buf, _ := httputil.DumpRequest(r, true)
+
+	i.data <- buf
+
+	http.Error(w, http.StatusText(200), 200)
+}
+
+func (i *HTTPInput) listen(address string) {
+	var err error
+
+	mux := http.NewServeMux()
+
+	mux.HandleFunc("/", i.handler)
+
+	i.listener, err = net.Listen("tcp", address)
+	if err != nil {
+		log.Fatal("HTTP input listener failure:", err)
+	}
+
+	go func() {
+		err = http.Serve(i.listener, mux)
+		if err != nil {
+			log.Fatal("HTTP input serve failure:", err)
+		}
+	}()
+}
+
+func (i *HTTPInput) String() string {
+	return "HTTP input: " + i.address
+}

input_http_test.go

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

input_tcp.go

@@ -61,7 +61,7 @@ func (i *TCPInput) handleConnection(conn net.Conn) {
 	reader := bufio.NewReader(conn)
 
 	for {
-		buf,err := reader.ReadBytes('¶')
+		buf, err := reader.ReadBytes('¶')
 		buf_len := len(buf)
 		if buf_len > 0 {
 			new_buf_len := len(buf) - 2

input_tcp_test.go

@@ -38,8 +38,8 @@ func TestTCPInput(t *testing.T) {
 
 	for i := 0; i < 100; i++ {
 		wg.Add(1)
-		new_buf := make([]byte, len(msg) + 2)
-		msg = append(msg,[]byte("¶")...)
+		new_buf := make([]byte, len(msg)+2)
+		msg = append(msg, []byte("¶")...)
 		copy(new_buf, msg)
 		conn.Write(new_buf)
 	}
@@ -82,8 +82,8 @@ func BenchmarkTCPInput(b *testing.B) {
 			for {
 				data := <-dataChan
 
-				new_buf := make([]byte, len(data) + 2)
-				data = append(data,[]byte("¶")...)
+				new_buf := make([]byte, len(data)+2)
+				data = append(data, []byte("¶")...)
 				copy(new_buf, data)
 				conn.Write(new_buf)
 			}

limiter.go

@@ -3,43 +3,91 @@ package main
 import (
 	"fmt"
 	"io"
+	"math/rand"
+	"strconv"
+	"strings"
 	"time"
 )
 
 type Limiter struct {
-	writer io.Writer
-	limit  int
+	plugin    interface{}
+	limit     int
+	isPercent bool
 
 	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()
+func parseLimitOptions(options string) (limit int, isPercent bool) {
+	if strings.Contains(options, "%") {
+		limit, _ = strconv.Atoi(strings.Split(options, "%")[0])
+		isPercent = true
+	} else {
+		limit, _ = strconv.Atoi(options)
+		isPercent = false
+	}
 
 	return
 }
 
-func (l *Limiter) Write(data []byte) (n int, err error) {
+func NewLimiter(plugin interface{}, options string) io.ReadWriter {
+	l := new(Limiter)
+	l.limit, l.isPercent = parseLimitOptions(options)
+	l.plugin = plugin
+	l.currentTime = time.Now().UnixNano()
+
+	// FileInput have its own rate limiting. Unlike other inputs we not just dropping requests, we can slow down or speed up request emittion.
+	if fi, ok := l.plugin.(*FileInput); ok && l.isPercent {
+		fi.speedFactor = float64(l.limit) / float64(100)
+	}
+
+	return l
+}
+
+func (l *Limiter) isLimited() bool {
+	// File input have its own limiting algorithm
+	if _, ok := l.plugin.(*FileInput); ok && l.isPercent {
+		return false
+	}
+
+	if l.isPercent {
+		return l.limit <= rand.Intn(100)
+	}
+
 	if (time.Now().UnixNano() - l.currentTime) > time.Second.Nanoseconds() {
 		l.currentTime = time.Now().UnixNano()
 		l.currentRPS = 0
 	}
 
 	if l.currentRPS >= l.limit {
+		return true
+	}
+
+	l.currentRPS++
+
+	return false
+}
+
+func (l *Limiter) Write(data []byte) (n int, err error) {
+	if l.isLimited() {
 		return 0, nil
 	}
 
-	n, err = l.writer.Write(data)
+	n, err = l.plugin.(io.Writer).Write(data)
 
-	l.currentRPS++
+	return
+}
+
+func (l *Limiter) Read(data []byte) (n int, err error) {
+	n, err = l.plugin.(io.Reader).Read(data)
+
+	if l.isLimited() {
+		return 0, nil
+	}
 
 	return
 }
 
 func (l *Limiter) String() string {
-	return fmt.Sprintf("Limiting %s to: %d", l.writer, l.limit)
+	return fmt.Sprintf("Limiting %s to: %d (isPercent: %b)", l.plugin, l.limit, l.isPercent)
 }

limiter_test.go

@@ -1,3 +1,5 @@
+// +build !race
+
 package main
 
 import (
@@ -6,14 +8,14 @@ import (
 	"testing"
 )
 
-func TestLimiter(t *testing.T) {
+func TestOutputLimiter(t *testing.T) {
 	wg := new(sync.WaitGroup)
 	quit := make(chan int)
 
 	input := NewTestInput()
 	output := NewLimiter(NewTestOutput(func(data []byte) {
 		wg.Done()
-	}), 10)
+	}), "10")
 	wg.Add(10)
 
 	Plugins.Inputs = []io.Reader{input}
@@ -29,3 +31,76 @@ func TestLimiter(t *testing.T) {
 
 	close(quit)
 }
+
+func TestInputLimiter(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewLimiter(NewTestInput(), "10")
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+	wg.Add(10)
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		input.(*Limiter).plugin.(*TestInput).EmitGET()
+	}
+
+	wg.Wait()
+
+	close(quit)
+}
+
+// Should limit all requests
+func TestPercentLimiter1(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+	output := NewLimiter(NewTestOutput(func(data []byte) {
+		wg.Done()
+	}), "0%")
+
+	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)
+}
+
+// Should not limit at all
+func TestPercentLimiter2(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	input := NewTestInput()
+	output := NewLimiter(NewTestOutput(func(data []byte) {
+		wg.Done()
+	}), "100%")
+	wg.Add(100)
+
+	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)
+}

output_file.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"encoding/gob"
+	"io"
 	"log"
 	"os"
 	"time"
@@ -18,12 +19,12 @@ type FileOutput struct {
 	file    *os.File
 }
 
-func NewFileOutput(path string) (o *FileOutput) {
-	o = new(FileOutput)
+func NewFileOutput(path string) io.Writer {
+	o := new(FileOutput)
 	o.path = path
 	o.Init(path)
 
-	return
+	return o
 }
 
 func (o *FileOutput) Init(path string) {

output_http.go

@@ -2,13 +2,13 @@ package main
 
 import (
 	"bufio"
-	"bytes"	
+	"bytes"
 	"io"
 	"log"
 	"net/http"
 	"net/url"
-	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 )
 
@@ -36,32 +36,33 @@ func ParseRequest(data []byte) (request *http.Request, err error) {
 	return
 }
 
+const InitialDynamicWorkers = 10
+
 type HTTPOutput struct {
 	address string
 	limit   int
-	buf     chan []byte
-	needWorker chan int
+	queue   chan []byte
 
-	urlRegexp             HTTPUrlRegexp
-	headerFilters         HTTPHeaderFilters
-	headerHashFilters     HTTPHeaderHashFilters
-    outputHTTPUrlRewrite  UrlRewriteMap
+	activeWorkers int64
+	needWorker    chan int
+
+	urlRegexp            HTTPUrlRegexp
+	headerFilters        HTTPHeaderFilters
+	headerHashFilters    HTTPHeaderHashFilters
+	outputHTTPUrlRewrite UrlRewriteMap
 
 	headers HTTPHeaders
 	methods HTTPMethods
 
 	elasticSearch *ESPlugin
 
-	bufStats *GorStat
+	queueStats *GorStat
 }
 
-func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, urlRegexp HTTPUrlRegexp, headerFilters HTTPHeaderFilters, headerHashFilters HTTPHeaderHashFilters, elasticSearchAddr string, outputHTTPUrlRewrite UrlRewriteMap) io.Writer {
+func NewHTTPOutput(address string, headers HTTPHeaders, methods HTTPMethods, urlRegexp HTTPUrlRegexp, headerFilters HTTPHeaderFilters, headerHashFilters HTTPHeaderHashFilters, elasticSearchAddr string, outputHTTPUrlRewrite UrlRewriteMap) io.Writer {
 
 	o := new(HTTPOutput)
 
-	optionsArr := strings.Split(options, "|")
-	address := optionsArr[0]
-
 	if !strings.HasPrefix(address, "http") {
 		address = "http://" + address
 	}
@@ -75,12 +76,18 @@ func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, url
 	o.headerHashFilters = headerHashFilters
 	o.outputHTTPUrlRewrite = outputHTTPUrlRewrite
 
-	o.buf = make(chan []byte, 100)
+	o.queue = make(chan []byte, 100)
 	if Settings.outputHTTPStats {
-		o.bufStats = NewGorStat("output_http")
+		o.queueStats = NewGorStat("output_http")
 	}
+
+	o.needWorker = make(chan int, 1)
+
+	// Initial workers count
 	if Settings.outputHTTPWorkers == -1 {
-		o.needWorker = make(chan int)
+		o.needWorker <- InitialDynamicWorkers
+	} else {
+		o.needWorker <- Settings.outputHTTPWorkers
 	}
 
 	if elasticSearchAddr != "" {
@@ -88,30 +95,21 @@ func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, url
 		o.elasticSearch.Init(elasticSearchAddr)
 	}
 
-	if len(optionsArr) > 1 {
-		o.limit, _ = strconv.Atoi(optionsArr[1])
-	}
+	go o.WorkerMaster()
 
-	go o.WorkerMaster(Settings.outputHTTPWorkers)
-
-	if o.limit > 0 {
-		return NewLimiter(o, o.limit)
-	} else {
-		return o
-	}
+	return o
 }
 
-func (o *HTTPOutput) WorkerMaster(n int) {
-	for i := 0; i < n; i++ {
-		go o.Worker()
-	}
+func (o *HTTPOutput) WorkerMaster() {
+	for {
+		new_workers := <-o.needWorker
+		for i := 0; i < new_workers; i++ {
+			go o.Worker()
+		}
 
-	if Settings.outputHTTPWorkers == -1 {
-		for {
-			new_workers := <-o.needWorker
-			for i := 0; i < new_workers; i++ {
-				go o.Worker()
-			}
+		// Disable dynamic scaling if workers poll fixed size
+		if Settings.outputHTTPWorkers != -1 {
+			return
 		}
 	}
 }
@@ -120,44 +118,55 @@ func (o *HTTPOutput) Worker() {
 	client := &http.Client{
 		CheckRedirect: customCheckRedirect,
 	}
-	death_count := 0
-	Loop:
-		for {
-			select {
-				case data := <-o.buf:
-				o.sendRequest(client, data)
-				death_count = 0
-			default:
-				if Settings.outputHTTPWorkers == -1 {
-					death_count += 1
-				}
-				if death_count > 20 {
-					break Loop
-				} else {
-					time.Sleep(time.Millisecond * 100)
-				}
 
+	death_count := 0
+
+	atomic.AddInt64(&o.activeWorkers, 1)
+
+	for {
+		select {
+		case data := <-o.queue:
+			o.sendRequest(client, data)
+			death_count = 0
+		case <-time.After(time.Millisecond * 100):
+			// When dynamic scaling enabled workers die after 2s of inactivity
+			if Settings.outputHTTPWorkers == -1 {
+				death_count += 1
+			} else {
+				continue
+			}
+
+			if death_count > 20 {
+				workersCount := atomic.LoadInt64(&o.activeWorkers)
+
+				// At least 1 worker should be alive
+				if workersCount != 1 {
+					atomic.AddInt64(&o.activeWorkers, -1)
+					return
+				}
 			}
 		}
+	}
 }
 
 func (o *HTTPOutput) Write(data []byte) (n int, err error) {
 	buf := make([]byte, len(data))
 	copy(buf, data)
 
-	o.buf <- buf
-	buf_len := len(o.buf)
+	o.queue <- buf
+
 	if Settings.outputHTTPStats {
-		o.bufStats.Write(len(o.buf))
+		o.queueStats.Write(len(o.queue))
 	}
 
 	if Settings.outputHTTPWorkers == -1 {
-		if buf_len > 10 {
-			if len(o.needWorker) == 0 {
-				o.needWorker <- buf_len
-			}
+		workersCount := atomic.LoadInt64(&o.activeWorkers)
+
+		if len(o.queue) > int(workersCount) {
+			o.needWorker <- len(o.queue)
 		}
 	}
+
 	return len(data), nil
 }
 
@@ -177,8 +186,8 @@ func (o *HTTPOutput) sendRequest(client *http.Client, data []byte) {
 		return
 	}
 
-    // Rewrite the path as necessary
-    request.URL.Path = o.outputHTTPUrlRewrite.Rewrite(request.URL.Path)
+	// Rewrite the path as necessary
+	request.URL.Path = o.outputHTTPUrlRewrite.Rewrite(request.URL.Path)
 
 	// Change HOST of original request
 	URL := o.address + request.URL.Path + "?" + request.URL.RawQuery
@@ -213,7 +222,6 @@ func (o *HTTPOutput) sendRequest(client *http.Client, data []byte) {
 }
 
 func SetHeader(request *http.Request, name string, value string) {
-
 	// Need to check here for the Host header as it needs to be set on the request and not as a separate header
 	// http.ReadRequest sets it by default to the URL Host of the request being read
 	if name == "Host" {

output_http_test.go

@@ -96,7 +96,7 @@ func BenchmarkHTTPOutput(b *testing.B) {
 		wg.Done()
 	})
 
-	output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}, "",UrlRewriteMap{})
+	output := NewHTTPOutput(listener.Addr().String(), headers, methods, HTTPUrlRegexp{}, HTTPHeaderFilters{}, HTTPHeaderHashFilters{}, "", UrlRewriteMap{})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{output}

output_tcp.go

@@ -5,41 +5,30 @@ import (
 	"io"
 	"log"
 	"net"
-	"strconv"
-	"strings"
 )
 
 type TCPOutput struct {
-	address string
-	limit   int
-	buf     chan []byte
+	address  string
+	limit    int
+	buf      chan []byte
 	bufStats *GorStat
 }
 
-func NewTCPOutput(options string) io.Writer {
+func NewTCPOutput(address string) io.Writer {
 	o := new(TCPOutput)
 
-	optionsArr := strings.Split(options, "|")
-	o.address = optionsArr[0]
+	o.address = address
 
 	o.buf = make(chan []byte, 100)
 	if Settings.outputTCPStats {
 		o.bufStats = NewGorStat("output_tcp")
 	}
 
-	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
-	}
+	return o
 }
 
 func (o *TCPOutput) worker() {
@@ -52,8 +41,8 @@ func (o *TCPOutput) worker() {
 }
 
 func (o *TCPOutput) Write(data []byte) (n int, err error) {
-	new_buf := make([]byte, len(data) + 2)
-	data = append(data,[]byte("¶")...)
+	new_buf := make([]byte, len(data)+2)
+	data = append(data, []byte("¶")...)
 	copy(new_buf, data)
 	o.buf <- new_buf
 	if Settings.outputTCPStats {

output_tcp_test.go

@@ -44,11 +44,11 @@ func startTCP(cb func([]byte)) net.Listener {
 	go func() {
 		for {
 			conn, _ := listener.Accept()
-			
+
 			go func() {
 				reader := bufio.NewReader(conn)
 				for {
-					buf,err := reader.ReadBytes('¶')
+					buf, err := reader.ReadBytes('¶')
 					new_buf_len := len(buf) - 2
 					new_buf := make([]byte, new_buf_len)
 					copy(new_buf, buf[:new_buf_len])

plugins.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"io"
+	"reflect"
+	"strings"
 )
 
 type InOutPlugins struct {
@@ -9,38 +11,91 @@ type InOutPlugins struct {
 	Outputs []io.Writer
 }
 
+type ReaderOrWriter interface{}
+
 var Plugins *InOutPlugins = new(InOutPlugins)
 
+func extractLimitOptions(options string) (string, string) {
+	split := strings.Split(options, "|")
+
+	if len(split) > 1 {
+		return split[0], split[1]
+	} else {
+		return split[0], ""
+	}
+}
+
+// Automatically detects type of plugin and initialize it
+//
+// See this article if curious about relfect stuff below: http://blog.burntsushi.net/type-parametric-functions-golang
+func registerPlugin(constructor interface{}, options ...interface{}) {
+	vc := reflect.ValueOf(constructor)
+
+	// Pre-processing options to make it work with reflect
+	vo := []reflect.Value{}
+	for _, oi := range options {
+		vo = append(vo, reflect.ValueOf(oi))
+	}
+
+	// Removing limit options from path
+	path, limit := extractLimitOptions(vo[0].String())
+
+	// Writing value back without limiter "|" options
+	vo[0] = reflect.ValueOf(path)
+
+	// Calling our constructor with list of given options
+	plugin := vc.Call(vo)[0].Interface()
+	plugin_wrapper := plugin
+
+	if limit != "" {
+		plugin_wrapper = NewLimiter(plugin, limit)
+	} else {
+		plugin_wrapper = plugin
+	}
+
+	if _, ok := plugin.(io.Reader); ok {
+		Plugins.Inputs = append(Plugins.Inputs, plugin_wrapper.(io.Reader))
+	}
+
+	if _, ok := plugin.(io.Writer); ok {
+		Plugins.Outputs = append(Plugins.Outputs, plugin_wrapper.(io.Writer))
+	}
+}
+
 func InitPlugins() {
 	for _, options := range Settings.inputDummy {
-		Plugins.Inputs = append(Plugins.Inputs, NewDummyInput(options))
+		registerPlugin(NewDummyInput, options)
 	}
 
 	for _, options := range Settings.outputDummy {
-		Plugins.Outputs = append(Plugins.Outputs, NewDummyOutput(options))
+		registerPlugin(NewDummyOutput, options)
 	}
 
 	for _, options := range Settings.inputRAW {
-		Plugins.Inputs = append(Plugins.Inputs, NewRAWInput(options))
+		registerPlugin(NewRAWInput, options)
 	}
 
 	for _, options := range Settings.inputTCP {
-		Plugins.Inputs = append(Plugins.Inputs, NewTCPInput(options))
+		registerPlugin(NewTCPInput, options)
 	}
 
 	for _, options := range Settings.outputTCP {
-		Plugins.Outputs = append(Plugins.Outputs, NewTCPOutput(options))
+		registerPlugin(NewTCPOutput, options)
 	}
 
 	for _, options := range Settings.inputFile {
-		Plugins.Inputs = append(Plugins.Inputs, NewFileInput(options))
+		registerPlugin(NewFileInput, options)
 	}
 
 	for _, options := range Settings.outputFile {
-		Plugins.Outputs = append(Plugins.Outputs, NewFileOutput(options))
+		registerPlugin(NewFileOutput, options)
+	}
+
+	for _, options := range Settings.inputHTTP {
+		registerPlugin(NewHTTPInput, options)
 	}
 
 	for _, options := range Settings.outputHTTP {
-		Plugins.Outputs = append(Plugins.Outputs, NewHTTPOutput(options, Settings.outputHTTPHeaders, Settings.outputHTTPMethods, Settings.outputHTTPUrlRegexp, Settings.outputHTTPHeaderFilters, Settings.outputHTTPHeaderHashFilters, Settings.outputHTTPElasticSearch, Settings.outputHTTPUrlRewrite))
+		registerPlugin(NewHTTPOutput, options, Settings.outputHTTPHeaders, Settings.outputHTTPMethods, Settings.outputHTTPUrlRegexp, Settings.outputHTTPHeaderFilters, Settings.outputHTTPHeaderHashFilters, Settings.outputHTTPElasticSearch, Settings.outputHTTPUrlRewrite)
 	}
 }

plugins_test.go

@@ -0,0 +1,47 @@
+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()
+
+	if len(Plugins.Inputs) != 2 {
+		t.Errorf("Should be 2 inputs %d", len(Plugins.Inputs))
+	}
+
+	if _, ok := Plugins.Inputs[0].(*DummyInput); !ok {
+		t.Errorf("First input should be DummyInput")
+	}
+
+	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 _, ok := Plugins.Outputs[0].(*DummyOutput); !ok {
+		t.Errorf("First output should be DummyOutput")
+	}
+
+	if l, ok := Plugins.Outputs[1].(*Limiter); ok {
+		if _, ok := l.plugin.(*HTTPOutput); !ok {
+			t.Errorf("HTTPOutput should be wrapped in limiter")
+		}
+	} else {
+		t.Errorf("Second output should be Limiter")
+	}
+
+}

raw_socket_listener/listener.go

@@ -1,10 +1,10 @@
 package raw_socket
 
 import (
-	"encoding/binary"
-	"log"
-	"net"
-	"strconv"
+    "encoding/binary"
+    "log"
+    "net"
+    "strconv"
 )
 
 // Capture traffic from socket using RAW_SOCKET's
@@ -15,127 +15,127 @@ import (
 // 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
+    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_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
+    c_del_message chan *TCPMessage // Used for notifications about completed or expired messages
 
-	addr string // IP to listen
-	port int    // Port to listen
+    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 = &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.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)
+    rawListener.addr = addr
+    rawListener.port, _ = strconv.Atoi(port)
 
-	go rawListener.listen()
-	go rawListener.readRAWSocket()
+    go rawListener.listen()
+    go rawListener.readRAWSocket()
 
-	return
+    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)
+    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)
-		}
-	}
+        // 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()
+    conn, e := net.ListenPacket("ip4:tcp", t.addr)
+    defer conn.Close()
 
-	if e != nil {
-		log.Fatal(e)
-	}
+    if e != nil {
+        log.Fatal(e)
+    }
 
-	buf := make([]byte, 4096*2)
+    buf := make([]byte, 4096*2)
 
-	for {
-		// Note: ReadFrom receive messages without IP header
-		n, addr, err := conn.ReadFrom(buf)
+    for {
+        // Note: ReadFrom receive messages without IP header
+        n, addr, err := conn.ReadFrom(buf)
 
-		if err != nil {
-			log.Println("Error:", err)
-			continue
-		}
+        if err != nil {
+            log.Println("Error:", err)
+            continue
+        }
 
-		if n > 0 {
-			t.parsePacket(addr, buf[:n])
-		}
-	}
+        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)
+    if t.isIncomingDataPacket(buf) {
+        new_buf := make([]byte, len(buf))
+        copy(new_buf, buf)
 
-		t.c_packets <- ParseTCPPacket(addr, new_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])
+    // 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
+    // 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
-		}
-	}
+        // 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
+    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() }()
+    defer func() { recover() }()
 
-	var message *TCPMessage
-	m_id := packet.Addr.String() + strconv.Itoa(int(packet.Ack))
+    var message *TCPMessage
+    m_id := packet.Addr.String() + strconv.Itoa(int(packet.Ack))
 
-	message, ok := t.messages[m_id]
+    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
-	}
+    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
+    // Adding packet to message
+    message.c_packets <- packet
 }
 
 // Receive TCP messages from the listener channel
 func (t *Listener) Receive() *TCPMessage {
-	return <-t.c_messages
+    return <-t.c_messages
 }

raw_socket_listener/tcp_message.go

@@ -1,9 +1,9 @@
 package raw_socket
 
 import (
-	"log"
-	"sort"
-	"time"
+    "log"
+    "sort"
+    "time"
 )
 
 const MSG_EXPIRE = 2000 * time.Millisecond
@@ -15,80 +15,91 @@ const MSG_EXPIRE = 2000 * time.Millisecond
 // Message can be compiled from unique packets with same message_id which sorted by sequence
 // Message is received if we didn't receive any packets for 2000ms
 type TCPMessage struct {
-	ID      string // Message ID
-	packets []*TCPPacket
+    ID      string // Message ID
+    packets []*TCPPacket
 
-	timer *time.Timer // Used for expire check
+    timer *time.Timer // Used for expire check
 
-	c_packets chan *TCPPacket
+    c_packets chan *TCPPacket
 
-	c_del_message chan *TCPMessage
+    c_del_message chan *TCPMessage
 }
 
 // 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 = &TCPMessage{ID: ID}
 
-	msg.c_packets = make(chan *TCPPacket)
-	msg.c_del_message = c_del // used for notifying that message completed or expired
+    msg.c_packets = make(chan *TCPPacket)
+    msg.c_del_message = c_del // used for notifying that message completed or expired
 
-	// Every time we receive packet we reset this timer
-	msg.timer = time.AfterFunc(MSG_EXPIRE, msg.Timeout)
+    // Every time we receive packet we reset this timer
+    msg.timer = time.AfterFunc(MSG_EXPIRE, msg.Timeout)
 
-	go msg.listen()
+    go msg.listen()
 
-	return
+    return
 }
 
 func (t *TCPMessage) listen() {
-	for {
-		select {
-		case packet, more := <-t.c_packets:
-			if more {
-				t.AddPacket(packet)
-			} else {
-				// Stop loop if channel closed
-				return
-			}
-		}
-	}
+    for {
+        select {
+        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() {
-	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
+    select {
+        // In some cases Timeout can be called multiple times (do not know how yet)
+        // Ensure that we did not close channel 2 times
+        case packet, ok := <- t.c_packets:
+            if ok {
+                t.AddPacket(packet)
+            } else {
+                return
+            }
+        default:
+            close(t.c_packets)
+            t.c_del_message <- t // Notify RAWListener that message is ready to be send to replay server
+    }
 }
 
 // Bytes sorts packets in right orders and return message content
 func (t *TCPMessage) Bytes() (output []byte) {
-	sort.Sort(BySeq(t.packets))
+    sort.Sort(BySeq(t.packets))
 
-	for _, v := range t.packets {
-		output = append(output, v.Data...)
-	}
+    for _, v := range t.packets {
+        output = append(output, v.Data...)
+    }
 
-	return
+    return
 }
 
 // AddPacket to the message and ensure packet uniqueness
 // TCP allows that packet can be re-send multiple times
 func (t *TCPMessage) AddPacket(packet *TCPPacket) {
-	packetFound := false
+    packetFound := false
 
-	for _, pkt := range t.packets {
-		if packet.Seq == pkt.Seq {
-			packetFound = true
-			break
-		}
-	}
+    for _, pkt := range t.packets {
+        if packet.Seq == pkt.Seq {
+            packetFound = true
+            break
+        }
+    }
 
-	if packetFound {
-		log.Println("Received packet with same sequence")
-	} else {
-		t.packets = append(t.packets, packet)
-	}
+    if packetFound {
+        log.Println("Received packet with same sequence")
+    } else {
+        t.packets = append(t.packets, packet)
+    }
 
-	// Reset message timeout timer
-	t.timer.Reset(MSG_EXPIRE)
+    // Reset message timeout timer
+    t.timer.Reset(MSG_EXPIRE)
 }

raw_socket_listener/tcp_packet.go

@@ -1,100 +1,100 @@
 package raw_socket
 
 import (
-	"encoding/binary"
-	"net"
-	"strconv"
-	"strings"
+    "encoding/binary"
+    "net"
+    "strconv"
+    "strings"
 )
 
 // TCP Flags
 const (
-	TCP_FIN = 1 << iota
-	TCP_SYN
-	TCP_RST
-	TCP_PSH
-	TCP_ACK
-	TCP_URG
-	TCP_ECE
-	TCP_CWR
-	TCP_NS
+    TCP_FIN = 1 << iota
+    TCP_SYN
+    TCP_RST
+    TCP_PSH
+    TCP_ACK
+    TCP_URG
+    TCP_ECE
+    TCP_CWR
+    TCP_NS
 )
 
 // Simple TCP packet parser
 //
 // Packet structure: http://en.wikipedia.org/wiki/Transmission_Control_Protocol
 type TCPPacket struct {
-	SrcPort    uint16
-	DestPort   uint16
-	Seq        uint32
-	Ack        uint32
-	DataOffset uint8
-	Flags      uint16
-	Window     uint16
-	Checksum   uint16
-	Urgent     uint16
+    SrcPort    uint16
+    DestPort   uint16
+    Seq        uint32
+    Ack        uint32
+    DataOffset uint8
+    Flags      uint16
+    Window     uint16
+    Checksum   uint16
+    Urgent     uint16
 
-	Data []byte
+    Data []byte
 
-	Addr net.Addr
+    Addr net.Addr
 }
 
 func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) {
-	p = &TCPPacket{Data: b}
-	p.ParseBasic()
-	p.Addr = addr
+    p = &TCPPacket{Data: b}
+    p.ParseBasic()
+    p.Addr = addr
 
-	return p
+    return p
 }
 
 // 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.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.ParseBasic()
+    t.SrcPort = binary.BigEndian.Uint16(t.Data[0:2])
+    t.DestPort = binary.BigEndian.Uint16(t.Data[2: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])
 }
 
 // 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.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:]
+    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)),
-		"Acknowledgment:" + strconv.Itoa(int(t.Ack)),
-		"Header len:" + strconv.Itoa(int(t.DataOffset)),
+    return strings.Join([]string{
+        "Source port: " + strconv.Itoa(int(t.SrcPort)),
+        "Dest port:" + strconv.Itoa(int(t.DestPort)),
+        "Sequence:" + strconv.Itoa(int(t.Seq)),
+        "Acknowledgment:" + strconv.Itoa(int(t.Ack)),
+        "Header len:" + strconv.Itoa(int(t.DataOffset)),
 
-		"Flag ns:" + strconv.FormatBool(t.Flags&TCP_NS != 0),
-		"Flag crw:" + strconv.FormatBool(t.Flags&TCP_CWR != 0),
-		"Flag ece:" + strconv.FormatBool(t.Flags&TCP_ECE != 0),
-		"Flag urg:" + strconv.FormatBool(t.Flags&TCP_URG != 0),
-		"Flag ack:" + strconv.FormatBool(t.Flags&TCP_ACK != 0),
-		"Flag psh:" + strconv.FormatBool(t.Flags&TCP_PSH != 0),
-		"Flag rst:" + strconv.FormatBool(t.Flags&TCP_RST != 0),
-		"Flag syn:" + strconv.FormatBool(t.Flags&TCP_SYN != 0),
-		"Flag fin:" + strconv.FormatBool(t.Flags&TCP_FIN != 0),
+        "Flag ns:" + strconv.FormatBool(t.Flags&TCP_NS != 0),
+        "Flag crw:" + strconv.FormatBool(t.Flags&TCP_CWR != 0),
+        "Flag ece:" + strconv.FormatBool(t.Flags&TCP_ECE != 0),
+        "Flag urg:" + strconv.FormatBool(t.Flags&TCP_URG != 0),
+        "Flag ack:" + strconv.FormatBool(t.Flags&TCP_ACK != 0),
+        "Flag psh:" + strconv.FormatBool(t.Flags&TCP_PSH != 0),
+        "Flag rst:" + strconv.FormatBool(t.Flags&TCP_RST != 0),
+        "Flag syn:" + strconv.FormatBool(t.Flags&TCP_SYN != 0),
+        "Flag fin:" + strconv.FormatBool(t.Flags&TCP_FIN != 0),
 
-		"Window size:" + strconv.Itoa(int(t.Window)),
-		"Checksum:" + strconv.Itoa(int(t.Checksum)),
+        "Window size:" + strconv.Itoa(int(t.Window)),
+        "Checksum:" + strconv.Itoa(int(t.Checksum)),
 
-		"Data:" + string(t.Data),
-	}, "\n")
+        "Data:" + string(t.Data),
+    }, "\n")
 }
 
 type BySeq []*TCPPacket
 
-func (a BySeq) Len() int			{ return len(a) }
-func (a BySeq) Swap(i, j int)		{ a[i], a[j] = a[j], a[i] }
-func (a BySeq) Less(i, j int) bool	{ return a[i].Seq < a[j].Seq }
+func (a BySeq) Len() int            { return len(a) }
+func (a BySeq) Swap(i, j int)       { a[i], a[j] = a[j], a[i] }
+func (a BySeq) Less(i, j int) bool  { return a[i].Seq < a[j].Seq }

settings.go

@@ -8,7 +8,7 @@ import (
 )
 
 const (
-	VERSION = "0.9.0"
+	VERSION = "0.9.2"
 )
 
 type AppSettings struct {
@@ -20,8 +20,8 @@ type AppSettings struct {
 	inputDummy  MultiOption
 	outputDummy MultiOption
 
-	inputTCP  MultiOption
-	outputTCP MultiOption
+	inputTCP       MultiOption
+	outputTCP      MultiOption
 	outputTCPStats bool
 
 	inputFile  MultiOption
@@ -29,6 +29,7 @@ type AppSettings struct {
 
 	inputRAW MultiOption
 
+	inputHTTP                   MultiOption
 	outputHTTP                  MultiOption
 	outputHTTPHeaders           HTTPHeaders
 	outputHTTPMethods           HTTPMethods
@@ -69,6 +70,8 @@ func init() {
 
 	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.inputHTTP, "input-http", "Read requests from HTTP, should be explicitly sent from your application:\n\t# Listen for http on 9000\n\tgor --input-http :9000 --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")

settings_header_filters_test.go

@@ -1,8 +1,8 @@
 package main
 
 import (
-	"testing"
 	"net/http"
+	"testing"
 )
 
 func TestHTTPHeaderFilters(t *testing.T) {
@@ -29,7 +29,7 @@ func TestHTTPHeaderFilters(t *testing.T) {
 	req.Header.Add("Header2", ":")
 	req.Header.Add("Header3", "Irrelevant")
 
-	if(!filters.Good(&req)) {
+	if !filters.Good(&req) {
 		t.Error("Request should pass filters")
 	}
 }

settings_header_hash_filters_test.go

@@ -1,8 +1,8 @@
 package main
 
 import (
-	"testing"
 	"net/http"
+	"testing"
 )
 
 func TestHTTPHeaderHashFilters(t *testing.T) {
@@ -27,17 +27,17 @@ func TestHTTPHeaderHashFilters(t *testing.T) {
 	req.Header = make(map[string][]string)
 	req.Header.Add("Header1", "test3414")
 
-	if(filters.Good(&req)) {
+	if filters.Good(&req) {
 		t.Error("Request should not pass filters, Header2 does not exist")
 	}
 
 	req.Header.Add("Header2", "test2")
-	if(filters.Good(&req)) {
+	if filters.Good(&req) {
 		t.Error("Request should not pass filters, Header2 hash too high")
 	}
 
 	req.Header.Set("Header2", "test3414")
-	if(!filters.Good(&req)) {
+	if !filters.Good(&req) {
 		t.Error("Request should pass filters")
 	}
 }

settings_url_map.go

@@ -27,10 +27,10 @@ func (r *UrlRewriteMap) Set(value string) error {
 }
 
 func (r *UrlRewriteMap) Rewrite(path string) string {
-        for _, f := range *r {
-                if f.src == path {
-                        return f.target
-                }
-        }
-        return path
+	for _, f := range *r {
+		if f.src == path {
+			return f.target
+		}
+	}
+	return path
 }

settings_url_map_test.go

@@ -5,7 +5,7 @@ import (
 )
 
 func TestUrlRewriteMap(t *testing.T) {
-    var url string;
+	var url string
 
 	rewrites := UrlRewriteMap{}
 
@@ -16,12 +16,12 @@ func TestUrlRewriteMap(t *testing.T) {
 
 	url = "/abc"
 
-    if(rewrites.Rewrite(url) == url) {
-        t.Error("Request url should have been rewritten, wasn't")
-    }
+	if rewrites.Rewrite(url) == url {
+		t.Error("Request url should have been rewritten, wasn't")
+	}
 
-    url = "/wibble"
-    if(rewrites.Rewrite(url) != url) {
-        t.Error("Request url should not have been rewritten, was")
-    }
+	url = "/wibble"
+	if rewrites.Rewrite(url) != url {
+		t.Error("Request url should not have been rewritten, was")
+	}
 }

settings_url_regexp_test.go

@@ -1,9 +1,9 @@
 package main
 
 import (
-	"testing"
 	"net/http"
 	"net/url"
+	"testing"
 )
 
 func TestHTTPUrlRegexp(t *testing.T) {
@@ -15,12 +15,12 @@ func TestHTTPUrlRegexp(t *testing.T) {
 	req.Host = "www.google.com"
 	var err error
 	req.URL, err = url.Parse("/admin/testpage1")
-	if(!filter.Good(&req) || err != nil) {
+	if !filter.Good(&req) || err != nil {
 		t.Error("Request should pass filters")
 	}
 
 	req.URL, err = url.Parse("/user/testpage2")
-	if(filter.Good(&req) || err != nil) {
+	if filter.Good(&req) || err != nil {
 		t.Error("Request should not pass filters")
 	}
 }

test_input.go

@@ -1,5 +1,10 @@
 package main
 
+import (
+	"crypto/rand"
+	"encoding/base64"
+)
+
 type TestInput struct {
 	data chan []byte
 }
@@ -26,6 +31,16 @@ 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) EmitFile() {
+	size := 1024 * 1024 // 1 MB
+	rb := make([]byte, size)
+	rand.Read(rb)
+
+	rs := base64.URLEncoding.EncodeToString(rb)
+
+	i.data <- []byte("POST /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\n------WebKitFormBoundaryePkpFF7tjBAqx29L\nContent-Disposition: form-data; name=\"uploadedfile\"; filename=\"hello.o\"\nContent-Type: application/x-object\n" + rs + "\r\n\r\n")
+}
+
 func (i *TestInput) EmitOPTIONS() {
 	i.data <- []byte("OPTIONS / HTTP/1.1\nHost: www.w3.org\r\n\r\n")
 }