Bump version and add Makefile

Tcp pool

.gitignore

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

.travis.yml

@@ -0,0 +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 -v"

CHANGELOG.md

@@ -1,5 +1,25 @@
+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
+* 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)
+* Added License file
+
+v0.3.3 - 22 Jun 2013
+* Using TCP instead of UDP for communication between Listener and Replay
+* Significantly improved performance
+* Fixed bugs causing locking and message dropping (concurrency issues)
+* Rewrote concurrency model to use more channels
+
 v0.3 - 10 Jun 2013
 * Use RAW_SOCKETS instead of tcpdump
 * Own TCP stack
 * All HTTP request types support
-* Simplified request parsing
+* Simplified request parsing

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)

LICENSE.txt

@@ -0,0 +1,191 @@
+Apache License
+Version 2.0, January 2004
+http://www.apache.org/licenses/
+
+TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+1. Definitions.
+
+"License" shall mean the terms and conditions for use, reproduction, and
+distribution as defined by Sections 1 through 9 of this document.
+
+"Licensor" shall mean the copyright owner or entity authorized by the copyright
+owner that is granting the License.
+
+"Legal Entity" shall mean the union of the acting entity and all other entities
+that control, are controlled by, or are under common control with that entity.
+For the purposes of this definition, "control" means (i) the power, direct or
+indirect, to cause the direction or management of such entity, whether by
+contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the
+outstanding shares, or (iii) beneficial ownership of such entity.
+
+"You" (or "Your") shall mean an individual or Legal Entity exercising
+permissions granted by this License.
+
+"Source" form shall mean the preferred form for making modifications, including
+but not limited to software source code, documentation source, and configuration
+files.
+
+"Object" form shall mean any form resulting from mechanical transformation or
+translation of a Source form, including but not limited to compiled object code,
+generated documentation, and conversions to other media types.
+
+"Work" shall mean the work of authorship, whether in Source or Object form, made
+available under the License, as indicated by a copyright notice that is included
+in or attached to the work (an example is provided in the Appendix below).
+
+"Derivative Works" shall mean any work, whether in Source or Object form, that
+is based on (or derived from) the Work and for which the editorial revisions,
+annotations, elaborations, or other modifications represent, as a whole, an
+original work of authorship. For the purposes of this License, Derivative Works
+shall not include works that remain separable from, or merely link (or bind by
+name) to the interfaces of, the Work and Derivative Works thereof.
+
+"Contribution" shall mean any work of authorship, including the original version
+of the Work and any modifications or additions to that Work or Derivative Works
+thereof, that is intentionally submitted to Licensor for inclusion in the Work
+by the copyright owner or by an individual or Legal Entity authorized to submit
+on behalf of the copyright owner. For the purposes of this definition,
+"submitted" means any form of electronic, verbal, or written communication sent
+to the Licensor or its representatives, including but not limited to
+communication on electronic mailing lists, source code control systems, and
+issue tracking systems that are managed by, or on behalf of, the Licensor for
+the purpose of discussing and improving the Work, but excluding communication
+that is conspicuously marked or otherwise designated in writing by the copyright
+owner as "Not a Contribution."
+
+"Contributor" shall mean Licensor and any individual or Legal Entity on behalf
+of whom a Contribution has been received by Licensor and subsequently
+incorporated within the Work.
+
+2. Grant of Copyright License.
+
+Subject to the terms and conditions of this License, each Contributor hereby
+grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free,
+irrevocable copyright license to reproduce, prepare Derivative Works of,
+publicly display, publicly perform, sublicense, and distribute the Work and such
+Derivative Works in Source or Object form.
+
+3. Grant of Patent License.
+
+Subject to the terms and conditions of this License, each Contributor hereby
+grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free,
+irrevocable (except as stated in this section) patent license to make, have
+made, use, offer to sell, sell, import, and otherwise transfer the Work, where
+such license applies only to those patent claims licensable by such Contributor
+that are necessarily infringed by their Contribution(s) alone or by combination
+of their Contribution(s) with the Work to which such Contribution(s) was
+submitted. If You institute patent litigation against any entity (including a
+cross-claim or counterclaim in a lawsuit) alleging that the Work or a
+Contribution incorporated within the Work constitutes direct or contributory
+patent infringement, then any patent licenses granted to You under this License
+for that Work shall terminate as of the date such litigation is filed.
+
+4. Redistribution.
+
+You may reproduce and distribute copies of the Work or Derivative Works thereof
+in any medium, with or without modifications, and in Source or Object form,
+provided that You meet the following conditions:
+
+You must give any other recipients of the Work or Derivative Works a copy of
+this License; and
+You must cause any modified files to carry prominent notices stating that You
+changed the files; and
+You must retain, in the Source form of any Derivative Works that You distribute,
+all copyright, patent, trademark, and attribution notices from the Source form
+of the Work, excluding those notices that do not pertain to any part of the
+Derivative Works; and
+If the Work includes a "NOTICE" text file as part of its distribution, then any
+Derivative Works that You distribute must include a readable copy of the
+attribution notices contained within such NOTICE file, excluding those notices
+that do not pertain to any part of the Derivative Works, in at least one of the
+following places: within a NOTICE text file distributed as part of the
+Derivative Works; within the Source form or documentation, if provided along
+with the Derivative Works; or, within a display generated by the Derivative
+Works, if and wherever such third-party notices normally appear. The contents of
+the NOTICE file are for informational purposes only and do not modify the
+License. You may add Your own attribution notices within Derivative Works that
+You distribute, alongside or as an addendum to the NOTICE text from the Work,
+provided that such additional attribution notices cannot be construed as
+modifying the License.
+You may add Your own copyright statement to Your modifications and may provide
+additional or different license terms and conditions for use, reproduction, or
+distribution of Your modifications, or for any such Derivative Works as a whole,
+provided Your use, reproduction, and distribution of the Work otherwise complies
+with the conditions stated in this License.
+
+5. Submission of Contributions.
+
+Unless You explicitly state otherwise, any Contribution intentionally submitted
+for inclusion in the Work by You to the Licensor shall be under the terms and
+conditions of this License, without any additional terms or conditions.
+Notwithstanding the above, nothing herein shall supersede or modify the terms of
+any separate license agreement you may have executed with Licensor regarding
+such Contributions.
+
+6. Trademarks.
+
+This License does not grant permission to use the trade names, trademarks,
+service marks, or product names of the Licensor, except as required for
+reasonable and customary use in describing the origin of the Work and
+reproducing the content of the NOTICE file.
+
+7. Disclaimer of Warranty.
+
+Unless required by applicable law or agreed to in writing, Licensor provides the
+Work (and each Contributor provides its Contributions) on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied,
+including, without limitation, any warranties or conditions of TITLE,
+NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are
+solely responsible for determining the appropriateness of using or
+redistributing the Work and assume any risks associated with Your exercise of
+permissions under this License.
+
+8. Limitation of Liability.
+
+In no event and under no legal theory, whether in tort (including negligence),
+contract, or otherwise, unless required by applicable law (such as deliberate
+and grossly negligent acts) or agreed to in writing, shall any Contributor be
+liable to You for damages, including any direct, indirect, special, incidental,
+or consequential damages of any character arising as a result of this License or
+out of the use or inability to use the Work (including but not limited to
+damages for loss of goodwill, work stoppage, computer failure or malfunction, or
+any and all other commercial damages or losses), even if such Contributor has
+been advised of the possibility of such damages.
+
+9. Accepting Warranty or Additional Liability.
+
+While redistributing the Work or Derivative Works thereof, You may choose to
+offer, and charge a fee for, acceptance of support, warranty, indemnity, or
+other liability obligations and/or rights consistent with this License. However,
+in accepting such obligations, You may act only on Your own behalf and on Your
+sole responsibility, not on behalf of any other Contributor, and only if You
+agree to indemnify, defend, and hold each Contributor harmless for any liability
+incurred by, or claims asserted against, such Contributor by reason of your
+accepting any such warranty or additional liability.
+
+END OF TERMS AND CONDITIONS
+
+APPENDIX: How to apply the Apache License to your work
+
+To apply the Apache License to your work, attach the following boilerplate
+notice, with the fields enclosed by brackets "[]" replaced with your own
+identifying information. (Don't include the brackets!) The text should be
+enclosed in the appropriate comment syntax for the file format. We also
+recommend that a file or class name and description of purpose be included on
+the same "printed page" as the copyright notice for easier identification within
+third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.

Makefile

@@ -0,0 +1,16 @@
+all: build-macosx 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
+
+build-macosx:
+    GOOS=darwin GOARCH=amd64 CGO_ENABLED=0 go build
+    tar -czf gor_macosx.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,88 +1,228 @@
+[![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 via UDP. 
-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/zZCFPCY.png)
+![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
+# 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
-The replay server 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 --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 coma.
+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
+
+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:
 ```
-$ 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.
+  -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.
 ```
 
-```
-$ 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
-```
-
-## Pre-build binaries
-
-[Download binaries (linux 32/64, darwin)](https://drive.google.com/folderview?id=0B46uay48NwcfWFowc1E4a1BISVU&usp=sharing)
+## 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 ./bin/gor.go` to get binary, or `go run ./bin/gor.go` to build and run (useful for development)
 
 ## FAQ
 
-### Why does the `gor listener` requires sudo or root access?
+### 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.
 
-### 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
 
-## TODO
+More about ulimit: http://blog.thecodingmachine.com/content/solving-too-many-open-files-exception-red5-or-any-other-application
 
-Use buffering for request throttling instead of simple rate limiting. 
+## Tuning
 
-Better stats
+To achieve the top most performance you should tune the source server system limits:
 
-Optimize for load testing cases
+    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
+
+* 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

bin/gor.go

@@ -0,0 +1,80 @@
+// 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.
+package main
+
+import (
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"runtime/debug"
+	"runtime/pprof"
+	"time"
+
+	"github.com/buger/gor"
+)
+
+var (
+	mode       string
+	cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
+	memprofile = flag.String("memprofile", "", "write memory profile to this file")
+)
+
+func main() {
+	// Don't exit on panic
+	defer func() {
+		if r := recover(); r != nil {
+			if _, ok := r.(error); !ok {
+				fmt.Printf("PANIC: pkg: %v %s \n", r, debug.Stack())
+			}
+		}
+	}()
+
+	fmt.Println("Version:", gor.VERSION)
+
+	flag.Parse()
+	gor.InitPlugins()
+
+	if len(gor.Plugins.Inputs) == 0 || len(gor.Plugins.Outputs) == 0 {
+		log.Fatal("Required at least 1 input and 1 output")
+	}
+
+	if *memprofile != "" {
+		profileMEM(*memprofile)
+	}
+
+	if *cpuprofile != "" {
+		profileCPU(*cpuprofile)
+	}
+
+	gor.Start(nil)
+}
+
+func profileCPU(cpuprofile string) {
+	if cpuprofile != "" {
+		f, err := os.Create(cpuprofile)
+		if err != nil {
+			log.Fatal(err)
+		}
+		pprof.StartCPUProfile(f)
+
+		time.AfterFunc(60*time.Second, func() {
+			pprof.StopCPUProfile()
+			f.Close()
+			log.Println("Stop profiling after 60 seconds")
+		})
+	}
+}
+
+func profileMEM(memprofile string) {
+	if memprofile != "" {
+		f, err := os.Create(memprofile)
+		if err != nil {
+			log.Fatal(err)
+		}
+		time.AfterFunc(60*time.Second, func() {
+			pprof.WriteHeapProfile(f)
+			f.Close()
+		})
+	}
+}

elasticsearch/elasticsearch.go

@@ -0,0 +1,153 @@
+package gor
+
+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 gor
+
+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 {
+			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 gor
+
+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,46 +0,0 @@
-// 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"
-	"os"
-)
-
-const (
-	VERSION = "0.3"
-)
-
-func main() {
-	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()
-
-	switch mode {
-	case "listen":
-		listener.Run()
-	case "replay":
-		replay.Run()
-	}
-
-}

input_dummy.go

@@ -0,0 +1,40 @@
+package gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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)
+}

limiter.go

@@ -0,0 +1,45 @@
+package gor
+
+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 gor
+
+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,80 +0,0 @@
-// Listener capture TCP traffic using RAW SOCKETS.
-// Note: it requires sudo or root access.
-//
-// Rigt now it suport only HTTP, and only GET requests.
-package listener
-
-import (
-	"bufio"
-	"bytes"
-	"fmt"
-	"log"
-	"net"
-	"net/http"
-	"os"
-)
-
-type HttpRequest struct {
-	Tag     string            // Not used yet
-	Method  string            // Right now only 'GET'
-	Url     string            // Request URL
-	Headers map[string]string // Request Headers
-}
-
-// Enable debug logging only if "--verbose" flag passed
-func Debug(v ...interface{}) {
-	if Settings.verbose {
-		log.Println(v...)
-	}
-}
-
-func greeting() {
-
-}
-
-// 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.port, "port")
-	fmt.Println("Forwarding requests to replay server:", Settings.ReplayServer())
-
-	// Connection to reaplay server
-	serverAddr, err := net.ResolveUDPAddr("udp4", Settings.ReplayServer())
-	conn, err := net.DialUDP("udp", nil, serverAddr)
-
-	if err != nil {
-		log.Fatal("Connection error", err)
-	}
-
-	// Sniffing traffic from given port
-	listener := RAWTCPListen("0.0.0.0", Settings.port)
-
-	for {
-		message := listener.Receive()
-
-		go func(m *TCPMessage) {
-			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:", string(m.Bytes()))
-				} else {
-					request.ParseMultipartForm(32 << 20)
-					Debug("Forwarding request:", request)
-				}
-			}
-
-			conn.Write(m.Bytes())
-		}(message)
-	}
-
-	conn.Close()
-}

listener/raw_tcp_listener.go

@@ -1,105 +0,0 @@
-package listener
-
-import (
-	"log"
-	"net"
-	"time"
-)
-
-type RAWTCPListener struct {
-	messages map[uint32]*TCPMessage // buffer of TCPMessages waiting to be send
-
-	c_packets  chan *TCPPacket
-	c_messages chan *TCPMessage
-
-	addr string
-	port int
-}
-
-func RAWTCPListen(addr string, port int) (listener *RAWTCPListener) {
-	listener = &RAWTCPListener{}
-	listener.messages = make(map[uint32]*TCPMessage)
-
-	listener.c_packets = make(chan *TCPPacket)
-	listener.c_messages = make(chan *TCPMessage)
-
-	listener.addr = addr
-	listener.port = port
-
-	go listener.listen()
-	go listener.readTCPPackets()
-
-	return
-}
-
-func (t *RAWTCPListener) listen() {
-
-	for {
-		var messages chan *TCPMessage
-		var message *TCPMessage
-
-		for _, msg := range t.messages {
-			if msg.Complete() {
-				messages = t.c_messages
-				message = msg
-			}
-		}
-
-		select {
-		case messages <- message:
-			delete(t.messages, message.ask)
-		case packet := <-t.c_packets:
-			t.processTCPPacket(packet)
-
-		// Ensure that this will be run at least each 200 ms, to ensure that all messages will be send
-		// Without it last message may not be send (it will be send only on next TCP packets)
-		case <-time.After(200 * time.Millisecond):
-		}
-	}
-}
-
-func (t *RAWTCPListener) readTCPPackets() {
-	conn, e := net.ListenPacket("ip4:tcp", t.addr)
-	defer conn.Close()
-
-	if e != nil {
-		log.Fatal(e)
-	}
-
-	for {
-		buf := make([]byte, 65565)
-		n, _, err := conn.ReadFrom(buf)
-
-		if err != nil {
-			Debug("Error:", err)
-		}
-
-		if n > 0 {
-			packet := NewTCPPacket(buf[:n])
-
-			if int(packet.dest_port) == t.port {
-				t.c_packets <- packet
-			}
-		}
-	}
-}
-
-//
-func (t *RAWTCPListener) processTCPPacket(packet *TCPPacket) {
-	// We interested only in packets that contain some data
-	if !(packet.f_ask && packet.f_psh) {
-		return
-	}
-
-	ask := packet.asknowledgement
-
-	if _, ok := t.messages[ask]; !ok {
-		t.messages[ask] = NewTCPMessage(ask)
-	}
-
-	t.messages[ask].AddPacket(packet)
-}
-
-func (t *RAWTCPListener) Receive() *TCPMessage {
-	return <-t.c_messages
-}

listener/settings.go

@@ -1,47 +0,0 @@
-package listener
-
-import (
-	"flag"
-	"os"
-	"strings"
-)
-
-const (
-	defaultPort             = 80
-	defaultNetworkInterface = "any"
-
-	defaultReplayAddress = "localhost:28020"
-)
-
-type ListenerSettings struct {
-	networkInterface string
-	port             int
-
-	replayAddress string
-
-	verbose bool
-}
-
-var Settings ListenerSettings = ListenerSettings{}
-
-func (s *ListenerSettings) ReplayServer() string {
-	if !strings.Contains(s.replayAddress, ":") {
-		return s.replayAddress + ":28020"
-	}
-
-	return s.replayAddress
-}
-
-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.networkInterface, "i", defaultNetworkInterface, "By default it try to listen on all network interfaces.To get list of interfaces run `ifconfig`")
-
-	flag.StringVar(&Settings.replayAddress, "r", defaultReplayAddress, "Address of replay server.")
-
-	flag.BoolVar(&Settings.verbose, "verbose", false, "Log requests")
-}

listener/tcp_message.go

@@ -1,64 +0,0 @@
-package listener
-
-import (
-	"sort"
-	"time"
-)
-
-// 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: asknowledgement - 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 OR if we received packet with "fin" flag
-type TCPMessage struct {
-	ask     uint32             // Message ID
-	packets map[int]*TCPPacket // map[packet.sequence]*TCPPacket
-	updated int64              // time of last packet
-}
-
-func NewTCPMessage(ask uint32) (msg *TCPMessage) {
-	msg = &TCPMessage{}
-	msg.packets = make(map[int]*TCPPacket)
-	msg.updated = time.Now().UnixNano()
-	msg.ask = ask
-	return
-}
-
-// Sort packets in right orders and return message content
-func (t *TCPMessage) Bytes() (output []byte) {
-	mk := make([]int, len(t.packets))
-
-	i := 0
-	for k, _ := range t.packets {
-		mk[i] = k
-		i++
-	}
-
-	sort.Ints(mk)
-
-	for _, k := range mk {
-		output = append(output, t.packets[k].data...)
-	}
-
-	return
-}
-
-// Add packet to the message
-func (t *TCPMessage) AddPacket(packet *TCPPacket) {
-	seq := int(packet.sequence)
-
-	if _, ok := t.packets[seq]; !ok {
-		t.packets[seq] = packet
-	} else {
-		Debug("Received packet with same sequence")
-	}
-
-	t.updated = time.Now().UnixNano()
-}
-
-// TCP message is complete if we not received any packets for 200ms since last packet
-func (t *TCPMessage) Complete() bool {
-	ns := time.Now().UnixNano()
-	return (ns - t.updated) > int64(200*time.Millisecond)
-}

listener/tcp_packet.go

@@ -1,106 +0,0 @@
-package listener
-
-import (
-	"bytes"
-	"encoding/binary"
-	"strconv"
-	"strings"
-)
-
-// Simple TCP packet parser
-//
-// Packet structure: http://en.wikipedia.org/wiki/Transmission_Control_Protocol
-type TCPPacket struct {
-	packet []byte
-
-	buf *bytes.Buffer
-
-	source_port uint16
-	dest_port   uint16
-
-	sequence        uint32
-	asknowledgement uint32
-
-	doff_reserved uint16
-	tcph_length   uint16
-
-	f_ns  bool
-	f_crw bool
-	f_ece bool
-	f_urg bool
-	f_ask bool
-	f_psh bool
-	f_rst bool
-	f_syn bool
-	f_fin bool
-
-	window_size uint16
-	checksum    uint16
-
-	data []byte
-}
-
-func NewTCPPacket(b []byte) (p *TCPPacket) {
-	buf := bytes.NewBuffer(b)
-	p = &TCPPacket{packet: b, buf: buf}
-	p.Parse()
-
-	return p
-}
-
-// Helper for binary.Read
-func (t *TCPPacket) read(data interface{}) {
-	binary.Read(t.buf, binary.BigEndian, data)
-}
-
-// Inspired by: https://gist.github.com/clicube/4978853
-func (t *TCPPacket) Parse() {
-	t.read(&t.source_port)
-	t.read(&t.dest_port)
-
-	t.read(&t.sequence)
-	t.read(&t.asknowledgement)
-	t.read(&t.doff_reserved)
-
-	t.tcph_length = t.doff_reserved >> 12 * 4
-
-	t.f_ns = (t.doff_reserved & 256) != 0
-	t.f_crw = (t.doff_reserved & 128) != 0
-	t.f_ece = (t.doff_reserved & 64) != 0
-	t.f_urg = (t.doff_reserved & 32) != 0
-	t.f_ask = (t.doff_reserved & 16) != 0
-	t.f_psh = (t.doff_reserved & 8) != 0
-	t.f_rst = (t.doff_reserved & 4) != 0
-	t.f_syn = (t.doff_reserved & 2) != 0
-	t.f_fin = (t.doff_reserved & 1) != 0
-
-	t.read(&t.window_size)
-	t.read(&t.checksum)
-
-	t.data = t.packet[t.tcph_length:]
-}
-
-func (t *TCPPacket) String() string {
-	return strings.Join([]string{
-		"Source port: " + strconv.Itoa(int(t.source_port)),
-		"Dest port:" + strconv.Itoa(int(t.dest_port)),
-		"Sequence:" + strconv.Itoa(int(t.sequence)),
-		"Acknowledgement:" + strconv.Itoa(int(t.asknowledgement)),
-		"Header len:" + strconv.Itoa(int(t.tcph_length)),
-
-		"Flag ns:" + strconv.FormatBool(t.f_ns),
-		"Flag crw:" + strconv.FormatBool(t.f_crw),
-		"Flag ece:" + strconv.FormatBool(t.f_ece),
-		"Flag urg:" + strconv.FormatBool(t.f_urg),
-		"Flag ask:" + strconv.FormatBool(t.f_ask),
-		"Flag psh:" + strconv.FormatBool(t.f_psh),
-		"Flag rst:" + strconv.FormatBool(t.f_rst),
-		"Flag syn:" + strconv.FormatBool(t.f_syn),
-		"Flag fin:" + strconv.FormatBool(t.f_fin),
-
-		"Window size:" + strconv.Itoa(int(t.window_size)),
-		"Checksum:" + strconv.Itoa(int(t.checksum)),
-
-		"Data:" + string(t.data),
-	}, "\n")
-}

output_dummy.go

@@ -0,0 +1,24 @@
+package gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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,140 @@
+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[uint32]*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[uint32]*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.Ack)
+
+		// 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, _, err := conn.ReadFrom(buf)
+
+		if err != nil {
+			log.Println("Error:", err)
+			continue
+		}
+
+		if n > 0 {
+			t.parsePacket(buf[:n])
+		}
+	}
+}
+
+func (t *Listener) parsePacket(buf []byte) {
+	if t.isIncomingDataPacket(buf) {
+		new_buf := make([]byte, len(buf))
+		copy(new_buf, buf)
+
+		t.c_packets <- ParseTCPPacket(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
+
+	message, ok := t.messages[packet.Ack]
+
+	if !ok {
+		// 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[packet.Ack] = 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

@@ -0,0 +1,102 @@
+package raw_socket
+
+import (
+	"log"
+	"sort"
+	"time"
+)
+
+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: 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 didn't receive any packets for 200ms
+type TCPMessage struct {
+	Ack     uint32 // Message ID
+	packets []*TCPPacket
+
+	timer *time.Timer // Used for expire check
+
+	c_packets chan *TCPPacket
+
+	c_del_message chan *TCPMessage
+}
+
+// NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted
+func NewTCPMessage(Ack uint32, c_del chan *TCPMessage) (msg *TCPMessage) {
+	msg = &TCPMessage{Ack: Ack}
+
+	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)
+
+	go msg.listen()
+
+	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
+			}
+		}
+	}
+}
+
+// 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
+}
+
+// Bytes sorts packets in right orders and return message content
+func (t *TCPMessage) Bytes() (output []byte) {
+	mk := make([]int, len(t.packets))
+
+	i := 0
+	for k, _ := range t.packets {
+		mk[i] = k
+		i++
+	}
+
+	sort.Ints(mk)
+
+	for _, k := range mk {
+		output = append(output, t.packets[k].Data...)
+	}
+
+	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
+
+	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)
+	}
+
+	// Reset message timeout timer
+	t.timer.Reset(MSG_EXPIRE)
+}

raw_socket_listener/tcp_packet.go

@@ -0,0 +1,90 @@
+package raw_socket
+
+import (
+	"encoding/binary"
+	"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
+)
+
+// 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
+
+	Data []byte
+}
+
+func ParseTCPPacket(b []byte) (p *TCPPacket) {
+	p = &TCPPacket{Data: b}
+	p.ParseBasic()
+
+	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])
+}
+
+// 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)),
+		"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),
+
+		"Window size:" + strconv.Itoa(int(t.Window)),
+		"Checksum:" + strconv.Itoa(int(t.Checksum)),
+
+		"Data:" + string(t.Data),
+	}, "\n")
+}

replay/replay.go

@@ -1,98 +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"
-	"log"
-	"net"
-	"net/http"
-)
-
-const bufSize = 1024 * 10
-
-// 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() {
-	var buf [bufSize]byte
-
-	addr, err := net.ResolveUDPAddr("udp", Settings.Address())
-	if err != nil {
-		log.Fatal("Can't start:", err)
-	}
-
-	conn, err := net.ListenUDP("udp", addr)
-	log.Println("Starting replay server at:", Settings.Address())
-
-	if err != nil {
-		log.Fatal("Can't start:", err)
-	}
-
-	defer conn.Close()
-
-	for _, host := range Settings.ForwardedHosts() {
-		log.Println("Forwarding requests to:", host.Url, "limit:", host.Limit)
-	}
-
-	requestFactory := NewRequestFactory()
-
-	for {
-		n, _, err := conn.ReadFromUDP(buf[0:])
-
-		if err != nil {
-			continue
-		}
-
-		if n > 0 {
-			if n > bufSize {
-				Debug("Too large udp packet", bufSize)
-			}
-
-			if request, err := ParseRequest(buf[0:n]); err != nil {
-				Debug("Error while parsing request", err, buf[0:n])
-			} else {
-				requestFactory.Add(request)
-			}
-		}
-	}
-
-}

replay/request_factory.go

@@ -1,92 +0,0 @@
-package replay
-
-import (
-	"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 {
-	responses chan *HttpResponse
-	requests  chan *http.Request
-}
-
-// RequestFactory contstuctor
-// One created, it starts listening for incoming requests: requests channel
-func NewRequestFactory() (factory *RequestFactory) {
-	factory = &RequestFactory{}
-	factory.responses = make(chan *HttpResponse)
-	factory.requests = make(chan *http.Request)
-
-	go factory.handleRequests()
-
-	return
-}
-
-// Forward http request to given host
-func (f *RequestFactory) sendRequest(host *ForwardHost, request *http.Request) {
-	client := &http.Client{}
-
-	URL := host.Url + request.URL.String()
-
-	request.RequestURI = ""
-	request.URL, _ = url.ParseRequestURI(URL)
-
-	if Settings.verbose {
-		Debug("Sending request:", request)
-	}
-
-	resp, err := client.Do(request)
-
-	if err == nil {
-		defer resp.Body.Close()
-	} else {
-		Debug("Request error:", err)
-	}
-
-	f.responses <- &HttpResponse{host, request, resp, err}
-}
-
-// Handle incoming requests, and they responses
-func (f *RequestFactory) handleRequests() {
-	hosts := Settings.ForwardedHosts()
-
-	for {
-		select {
-		case req := <-f.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.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.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,80 +0,0 @@
-package replay
-
-import (
-	"strconv"
-	"strings"
-	"os"
-	"flag"
-)
-
-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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+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 gor
+
+type writeCallback func(data []byte)
+
+type TestOutput struct {
+	cb writeCallback
+}
+
+func NewTestOutput(cb writeCallback) (i *TestOutput) {
+	i = new(TestOutput)
+	i.cb = cb
+
+	return
+}
+
+func (i *TestOutput) Write(data []byte) (int, error) {
+	i.cb(data)
+
+	return len(data), nil
+}
+
+func (i *TestOutput) String() string {
+	return "Test Input"
+}