use go 1.6.1 on travis ci

Signed-off-by: Salimane Adjao Moustapha <me@salimane.com>

.gitignore

@@ -4,3 +4,11 @@
 *.out
 
 *.bin
+
+*.gz
+
+*.class
+
+*.test
+.idea
+gor

.travis.yml

@@ -1,3 +1,7 @@
 language: go
-go: 1.4.2
-script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.4.2)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test ./... -v -timeout 15s"
+go:
+  - 1.6.1
+
+install: go get ./...
+
+script: go test ./... -v -timeout 60s -race

Dockerfile

@@ -2,9 +2,25 @@ FROM google/golang:1.4
 
 RUN cd /goroot/src/ && GOOS=linux GOARCH=386 ./make.bash --no-clean
 
-WORKDIR /gopath/src/github.com/buger/gor/
+RUN apt-get update && apt-get install ruby vim-common -y
 
+# Install Java for middleware testing
+RUN echo "deb http://ppa.launchpad.net/webupd8team/java/ubuntu trusty main" | tee /etc/apt/sources.list.d/webupd8team-java.list
+RUN echo "deb-src http://ppa.launchpad.net/webupd8team/java/ubuntu trusty main" | tee -a /etc/apt/sources.list.d/webupd8team-java.list
+RUN apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv-keys EEA14886
+RUN apt-get update -y
+RUN echo oracle-java7-installer shared/accepted-oracle-license-v1-1 select true | /usr/bin/debconf-set-selections
+RUN apt-get install oracle-java8-installer -y
+
+RUN wget http://apache-mirror.rbc.ru/pub/apache//commons/io/binaries/commons-io-2.4-bin.tar.gz -P /tmp
+RUN tar xzf /tmp/commons-io-2.4-bin.tar.gz -C /tmp
+
+WORKDIR /gopath/src/github.com/buger/gor/
 ADD . /gopath/src/github.com/buger/gor/
 
+RUN javac -cp /tmp/commons-io-2.4/commons-io-2.4.jar ./examples/middleware/echo.java
+
+RUN apt-get install graphviz -y
+
 RUN go get -u github.com/golang/lint/golint
 RUN go get

Makefile

@@ -1,18 +1,22 @@
-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 test_input.go elasticsearch.go http_modifier.go http_modifier_settings.go http_client.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 test_input.go elasticsearch.go http_modifier.go http_modifier_settings.go http_client.go middleware.go protocol.go
 
 SOURCE_PATH = /gopath/src/github.com/buger/gor/
 
 release: release-x86 release-x64
 
 release-x64:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 --env CGO_ENABLED=0 -i gor go build && tar -czf gor_x64.tar.gz gor && rm gor
+	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 --env CGO_ENABLED=0 -i gor go build -ldflags "-X main.VERSION $(VERSION)"&& tar -czf gor_$(VERSION)_x64.tar.gz gor && rm gor
 
 release-x86:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 --env CGO_ENABLED=0 -i gor go build && tar -czf gor_x86.tar.gz gor && rm gor
+	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 --env CGO_ENABLED=0 -i gor go build -ldflags "-X main.VERSION $(VERSION)" && tar -czf gor_$(VERSION)_x86.tar.gz gor && rm gor
 
 dbuild:
 	docker build -t gor .
 
+
+profile:
+	go build && ./gor --output-http="http://localhost:9000" --input-dummy 0 --input-raw :9000 --input-http :9000 --memprofile=./mem.out --cpuprofile=./cpu.out --stats --output-http-stats --output-http-timeout 100ms
+
 dlint:
 	docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor golint $(PKG)
 
@@ -20,7 +24,7 @@ drace:
 	docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test ./... $(ARGS) -v -race -timeout 15s
 
 dtest:
-	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go test ./... $(ARGS) -v -timeout 10s
+	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go test ./... -timeout 60s $(ARGS) -v
 
 dcover:
 	docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test $(ARGS) -race -v -timeout 15s -coverprofile=coverage.out
@@ -37,13 +41,16 @@ dbench:
 
 # Used mainly for debugging, because docker container do not have access to parent machine ports
 drun:
-	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000"  --verbose
+	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw :9000 --input-http :9000 --verbose --debug --middleware "./examples/middleware/echo.sh"
+
+drun-2:
+	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-file ./fixtures/requests.gor --output-dummy=0
 
 drecord:
-	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-file=requests.bin --verbose
+	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-file=requests.gor --verbose --debug
 
 dreplay:
 	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-file=requests.bin --output-tcp=:9000 --verbose -h
 
 dbash:
-	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor /bin/bash
+	docker run -v `pwd`:$(SOURCE_PATH) -t -i gor /bin/bash

README.md

@@ -2,8 +2,7 @@
 
 ## 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.
+Gor is an open-source tool for capturing and replaying live HTTP traffic into a test environment in order to continuously test your system with real data. It can be used to increase confidence in code deployments, configuration changes and infrastructure changes.
 
 
 Now you can test your code on real user sessions in an automated and repeatable fashion.
@@ -61,11 +60,11 @@ gor --input-tcp :28020 --output-http "http://staging.com"  --output-http "http:/
 ```
 
 ### HTTP output workers
-By default Gor creates dynamic pull of workers: it starts with 10 and 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.  
+By default Gor creates a dynamic pool of workers: it starts with 10 and creates 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.
 You may specify fixed number of workers using  `--output-http-workers=20` option.
 
 ### Follow redirects
-By default Gor will ignore all redirects since they are handled by clients using your app, but in scenarios when your replayed environment introduce new redirects, you can enable them like this: 
+By default Gor will ignore all redirects since they are handled by clients using your app, but in scenarios where your replayed environment introduces new redirects, you can enable them like this: 
 ```
 gor --input-tcp replay.local:28020 --output-http http://staging.com --output-http-redirects 2
 ```
@@ -78,10 +77,10 @@ gor --input-tcp replay.local:28020 --output-http http://staging.com --output-htt
 ```
 
 ### Rate limiting
-Rate limiting can be useful if you want forward only part of production traffic and not overload your staging environment. There is 2 strategies: dropping random requests or dropping fraction of requests based on Header or URL param value. 
+Rate limiting can be useful if you only want to forward parts of production traffic and not overload your staging environment. There are 2 strategies: dropping random requests or dropping fractions of requests based on Header or URL param value. 
 
 #### Dropping random requests
-Every input and output support random rate limiting.
+Every input and output supports random rate limiting.
 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.
@@ -149,7 +148,7 @@ gor --input-raw :80 --output-http "http://staging.server" \
 ```
 
 ### Rewriting original request
-Gor supports built-in basic rewriting support, for complex logic see https://github.com/buger/gor/pull/162
+Gor supports some basic request rewriting support. For complex logic you can use middleware, see below.
 
 #### Rewrite URL based on a mapping
 ```
@@ -177,6 +176,74 @@ Host header gets special treatment. By default Host get set to the value specifi
 
 If you app accepts traffic from multiple domain, and you want to keep original headers, there is specific `--http-original-host` with tells Gor do not touch Host header at all.
 
+### Middleware
+Middleware is a program that accepts request and response payload at STDIN and emits modified requests at STDOUT. You can implement any custom logic like stripping private data, advanced rewriting, support for oAuth and etc.
+
+```
+                   Original request      +--------------+
++-------------+----------STDIN---------->+              |
+|  Gor input  |                          |  Middleware  |
++-------------+----------STDIN---------->+              |
+                   Original response     +------+---+---+
+                                                |   ^
++-------------+    Modified request             v   |
+| Gor output  +<---------STDOUT-----------------+   |
++-----+-------+                                     |
+      |                                             |
+      |            Replayed response                |
+      +------------------STDIN----------------->----+
+```
+
+Middleware can be written in any language, see `examples/middleware` folder for examples.
+Middleware program should accept the fact that all communication with Gor is asynchronous, there is no guarantee that original request and response messages will come one after each other. Your app should take care of the state if logic depends on original or replayed response, see `examples/middleware/token_modifier.go` as example.
+
+Simple bash echo middleware (returns same request) will look like this:
+```bash
+while read line; do
+  echo $line
+end
+```
+
+Middleware can be enabled using `--middleware` option, by specifying path to executable file:
+```
+gor --input-raw :80 --middleware "/opt/middleware_executable" --output-http "http://staging.server"
+```
+
+#### Communication protocol
+All messages should be hex encoded, new line character specifieds the end of the message, eg. new message per line.
+
+Decoded payload consist of 2 parts: header and HTTP payload, separated by new line character.  
+
+Example request payload:
+
+```
+1 932079936fa4306fc308d67588178d17d823647c 1439818823587396305
+GET /a HTTP/1.1
+Host: 127.0.0.1
+
+```
+
+Example response payload:
+
+```
+2 8e091765ae902fef8a2b7d9dd960e9d52222bd8c 2782013
+HTTP/1.1 200 OK
+Date: Mon, 17 Aug 2015 13:40:23 GMT
+Content-Length: 0
+Content-Type: text/plain; charset=utf-8
+
+```
+
+Header contains request meta information separated by spaces. First value is payload type, possible values: `1` - request, `2` - original response, `3` - replayed response.
+Next goes request id: unique among all requests (sha1 of time and Ack), but remain same for original and replayed response, so you can create associations between request and responses. Third argument varies depending on payload type: for request - start time, for responses - round-trip time.
+
+HTTP payload is unmodified HTTP requests/responses intercepted from network. You can read more about request format [here](http://www.jmarshall.com/easy/http/), [here](https://en.wikipedia.org/wiki/Hypertext_Transfer_Protocol) and [here](http://www.w3.org/Protocols/rfc2616/rfc2616.html). You can operate with payload as you want, add headers, change path, and etc. Basically you just editing a string, just ensure that it is RCF compliant.
+
+At the end modified (or untouched) request should be emitted back to STDOUT, keeping original header, and hex-encoded. If you want to filter request, just not send it. Emitting responses back is required, even if you did not touch them.
+
+#### Advanced example
+Imagine that you have auth system that randomly generate access tokens, which used later for accessing secure content. Since there is no pre-defined token value, naive approach without middleware (or if middleware use only request payloads) will fail, because replayed server have own tokens, not synced with origin. To fix this, our middleware should take in account responses of replayed and origin server, store `originalToken -> replayedToken` aliases and rewrite all requests using this token to use replayed alias. See `examples/middleware/token_modifier.go` and `middleware_test.go#TestTokenMiddleware` as example of described scheme.
+
 ### Saving requests to file and replaying them
 You can save requests to file, and replay them later:
 ```
@@ -280,7 +347,7 @@ https://github.com/buger/gor/releases
   gor --input-raw :8080 --output-http staging.com --http-allow-method GET --http-allow-method OPTIONS
   -http-allow-url=[]: A regexp to match requests against. Filter get matched agains full url with domain. Anything else will be dropped:
    gor --input-raw :8080 --output-http staging.com --http-allow-url ^www.
-  -http-diallow-url=[]: A regexp to match requests against. Filter get matched agains full url with domain. Anything else will be dropped:
+  -http-disallow-url=[]: A regexp to match requests against. Filter get matched agains full url with domain. Anything else will be forwarded:
    gor --input-raw :8080 --output-http staging.com --http-disallow-url ^www.
   -http-header-limiter=[]: Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific header:
    gor --input-raw :8080 --output-http staging.com --http-header-imiter user-id:25%

emitter.go

@@ -1,21 +1,39 @@
 package main
 
 import (
+	"bytes"
 	"io"
 	"time"
 )
 
 // Start initialize loop for sending data from inputs to outputs
 func Start(stop chan int) {
-	for _, in := range Plugins.Inputs {
-		go CopyMulty(in, Plugins.Outputs...)
+	if Settings.middleware != "" {
+		middleware := NewMiddleware(Settings.middleware)
+
+		for _, in := range Plugins.Inputs {
+			middleware.ReadFrom(in)
+		}
+
+		// We going only to read responses, so using same ReadFrom method
+		for _, out := range Plugins.Outputs {
+			if r, ok := out.(io.Reader); ok {
+				middleware.ReadFrom(r)
+			}
+		}
+
+		go CopyMulty(middleware, Plugins.Outputs...)
+	} else {
+		for _, in := range Plugins.Inputs {
+			go CopyMulty(in, Plugins.Outputs...)
+		}
 	}
 
 	for {
 		select {
 		case <-stop:
 			return
-		case <-time.After(1 * time.Second):
+		case <-time.After(time.Second):
 		}
 	}
 }
@@ -28,20 +46,37 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
 
 	for {
 		nr, er := src.Read(buf)
+
 		if nr > 0 && len(buf) > nr {
-			payload := buf[0:nr]
+			payload := buf[:nr]
 
-			if modifier != nil {
-				payload = modifier.Rewrite(payload)
-
-				// If modifier tells to skip request
-				if len(payload) == 0 {
-					continue
-				}
+			_maxN := nr
+			if nr > 500 {
+				_maxN = 500
 			}
 
 			if Settings.debug {
-				Debug("[EMITTER] Sending payload, size:", len(payload), "First 500 bytes:", string(payload[0:500]))
+				Debug("[EMITTER] input:", string(payload[0:_maxN]), nr, "from:", src)
+			}
+
+			if modifier != nil && isRequestPayload(payload) {
+				headSize := bytes.IndexByte(payload, '\n') + 1
+				body := payload[headSize:]
+				originalBodyLen := len(body)
+				body = modifier.Rewrite(body)
+
+				// If modifier tells to skip request
+				if len(body) == 0 {
+					continue
+				}
+
+				if originalBodyLen != len(body) {
+					payload = append(payload[:headSize], body...)
+				}
+
+				if Settings.debug {
+					Debug("[EMITTER] Rewrittern input:", len(payload), "First 500 bytes:", string(payload[0:_maxN]))
+				}
 			}
 
 			if Settings.splitOutput {

examples/middleware/echo.java

@@ -0,0 +1,30 @@
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+
+import org.apache.commons.io.IOUtils;
+
+public class echo {
+    public static void main(String[] args) {
+        if(args != null){
+            for(String arg : args){
+                System.out.println(arg);
+            }
+
+        }
+
+        BufferedReader stdin = new BufferedReader(new InputStreamReader(
+                System.in));
+        String line = null;
+
+        try {
+            while ((line = stdin.readLine()) != null) {
+
+                System.out.println(line);
+
+            }
+        } catch (IOException e) {
+            IOUtils.closeQuietly(stdin);
+        }
+    }
+}

examples/middleware/echo.rb

@@ -0,0 +1,15 @@
+#!/usr/bin/env ruby
+# encoding: utf-8
+while data = STDIN.gets
+  next unless data
+  data = data.chomp
+
+  decoded = [data].pack("H*")
+  encoded = decoded.unpack("H*").first
+
+  STDOUT.puts encoded
+
+  STDERR.puts "[DEBUG][MIDDLEWARE] Original data: #{data}"
+  STDERR.puts "[DEBUG][MIDDLEWARE] Decoded request: #{decoded}"
+  STDERR.puts "[DEBUG][MIDDLEWARE] Encoded data: #{encoded}"
+end

examples/middleware/echo.sh

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+#
+# `xxd` utility included into vim-common package
+# It allow hex decoding/encoding
+
+function log {
+    # Logging to stderr, because stdout/stdin used for data transfer
+    >&2 echo "[DEBUG][ECHO] $1"
+}
+
+while read line; do
+    decoded=$(echo -e "$line" | xxd -r -p)
+
+    header=$(echo -e "$decoded" | head -n +1)
+    payload=$(echo -e "$decoded" | tail -n +2)
+
+    encoded=$(echo -e "$header\n$payload" | xxd -p | tr -d "\\n")
+
+    log ""
+    log "==================================="
+
+    case ${header:0:1} in
+    "1")
+        log "Request type: Request"
+        ;;
+    "2")
+        log "Request type: Original Response"
+        ;;
+    "3")
+        log "Request type: Replayed Response"
+        ;;
+    *)
+        log "Unknown request type $header"
+    esac
+    echo "$encoded"
+
+    log "==================================="
+
+    log "Original data: $line"
+    log "Decoded request: $decoded"
+    log "Encoded data: $encoded"
+done;

examples/middleware/token_modifier.go

@@ -0,0 +1,119 @@
+/*
+This middleware made for auth system that randomly generate access tokens, which used later for accessing secure content. Since there is no pre-defined token value, naive approach without middleware (or if middleware use only request payloads) will fail, because replayed server have own tokens, not synced with origin. To fix this, our middleware should take in account responses of replayed and origin server, store `originalToken -> replayedToken` aliases and rewrite all requests using this token to use replayed alias. See `middleware_test.go#TestTokenMiddleware` test for examples of using this middleware.
+
+How middleware works:
+
+                   Original request      +--------------+
++-------------+----------STDIN---------->+              |
+|  Gor input  |                          |  Middleware  |
++-------------+----------STDIN---------->+              |
+                   Original response     +------+---+---+
+                                                |   ^
++-------------+    Modified request             v   |
+| Gor output  +<---------STDOUT-----------------+   |
++-----+-------+                                     |
+      |                                             |
+      |            Replayed response                |
+      +------------------STDIN----------------->----+
+*/
+
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/hex"
+	"fmt"
+	"github.com/buger/gor/proto"
+	"os"
+)
+
+// requestID -> originalToken
+var originalTokens map[string][]byte
+
+// originalToken -> replayedToken
+var tokenAliases map[string][]byte
+
+func main() {
+	originalTokens = make(map[string][]byte)
+	tokenAliases = make(map[string][]byte)
+
+	scanner := bufio.NewScanner(os.Stdin)
+
+	for scanner.Scan() {
+		encoded := scanner.Bytes()
+		buf := make([]byte, len(encoded)/2)
+		hex.Decode(buf, encoded)
+
+		go process(buf)
+	}
+}
+
+func process(buf []byte) {
+	// First byte indicate payload type, possible values:
+	//  1 - Request
+	//  2 - Response
+	//  3 - ReplayedResponse
+	payloadType := buf[0]
+	headerSize := bytes.IndexByte(buf, '\n') + 1
+	header := buf[:headerSize-1]
+
+	// Header contains space separated values of: request type, request id, and request start time (or round-trip time for responses)
+	meta := bytes.Split(header, []byte(" "))
+	// For each request you should receive 3 payloads (request, response, replayed response) with same request id
+	reqID := string(meta[1])
+	payload := buf[headerSize:]
+
+	Debug("Received payload:", string(buf))
+
+	switch payloadType {
+	case '1': // Request
+		if bytes.Equal(proto.Path(payload), []byte("/token")) {
+			originalTokens[reqID] = []byte{}
+			Debug("Found token request:", reqID)
+		} else {
+			token, vs, _ := proto.PathParam(payload, []byte("token"))
+
+			if vs != -1 { // If there is GET token param
+				if alias, ok := tokenAliases[string(token)]; ok {
+					// Rewrite original token to alias
+					payload = proto.SetPathParam(payload, []byte("token"), alias)
+
+					// Copy modified payload to our buffer
+					buf = append(buf[:headerSize], payload...)
+				}
+			}
+		}
+
+		// Emitting data back
+		os.Stdout.Write(encode(buf))
+	case '2': // Original response
+		if _, ok := originalTokens[reqID]; ok {
+			// Token is inside response body
+			secureToken := proto.Body(payload)
+			originalTokens[reqID] = secureToken
+			Debug("Remember origial token:", string(secureToken))
+		}
+	case '3': // Replayed response
+		if originalToken, ok := originalTokens[reqID]; ok {
+			delete(originalTokens, reqID)
+			secureToken := proto.Body(payload)
+			tokenAliases[string(originalToken)] = secureToken
+
+			Debug("Create alias for new token token, was:", string(originalToken), "now:", string(secureToken))
+		}
+	}
+}
+
+func encode(buf []byte) []byte {
+	dst := make([]byte, len(buf)*2+1)
+	hex.Encode(dst, buf)
+	dst[len(dst)-1] = '\n'
+
+	return dst
+}
+
+func Debug(args ...interface{}) {
+	fmt.Fprint(os.Stderr, "[DEBUG][TOKEN-MOD] ")
+	fmt.Fprintln(os.Stderr, args...)
+}

gor.go

@@ -8,7 +8,7 @@ import (
 	"log"
 	"os"
 	"runtime"
-	"runtime/debug"
+	_ "runtime/debug"
 	"runtime/pprof"
 	"time"
 )
@@ -20,20 +20,18 @@ var (
 )
 
 func main() {
+	// // Don't exit on panic
+	// defer func() {
+	// 	if r := recover(); r != nil {
+	// 		fmt.Printf("PANIC: pkg: %v %s \n", r, debug.Stack())
+	// 	}
+	// }()
+
 	// If not set via env cariable
 	if len(os.Getenv("GOMAXPROCS")) == 0 {
 		runtime.GOMAXPROCS(runtime.NumCPU() * 2)
 	}
 
-	// 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:", VERSION)
 
 	flag.Parse()
@@ -62,10 +60,10 @@ func profileCPU(cpuprofile string) {
 		}
 		pprof.StartCPUProfile(f)
 
-		time.AfterFunc(60*time.Second, func() {
+		time.AfterFunc(30*time.Second, func() {
 			pprof.StopCPUProfile()
 			f.Close()
-			log.Println("Stop profiling after 60 seconds")
+			log.Println("Stop profiling after 30 seconds")
 		})
 	}
 }
@@ -76,7 +74,7 @@ func profileMEM(memprofile string) {
 		if err != nil {
 			log.Fatal(err)
 		}
-		time.AfterFunc(60*time.Second, func() {
+		time.AfterFunc(30*time.Second, func() {
 			pprof.WriteHeapProfile(f)
 			f.Close()
 		})

gor_stat.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"log"
+	"runtime"
 	"strconv"
 	"time"
 )
@@ -27,7 +28,7 @@ func NewGorStat(statName string) (s *GorStat) {
 	s.count = 0
 
 	if Settings.stats {
-		log.Println(s.statName + ":latest,mean,max,count,count/second")
+		log.Println(s.statName + ":latest,mean,max,count,count/second,gcount")
 		go s.reportStats()
 	}
 	return
@@ -54,7 +55,7 @@ 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) + "," + strconv.Itoa(runtime.NumGoroutine())
 }
 
 func (s *GorStat) reportStats() {

http_client.go

@@ -18,11 +18,12 @@ var defaultPorts = map[string]string{
 }
 
 type HTTPClientConfig struct {
-	FollowRedirects int
-	Debug           bool
-	OriginalHost    bool
-	Timeout         time.Duration
-	ResponseBufferSize  int
+	FollowRedirects    int
+	Debug              bool
+	OriginalHost       bool
+	ConnectionTimeout  time.Duration
+	Timeout            time.Duration
+	ResponseBufferSize int
 }
 
 type HTTPClient struct {
@@ -42,7 +43,9 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient {
 
 	u, _ := url.Parse(baseURL)
 	if !strings.Contains(u.Host, ":") {
-		u.Host += ":" + defaultPorts[u.Scheme]
+		if u.Scheme != "http" {
+			u.Host += ":" + defaultPorts[u.Scheme]
+		}
 	}
 
 	if config.Timeout.Nanoseconds() == 0 {
@@ -50,7 +53,7 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient {
 	}
 
 	if config.ResponseBufferSize == 0 {
-		config.ResponseBufferSize = 512*1024 // 500kb
+		config.ResponseBufferSize = 100 * 1024 // 100kb
 	}
 
 	client := new(HTTPClient)
@@ -66,7 +69,11 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient {
 func (c *HTTPClient) Connect() (err error) {
 	c.Disconnect()
 
-	c.conn, err = net.Dial("tcp", c.host)
+	if !strings.Contains(c.host, ":") {
+		c.conn, err = net.DialTimeout("tcp", c.host+":80", c.config.ConnectionTimeout)
+	} else {
+		c.conn, err = net.DialTimeout("tcp", c.host, c.config.ConnectionTimeout)
+	}
 
 	if c.scheme == "https" {
 		tlsConn := tls.Client(c.conn, &tls.Config{InsecureSkipVerify: true})
@@ -85,7 +92,7 @@ func (c *HTTPClient) Disconnect() {
 	if c.conn != nil {
 		c.conn.Close()
 		c.conn = nil
-		Debug("Disconnected: ", c.baseURL)
+		Debug("[HTTP] Disconnected: ", c.baseURL)
 	}
 }
 
@@ -118,6 +125,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) {
 		Debug("[HTTPClient] Connecting:", c.baseURL)
 		if err = c.Connect(); err != nil {
 			log.Println("[HTTPClient] Connection error:", err)
+			response = errorPayload(HTTP_CONNECTION_ERROR)
 			return
 		}
 	}
@@ -136,6 +144,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) {
 
 	if _, err = c.conn.Write(data); err != nil {
 		Debug("[HTTPClient] Write error:", err, c.baseURL)
+		response = errorPayload(HTTP_TIMEOUT)
 		return
 	}
 
@@ -154,13 +163,12 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) {
 
 	if err != nil {
 		Debug("[HTTPClient] Response read error", err, c.conn)
+		response = errorPayload(HTTP_TIMEOUT)
 		return
 	}
 
 	payload := c.respBuf[:n]
 
-	Debug("[HTTPClient] Received:", n)
-
 	if c.config.Debug {
 		Debug("[HTTPClient] Received:", string(payload))
 	}
@@ -193,3 +201,28 @@ func (c *HTTPClient) Get(path string) (response []byte, err error) {
 
 	return c.Send([]byte(payload))
 }
+
+const (
+	// https://support.cloudflare.com/hc/en-us/articles/200171936-Error-520-Web-server-is-returning-an-unknown-error
+	HTTP_UNKNOWN_ERROR = "520"
+	// https://support.cloudflare.com/hc/en-us/articles/200171916-Error-521-Web-server-is-down
+	HTTP_CONNECTION_ERROR = "521"
+	// https://support.cloudflare.com/hc/en-us/articles/200171906-Error-522-Connection-timed-out
+	HTTP_CONNECTION_TIMEOUT = "522"
+	// https://support.cloudflare.com/hc/en-us/articles/200171946-Error-523-Origin-is-unreachable
+	HTTP_UNREACHABLE = "523"
+	// https://support.cloudflare.com/hc/en-us/articles/200171926-Error-524-A-timeout-occurred
+	HTTP_TIMEOUT = "524"
+)
+
+var errorPayloadTemplate = "HTTP/1.1 202 Accepted\r\nDate: Mon, 17 Aug 2015 14:10:11 GMT\r\nContent-Length: 0\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n"
+
+func errorPayload(errorCode string) []byte {
+	payload := make([]byte, len(errorPayloadTemplate))
+	copy(payload, errorPayloadTemplate)
+
+	copy(payload[29:58], []byte(time.Now().Format(time.RFC1123)))
+	copy(payload[9:12], errorCode)
+
+	return payload
+}

http_client_test.go

@@ -3,20 +3,24 @@ package main
 import (
 	"bytes"
 	"crypto/rand"
+	"github.com/buger/gor/proto"
 	"io/ioutil"
+	_ "log"
 	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/http/httputil"
+	_ "reflect"
 	"sync"
 	"testing"
+	"time"
 	_ "time"
 )
 
 func TestHTTPClientURLPort(t *testing.T) {
 	c1 := NewHTTPClient("http://example.com", &HTTPClientConfig{})
-	if c1.baseURL != "http://example.com:80" {
-		t.Error("Sould add 80 port for http:", c1.baseURL)
+	if c1.baseURL != "http://example.com" {
+		t.Error("Sould not add 80 port for http:", c1.baseURL)
 	}
 
 	c2 := NewHTTPClient("https://example.com", &HTTPClientConfig{})
@@ -30,8 +34,8 @@ func TestHTTPClientURLPort(t *testing.T) {
 	}
 
 	c4 := NewHTTPClient("example.com", &HTTPClientConfig{})
-	if c4.baseURL != "http://example.com:80" {
-		t.Error("Sould add default protocol:", c4.baseURL)
+	if c4.baseURL != "http://example.com" {
+		t.Error("Sould not add default protocol:", c4.baseURL)
 	}
 }
 
@@ -71,6 +75,7 @@ func TestHTTPClientSend(t *testing.T) {
 
 		wg.Done()
 	}))
+	defer server.Close()
 
 	client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true})
 
@@ -99,6 +104,7 @@ func TestHTTPClientResponseBuffer(t *testing.T) {
 
 		wg.Done()
 	}))
+	defer server.Close()
 
 	client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: false, ResponseBufferSize: 1024})
 
@@ -149,6 +155,7 @@ func TestHTTPClientHTTPSSend(t *testing.T) {
 
 		wg.Done()
 	}))
+	defer server.Close()
 
 	client := NewHTTPClient(server.URL, &HTTPClientConfig{})
 
@@ -167,12 +174,16 @@ func TestHTTPClientServerInstantDisconnect(t *testing.T) {
 	GETPayload := []byte("GET / HTTP/1.1\r\n\r\n")
 
 	ln, _ := net.Listen("tcp", ":0")
+	defer ln.Close()
 
 	go func() {
 		for {
-			conn, _ := ln.Accept()
-			conn.Close()
+			conn, err := ln.Accept()
 
+			if err != nil {
+				break
+			}
+			conn.Close()
 			wg.Done()
 		}
 	}()
@@ -192,12 +203,13 @@ func TestHTTPClientServerNoKeepAlive(t *testing.T) {
 	GETPayload := []byte("GET / HTTP/1.1\r\n\r\n")
 
 	ln, _ := net.Listen("tcp", ":0")
+	defer ln.Close()
 
 	go func() {
 		for {
 			conn, err := ln.Accept()
 			if err != nil {
-				// handle error
+				break
 			}
 
 			buf := make([]byte, 4096)
@@ -237,6 +249,7 @@ func TestHTTPClientRedirect(t *testing.T) {
 
 		wg.Done()
 	}))
+	defer server.Close()
 
 	client := NewHTTPClient(server.URL, &HTTPClientConfig{FollowRedirects: 1, Debug: false})
 
@@ -268,6 +281,7 @@ func TestHTTPClientRedirectLimit(t *testing.T) {
 
 		wg.Done()
 	}))
+	defer server.Close()
 
 	client := NewHTTPClient(server.URL, &HTTPClientConfig{FollowRedirects: 2, Debug: false})
 
@@ -291,6 +305,7 @@ func TestHTTPClientHandleHTTP10(t *testing.T) {
 
 		wg.Done()
 	}))
+	defer server.Close()
 
 	client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true})
 
@@ -299,3 +314,87 @@ func TestHTTPClientHandleHTTP10(t *testing.T) {
 
 	wg.Wait()
 }
+
+func TestHTTPClientErrors(t *testing.T) {
+	req := []byte("GET http://foobar.com/path HTTP/1.0\r\n\r\n")
+
+	// Port not exists
+	client := NewHTTPClient("http://127.0.0.1:1", &HTTPClientConfig{Debug: true})
+	if resp, err := client.Send(req); err != nil {
+		if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) {
+			t.Error("Should return status 521 for connection refused, instead:", string(s))
+		}
+	} else {
+		t.Error("Should throw error")
+	}
+
+	client = NewHTTPClient("http://not.existing", &HTTPClientConfig{Debug: true})
+	if resp, err := client.Send(req); err != nil {
+		if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) {
+			t.Error("Should return status 521 for no such host, instead:", string(s))
+		}
+	} else {
+		t.Error("Should throw error")
+	}
+
+	// Non routable IP address to simulate connection timeout
+	client = NewHTTPClient("http://10.255.255.1", &HTTPClientConfig{Debug: true, ConnectionTimeout: 100 * time.Millisecond})
+
+	if resp, err := client.Send(req); err != nil {
+		if s := proto.Status(resp); !bytes.Equal(s, []byte("521")) {
+			t.Error("Should return status 521 for io/timeout:", string(s))
+		}
+	} else {
+		t.Error("Should throw error")
+	}
+
+	// Connecting but io timeout on read
+	ln, _ := net.Listen("tcp", ":0")
+	client = NewHTTPClient("http://"+ln.Addr().String(), &HTTPClientConfig{Debug: true, Timeout: 10 * time.Millisecond})
+	defer ln.Close()
+
+	if resp, err := client.Send(req); err != nil {
+		if s := proto.Status(resp); !bytes.Equal(s, []byte("524")) {
+			t.Error("Should return status 524 for io read, instead:", string(s))
+		}
+	} else {
+		t.Error("Should throw error")
+	}
+
+	// Response read error read tcp [::1]:51128: connection reset by peer &{{0xc20802a000}}
+	ln1, _ := net.Listen("tcp", ":0")
+	go func() {
+		ln1.Accept()
+	}()
+	defer ln1.Close()
+
+	client = NewHTTPClient("http://"+ln1.Addr().String(), &HTTPClientConfig{Debug: true, Timeout: 10 * time.Millisecond})
+
+	if resp, err := client.Send(req); err != nil {
+		if s := proto.Status(resp); !bytes.Equal(s, []byte("524")) {
+			t.Error("Should return status 524 for connection reset by peer, instead:", string(s))
+		}
+	} else {
+		t.Error("Should throw error")
+	}
+
+	ln2, _ := net.Listen("tcp", ":0")
+	go func() {
+		buf := make([]byte, 64*1024)
+		conn, _ := ln2.Accept()
+
+		conn.Read(buf)
+		defer conn.Close()
+	}()
+	defer ln2.Close()
+
+	client = NewHTTPClient("http://"+ln2.Addr().String(), &HTTPClientConfig{Debug: true, Timeout: 10 * time.Millisecond})
+
+	if resp, err := client.Send(req); err != nil {
+		if s := proto.Status(resp); !bytes.Equal(s, []byte("524")) {
+			t.Error("Should return status 524 for connection reset by peer, instead:", string(s))
+		}
+	} else {
+		t.Error("Should throw error")
+	}
+}

http_modifier.go

@@ -30,6 +30,10 @@ func NewHTTPModifier(config *HTTPModifierConfig) *HTTPModifier {
 }
 
 func (m *HTTPModifier) Rewrite(payload []byte) (response []byte) {
+	if !proto.IsHTTPPayload(payload) {
+		return payload
+	}
+
 	if len(m.config.methods) > 0 {
 		method := proto.Method(payload)
 

input_dummy.go

@@ -21,6 +21,7 @@ func NewDummyInput(options string) (di *DummyInput) {
 
 func (i *DummyInput) Read(data []byte) (int, error) {
 	buf := <-i.data
+
 	copy(data, buf)
 
 	return len(buf), nil
@@ -32,7 +33,12 @@ func (i *DummyInput) emit() {
 	for {
 		select {
 		case <-ticker.C:
-			i.data <- []byte("GET / HTTP/1.1\r\n\r\n")
+			uuid := uuid()
+			reqh := payloadHeader(RequestPayload, uuid, time.Now().UnixNano())
+			i.data <- append(reqh, []byte("POST /pub/WWW/å HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\nContent-Length: 7\r\n\r\na=1&b=2")...)
+
+			resh := payloadHeader(ResponsePayload, uuid, 1)
+			i.data <- append(resh, []byte("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")...)
 		}
 	}
 }

input_file.go

@@ -1,9 +1,10 @@
 package main
 
 import (
-	"encoding/gob"
+	"bufio"
 	"log"
 	"os"
+	"strconv"
 	"time"
 )
 
@@ -11,7 +12,7 @@ import (
 type FileInput struct {
 	data        chan []byte
 	path        string
-	decoder     *gob.Decoder
+	file        *os.File
 	speedFactor float64
 }
 
@@ -35,7 +36,7 @@ func (i *FileInput) init(path string) {
 		log.Fatal(i, "Cannot open file %q. Error: %s", path, err)
 	}
 
-	i.decoder = gob.NewDecoder(file)
+	i.file = file
 }
 
 func (i *FileInput) Read(data []byte) (int, error) {
@@ -52,27 +53,36 @@ func (i *FileInput) String() string {
 func (i *FileInput) emit() {
 	var lastTime int64
 
-	for {
-		raw := new(RawRequest)
-		err := i.decoder.Decode(raw)
+	// reader := bufio.NewReader(conn)
+	scanner := bufio.NewScanner(i.file)
+	scanner.Split(payloadScanner)
 
-		if err != nil {
-			return
-		}
+	for scanner.Scan() {
+		buf := scanner.Bytes()
+		meta := payloadMeta(buf)
 
-		if lastTime != 0 {
-			timeDiff := raw.Timestamp - lastTime
+		if len(meta) > 2 && meta[0][0] == RequestPayload {
+			ts, _ := strconv.ParseInt(string(meta[2]), 10, 64)
 
-			// We can speedup or slowdown execution based on speedFactor
-			if i.speedFactor != 1 {
-				timeDiff = int64(float64(raw.Timestamp-lastTime) / i.speedFactor)
+			if lastTime != 0 {
+				timeDiff := ts - lastTime
+
+				if i.speedFactor != 1 {
+					timeDiff = int64(float64(timeDiff) / i.speedFactor)
+				}
+
+				time.Sleep(time.Duration(timeDiff))
 			}
 
-			time.Sleep(time.Duration(timeDiff))
+			lastTime = ts
 		}
 
-		lastTime = raw.Timestamp
+		// scanner returs only pointer, so to remove data-race we have to allocate new array
+		newBuf := make([]byte, len(buf))
+		copy(newBuf, buf)
 
-		i.data <- raw.Request
+		i.data <- newBuf
 	}
+
+	log.Println("FileInput: end of file")
 }

input_http.go

@@ -5,6 +5,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httputil"
+	"time"
 )
 
 // HTTPInput used for sending requests to Gor via http
@@ -17,7 +18,7 @@ type HTTPInput struct {
 // NewHTTPInput constructor for HTTPInput. Accepts address with port which he will listen on.
 func NewHTTPInput(address string) (i *HTTPInput) {
 	i = new(HTTPInput)
-	i.data = make(chan []byte)
+	i.data = make(chan []byte, 10000)
 	i.address = address
 
 	i.listen(address)
@@ -27,17 +28,27 @@ func NewHTTPInput(address string) (i *HTTPInput) {
 
 func (i *HTTPInput) Read(data []byte) (int, error) {
 	buf := <-i.data
-	copy(data, buf)
 
-	return len(buf), nil
+	header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano())
+
+	copy(data[0:len(header)], header)
+	copy(data[len(header):], buf)
+
+	return len(buf) + len(header), nil
 }
 
 func (i *HTTPInput) handler(w http.ResponseWriter, r *http.Request) {
-	buf, _ := httputil.DumpRequest(r, true)
-
-	i.data <- buf
+	r.URL.Scheme = "http"
+	r.URL.Host = i.listener.Addr().String()
 
+	buf, _ := httputil.DumpRequestOut(r, true)
 	http.Error(w, http.StatusText(200), 200)
+
+	select {
+	case i.data <- buf:
+	default:
+		Debug("[INPUT-HTTP] Dropping requests because output can't process them fast enough")
+	}
 }
 
 func (i *HTTPInput) listen(address string) {

input_http_test.go

@@ -1,8 +1,11 @@
 package main
 
 import (
+	"github.com/buger/gor/proto"
 	"io"
+	"log"
 	"net/http"
+	"os/exec"
 	"strings"
 	"sync"
 	"testing"
@@ -33,3 +36,38 @@ func TestHTTPInput(t *testing.T) {
 
 	close(quit)
 }
+
+func TestInputHTTPLargePayload(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	// Generate 1000kb file
+	dd := exec.Command("dd", "if=/dev/urandom", "of=/tmp/large", "bs=1MB", "count=4")
+	err := dd.Run()
+	if err != nil {
+		log.Fatal("dd error:", err)
+	}
+
+	input := NewHTTPInput(":0")
+	output := NewTestOutput(func(data []byte) {
+		if len(proto.Body(payloadBody(data))) != 4000000 {
+			t.Error("Should receive full file")
+		}
+		wg.Done()
+	})
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	go Start(quit)
+
+	wg.Add(1)
+	address := strings.Replace(input.listener.Addr().String(), "[::]", "127.0.0.1", -1)
+	curl := exec.Command("curl", "http://"+address, "--data-binary", "@/tmp/large")
+	err = curl.Run()
+	if err != nil {
+		log.Fatal("curl error:", err)
+	}
+
+	wg.Wait()
+	close(quit)
+}

input_raw.go

@@ -5,19 +5,25 @@ import (
 	"log"
 	"net"
 	"strings"
+	"time"
 )
 
 // RAWInput used for intercepting traffic for given address
 type RAWInput struct {
-	data    chan []byte
-	address string
+	data     chan *raw.TCPMessage
+	address  string
+	expire   time.Duration
+	quit     chan bool
+	listener *raw.Listener
 }
 
 // NewRAWInput constructor for RAWInput. Accepts address with port as argument.
-func NewRAWInput(address string) (i *RAWInput) {
+func NewRAWInput(address string, expire time.Duration) (i *RAWInput) {
 	i = new(RAWInput)
-	i.data = make(chan []byte)
+	i.data = make(chan *raw.TCPMessage)
 	i.address = address
+	i.expire = expire
+	i.quit = make(chan bool)
 
 	go i.listen(address)
 
@@ -25,31 +31,55 @@ func NewRAWInput(address string) (i *RAWInput) {
 }
 
 func (i *RAWInput) Read(data []byte) (int, error) {
-	buf := <-i.data
-	copy(data, buf)
+	msg := <-i.data
+	buf := msg.Bytes()
 
-	return len(buf), nil
+	var header []byte
+
+	if msg.IsIncoming {
+		header = payloadHeader(RequestPayload, msg.UUID(), msg.Start.UnixNano())
+	} else {
+		header = payloadHeader(ResponsePayload, msg.UUID(), msg.End.UnixNano()-msg.RequestStart.UnixNano())
+	}
+
+	copy(data[0:len(header)], header)
+	copy(data[len(header):], buf)
+
+	return len(buf) + len(header), nil
 }
 
 func (i *RAWInput) listen(address string) {
 	address = strings.Replace(address, "[::]", "127.0.0.1", -1)
 
+	Debug("Listening for traffic on: " + address)
+
 	host, port, err := net.SplitHostPort(address)
 
 	if err != nil {
 		log.Fatal("input-raw: error while parsing address", err)
 	}
 
-	listener := raw.NewListener(host, port)
+	i.listener = raw.NewListener(host, port, i.expire, true)
 
 	for {
-		// Receiving TCPMessage object
-		m := listener.Receive()
+		select {
+		case <-i.quit:
+			return
+		default:
+		}
 
-		i.data <- m.Bytes()
+		// Receiving TCPMessage object
+		m := i.listener.Receive()
+
+		i.data <- m
 	}
 }
 
 func (i *RAWInput) String() string {
 	return "RAW Socket input: " + i.address
 }
+
+func (i *RAWInput) Close() {
+	i.listener.Close()
+	close(i.quit)
+}

input_raw_test.go

@@ -11,40 +11,56 @@ import (
 	"os/exec"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 )
 
-func TestRAWInput(t *testing.T) {
+const testRawExpire = time.Millisecond * 200
 
+func TestRAWInput(t *testing.T) {
 	wg := new(sync.WaitGroup)
 	quit := make(chan int)
 
-	listener := startHTTP(func(req *http.Request) {})
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	defer origin.Close()
+	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
+
+	var respCounter, reqCounter int64
+
+	input := NewRAWInput(originAddr, testRawExpire)
+	defer input.Close()
 
-	input := NewRAWInput(listener.Addr().String())
 	output := NewTestOutput(func(data []byte) {
+		if data[0] == '1' {
+			atomic.AddInt64(&reqCounter, 1)
+		} else {
+			atomic.AddInt64(&respCounter, 1)
+		}
+
+		if Settings.debug {
+			log.Println(reqCounter, respCounter)
+		}
+
 		wg.Done()
 	})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{output}
 
-	address := strings.Replace(listener.Addr().String(), "[::]", "127.0.0.1", -1)
-
-	client := NewHTTPClient(address, &HTTPClientConfig{})
+	client := NewHTTPClient(origin.URL, &HTTPClientConfig{})
 
 	time.Sleep(time.Millisecond)
 
 	go Start(quit)
 
 	for i := 0; i < 100; i++ {
-		wg.Add(1)
+		// request + response
+		wg.Add(2)
 		client.Get("/")
 	}
 
 	wg.Wait()
-
 	close(quit)
 }
 
@@ -55,46 +71,54 @@ func TestInputRAW100Expect(t *testing.T) {
 	fileContent, _ := ioutil.ReadFile("README.md")
 
 	// Origing and Replay server initialization
-	origin := startHTTP(func(req *http.Request) {
-		defer req.Body.Close()
-		ioutil.ReadAll(req.Body)
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+		ioutil.ReadAll(r.Body)
 
 		wg.Done()
-	})
+	}))
+	defer origin.Close()
 
-	originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
+	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
 
-	input := NewRAWInput(originAddr)
+	input := NewRAWInput(originAddr, time.Second)
+	defer input.Close()
 
 	// We will use it to get content of raw HTTP request
 	testOutput := NewTestOutput(func(data []byte) {
-		if strings.Contains(string(data), "Expect: 100-continue") {
-			t.Error("Should not contain 100-continue header")
+		switch data[0] {
+		case RequestPayload:
+			if strings.Contains(string(data), "Expect: 100-continue") {
+				t.Error("Should not contain 100-continue header")
+			}
+			wg.Done()
+		case ResponsePayload:
+			wg.Done()
 		}
-		wg.Done()
 	})
 
-	listener := startHTTP(func(req *http.Request) {
-		defer req.Body.Close()
-		body, _ := ioutil.ReadAll(req.Body)
+	replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+		body, _ := ioutil.ReadAll(r.Body)
 
 		if !bytes.Equal(body, fileContent) {
-			buf, _ := httputil.DumpRequest(req, true)
+			buf, _ := httputil.DumpRequest(r, true)
 			t.Error("Wrong POST body:", string(buf))
 		}
 
 		wg.Done()
-	})
-	replayAddr := listener.Addr().String()
+	}))
+	defer replay.Close()
 
-	httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{})
+	httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{testOutput, httpOutput}
 
 	go Start(quit)
 
-	wg.Add(3)
+	// Origin + Response/Request Test Output + Request Http Output
+	wg.Add(4)
 	curl := exec.Command("curl", "http://"+originAddr, "--data-binary", "@README.md")
 	err := curl.Run()
 	if err != nil {
@@ -112,31 +136,31 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
 	fileContent, _ := ioutil.ReadFile("README.md")
 
 	// Origing and Replay server initialization
-	origin := startHTTP(func(req *http.Request) {
-		defer req.Body.Close()
-		ioutil.ReadAll(req.Body)
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+		ioutil.ReadAll(r.Body)
 
 		wg.Done()
-	})
+	}))
 
-	originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
+	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
+	input := NewRAWInput(originAddr, time.Second)
+	defer input.Close()
 
-	input := NewRAWInput(originAddr)
-
-	listener := startHTTP(func(req *http.Request) {
-		defer req.Body.Close()
-		body, _ := ioutil.ReadAll(req.Body)
+	replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+		body, _ := ioutil.ReadAll(r.Body)
 
 		if !bytes.Equal(body, fileContent) {
-			buf, _ := httputil.DumpRequest(req, true)
+			buf, _ := httputil.DumpRequest(r, true)
 			t.Error("Wrong POST body:", string(buf))
 		}
 
 		wg.Done()
-	})
-	replayAddr := listener.Addr().String()
+	}))
+	defer replay.Close()
 
-	httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{Debug: true})
+	httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: true})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{httpOutput}
@@ -167,8 +191,7 @@ func TestInputRAWLargePayload(t *testing.T) {
 		log.Fatal("dd error:", err)
 	}
 
-	// Origing and Replay server initialization
-	origin := startHTTP(func(req *http.Request) {
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		defer req.Body.Close()
 		body, _ := ioutil.ReadAll(req.Body)
 
@@ -177,10 +200,11 @@ func TestInputRAWLargePayload(t *testing.T) {
 		}
 
 		wg.Done()
-	})
-	originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
+	}))
+	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
 
-	input := NewRAWInput(originAddr)
+	input := NewRAWInput(originAddr, time.Second)
+	defer input.Close()
 
 	replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		req.Body = http.MaxBytesReader(w, req.Body, 1*1024*1024)

input_tcp.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"bufio"
-	"encoding/hex"
 	"fmt"
 	"log"
 	"net"
@@ -10,7 +9,6 @@ import (
 )
 
 // TCPInput used for internal communication
-// It expected hex encoded data
 type TCPInput struct {
 	data     chan []byte
 	address  string
@@ -62,16 +60,10 @@ func (i *TCPInput) handleConnection(conn net.Conn) {
 
 	reader := bufio.NewReader(conn)
 	scanner := bufio.NewScanner(reader)
+	scanner.Split(payloadScanner)
 
 	for scanner.Scan() {
-		encodedPayload := scanner.Bytes()
-		// Hex encoding always 2x number of bytes
-		decoded := make([]byte, len(encodedPayload)/2)
-		_, err := hex.Decode(decoded, encodedPayload)
-		if err != nil {
-			log.Println("[TCPInput] failed to hex decode TCP payload:", err)
-		}
-		i.data <- decoded
+		i.data <- scanner.Bytes()
 	}
 
 	if err := scanner.Err(); err != nil {

input_tcp_test.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"encoding/hex"
 	"io"
 	"log"
 	"net"
@@ -35,14 +34,12 @@ func TestTCPInput(t *testing.T) {
 		log.Fatal(err)
 	}
 
-	msg := []byte("GET / HTTP/1.1\r\n\r\n")
+	msg := []byte("1 1 1\nGET / HTTP/1.1\r\n\r\n")
 
 	for i := 0; i < 100; i++ {
 		wg.Add(1)
-
-		encoded := make([]byte, len(msg)*2+1)
-		hex.Encode(encoded, msg)
-		conn.Write(append(encoded, '\n'))
+		conn.Write(msg)
+		conn.Write([]byte(payloadSeparator))
 	}
 
 	wg.Wait()

middleware.go

@@ -0,0 +1,120 @@
+package main
+
+import (
+	"bufio"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"os/exec"
+	"strings"
+	"sync"
+)
+
+type Middleware struct {
+	command string
+
+	data chan []byte
+
+	mu sync.Mutex
+
+	Stdin  io.Writer
+	Stdout io.Reader
+}
+
+func NewMiddleware(command string) *Middleware {
+	m := new(Middleware)
+	m.command = command
+	m.data = make(chan []byte, 1000)
+
+	commands := strings.Split(command, " ")
+	cmd := exec.Command(commands[0], commands[1:]...)
+
+	m.Stdout, _ = cmd.StdoutPipe()
+	m.Stdin, _ = cmd.StdinPipe()
+
+	if Settings.verbose {
+		cmd.Stderr = os.Stderr
+	}
+
+	go m.read(m.Stdout)
+
+	go func() {
+		err := cmd.Start()
+
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		err = cmd.Wait()
+
+		if err != nil {
+			log.Fatal(err)
+		}
+	}()
+
+	return m
+}
+
+func (m *Middleware) ReadFrom(plugin io.Reader) {
+	Debug("[MIDDLEWARE-MASTER] Starting reading from", plugin)
+	go m.copy(m.Stdin, plugin)
+}
+
+func (m *Middleware) copy(to io.Writer, from io.Reader) {
+	buf := make([]byte, 5*1024*1024)
+	dst := make([]byte, len(buf)*2)
+
+	for {
+		nr, _ := from.Read(buf)
+		if nr > 0 && len(buf) > nr {
+
+			hex.Encode(dst, buf[0:nr])
+			dst[nr*2] = '\n'
+
+			m.mu.Lock()
+			to.Write(dst[0 : nr*2+1])
+			m.mu.Unlock()
+
+			if Settings.debug {
+				Debug("[MIDDLEWARE-MASTER] Sending:", string(buf[0:nr]), "From:", from)
+			}
+		}
+	}
+}
+
+func (m *Middleware) read(from io.Reader) {
+	scanner := bufio.NewScanner(from)
+
+	for scanner.Scan() {
+		bytes := scanner.Bytes()
+		buf := make([]byte, len(bytes)/2)
+		if _, err := hex.Decode(buf, bytes); err != nil {
+			fmt.Fprintln(os.Stderr, "Failed to decode input payload", err, len(bytes))
+		}
+
+		if Settings.debug {
+			Debug("[MIDDLEWARE-MASTER] Received:", string(buf))
+		}
+
+		m.data <- buf
+	}
+
+	if err := scanner.Err(); err != nil {
+		fmt.Fprintln(os.Stderr, "Traffic modifier command failed:", err)
+	}
+
+	return
+}
+
+func (m *Middleware) Read(data []byte) (int, error) {
+	buf := <-m.data
+	copy(data, buf)
+
+	return len(buf), nil
+}
+
+func (m *Middleware) String() string {
+	return fmt.Sprintf("Modifying traffic using '%s' command", m.command)
+}

middleware_test.go

@@ -0,0 +1,216 @@
+package main
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/hex"
+	"github.com/buger/gor/proto"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+type fakeServiceCb func(string, int, []byte)
+
+// Simple service that generate token on request, and require this token for accesing to secure area
+func NewFakeSecureService(wg *sync.WaitGroup, cb fakeServiceCb) *httptest.Server {
+	active_tokens := make([]string, 0)
+	var mu sync.Mutex
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		mu.Lock()
+		defer mu.Unlock()
+		Debug("Received request: " + req.URL.String())
+
+		switch req.URL.Path {
+		case "/token":
+			// Generate random token
+			token_length := 10
+			buf := make([]byte, token_length)
+			rand.Read(buf)
+			token := hex.EncodeToString(buf)
+			active_tokens = append(active_tokens, token)
+
+			w.Write([]byte(token))
+
+			cb(req.URL.Path, 200, []byte(token))
+		case "/secure":
+			token := req.URL.Query().Get("token")
+			token_found := false
+
+			for _, t := range active_tokens {
+				if t == token {
+					token_found = true
+					break
+				}
+			}
+
+			if token_found {
+				w.WriteHeader(http.StatusAccepted)
+				cb(req.URL.Path, 202, []byte(nil))
+			} else {
+				w.WriteHeader(http.StatusForbidden)
+				cb(req.URL.Path, 403, []byte(nil))
+			}
+		}
+
+		wg.Done()
+	}))
+
+	return server
+}
+
+func TestFakeSecureService(t *testing.T) {
+	var resp, token []byte
+
+	wg := new(sync.WaitGroup)
+
+	server := NewFakeSecureService(wg, func(path string, status int, resp []byte) {
+	})
+	defer server.Close()
+
+	wg.Add(3)
+
+	client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true})
+	resp, _ = client.Get("/token")
+	token = proto.Body(resp)
+
+	// Right token
+	resp, _ = client.Get("/secure?token=" + string(token))
+	if !bytes.Equal(proto.Status(resp), []byte("202")) {
+		t.Error("Valid token should return status 202:", string(proto.Status(resp)))
+	}
+
+	// Wrong tokens forbidden
+	resp, _ = client.Get("/secure?token=wrong")
+	if !bytes.Equal(proto.Status(resp), []byte("403")) {
+		t.Error("Wrong token should returns status 403:", string(proto.Status(resp)))
+	}
+
+	wg.Wait()
+}
+
+func TestEchoMiddleware(t *testing.T) {
+	wg := new(sync.WaitGroup)
+
+	from := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Env", "prod")
+		w.Header().Set("RequestPath", r.URL.Path)
+		wg.Done()
+	}))
+	defer from.Close()
+
+	to := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Env", "test")
+		w.Header().Set("RequestPath", r.URL.Path)
+		wg.Done()
+	}))
+	defer to.Close()
+
+	quit := make(chan int)
+
+	Settings.middleware = "./examples/middleware/echo.sh"
+
+	// Catch traffic from one service
+	input := NewRAWInput(from.Listener.Addr().String(), testRawExpire)
+	defer input.Close()
+
+	// And redirect to another
+	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: false})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	// Start Gor
+	go Start(quit)
+
+	time.Sleep(time.Millisecond)
+
+	// Should receive 2 requests from original + 2 from replayed
+	client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: false})
+
+	for i := 0; i < 10; i++ {
+		wg.Add(4)
+		// Request should be echoed
+		client.Get("/a")
+		client.Get("/b")
+	}
+
+	wg.Wait()
+	close(quit)
+	time.Sleep(200 * time.Millisecond)
+
+	Settings.middleware = ""
+}
+
+func TestTokenMiddleware(t *testing.T) {
+	var resp, token []byte
+
+	wg := new(sync.WaitGroup)
+
+	from := NewFakeSecureService(wg, func(path string, status int, tok []byte) {
+		time.Sleep(10 * time.Millisecond)
+	})
+	defer from.Close()
+
+	to := NewFakeSecureService(wg, func(path string, status int, tok []byte) {
+		switch path {
+		case "/secure":
+			if status != 202 {
+				t.Error("Server should receive valid rewritten token")
+			}
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	})
+	defer to.Close()
+
+	quit := make(chan int)
+
+	Settings.middleware = "go run ./examples/middleware/token_modifier.go"
+
+	fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
+	// Catch traffic from one service
+	input := NewRAWInput(fromAddr, testRawExpire)
+	defer input.Close()
+
+	// And redirect to another
+	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: true})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	// Start Gor
+	go Start(quit)
+
+	// Wait for middleware to initialize
+	// Give go compiller time to build programm
+	time.Sleep(500 * time.Millisecond)
+
+	// Should receive 2 requests from original + 2 from replayed
+	wg.Add(4)
+
+	client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: false})
+
+	// Sending traffic to original service
+	resp, _ = client.Get("/token")
+	token = proto.Body(resp)
+
+	// When delay is too smal, middleware does not always rewrite requests in time
+	// Hopefuly client will have delay more then 10ms :)
+	time.Sleep(10 * time.Millisecond)
+
+	resp, _ = client.Get("/secure?token=" + string(token))
+	if !bytes.Equal(proto.Status(resp), []byte("202")) {
+		t.Error("Valid token should return 202:", proto.Status(resp))
+	}
+
+	wg.Wait()
+	close(quit)
+	time.Sleep(100 * time.Millisecond)
+	Settings.middleware = ""
+}

output_file.go

@@ -1,24 +1,15 @@
 package main
 
 import (
-	"encoding/gob"
 	"io"
 	"log"
 	"os"
-	"time"
 )
 
-// RawRequest stores original start time and request payload
-type RawRequest struct {
-	Timestamp int64
-	Request   []byte
-}
-
 // FileOutput output plugin
 type FileOutput struct {
-	path    string
-	encoder *gob.Encoder
-	file    *os.File
+	path string
+	file *os.File
 }
 
 // NewFileOutput constructor for FileOutput, accepts path
@@ -38,14 +29,15 @@ func (o *FileOutput) init(path string) {
 	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}
+	if !isOriginPayload(data) {
+		return len(data), nil
+	}
 
-	o.encoder.Encode(raw)
+	o.file.Write(data)
+	o.file.Write([]byte(payloadSeparator))
 
 	return len(data), nil
 }

output_http.go

@@ -1,14 +1,20 @@
 package main
 
 import (
+	"github.com/buger/gor/proto"
 	"io"
-	"log"
 	"sync/atomic"
 	"time"
 )
 
 const initialDynamicWorkers = 10
 
+type response struct {
+	payload       []byte
+	uuid          []byte
+	roundTripTime int64
+}
+
 // HTTPOutputConfig struct for holding http output configuration
 type HTTPOutputConfig struct {
 	redirectLimit int
@@ -22,6 +28,8 @@ type HTTPOutputConfig struct {
 	OriginalHost bool
 
 	Debug bool
+
+	TrackResponses bool
 }
 
 // HTTPOutput plugin manage pool of workers which send request to replayed server
@@ -37,6 +45,8 @@ type HTTPOutput struct {
 	limit   int
 	queue   chan []byte
 
+	responses chan response
+
 	needWorker chan int
 
 	config *HTTPOutputConfig
@@ -49,7 +59,6 @@ type HTTPOutput struct {
 // NewHTTPOutput constructor for HTTPOutput
 // Initialize workers
 func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer {
-
 	o := new(HTTPOutput)
 
 	o.address = address
@@ -59,7 +68,8 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer {
 		o.queueStats = NewGorStat("output_http")
 	}
 
-	o.queue = make(chan []byte, 100)
+	o.queue = make(chan []byte, 1000)
+	o.responses = make(chan response, 1000)
 	o.needWorker = make(chan int, 1)
 
 	// Initial workers count
@@ -74,6 +84,10 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer {
 		o.elasticSearch.Init(o.config.elasticSearch)
 	}
 
+	if len(Settings.middleware) > 0 {
+		o.config.TrackResponses = true
+	}
+
 	go o.workerMaster()
 
 	return o
@@ -132,6 +146,10 @@ func (o *HTTPOutput) startWorker() {
 }
 
 func (o *HTTPOutput) Write(data []byte) (n int, err error) {
+	if !isRequestPayload(data) {
+		return len(data), nil
+	}
+
 	buf := make([]byte, len(data))
 	copy(buf, data)
 
@@ -152,13 +170,40 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) {
 	return len(data), nil
 }
 
+func (o *HTTPOutput) Read(data []byte) (int, error) {
+	resp := <-o.responses
+
+	Debug("[OUTPUT-HTTP] Received response:", string(resp.payload))
+
+	header := payloadHeader(ReplayedResponsePayload, resp.uuid, resp.roundTripTime)
+	copy(data[0:len(header)], header)
+	copy(data[len(header):], resp.payload)
+
+	return len(resp.payload) + len(header), nil
+}
+
 func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) {
+	meta := payloadMeta(request)
+	if len(meta) < 2 {
+		return
+	}
+	uuid := meta[1]
+
+	body := payloadBody(request)
+	if !proto.IsHTTPPayload(body) {
+		return
+	}
+
 	start := time.Now()
-	resp, err := client.Send(request)
+	resp, err := client.Send(body)
 	stop := time.Now()
 
 	if err != nil {
-		log.Println("Request error:", err)
+		Debug("Request error:", err)
+	}
+
+	if o.config.TrackResponses {
+		o.responses <- response{resp, uuid, stop.UnixNano() - start.UnixNano()}
 	}
 
 	if o.elasticSearch != nil {

output_http_test.go

@@ -3,7 +3,6 @@ package main
 import (
 	"io"
 	"io/ioutil"
-	"net"
 	"net/http"
 	"net/http/httptest"
 	_ "net/http/httputil"
@@ -12,25 +11,13 @@ import (
 	"time"
 )
 
-func startHTTP(cb func(*http.Request)) net.Listener {
-	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		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()
 
-	listener := startHTTP(func(req *http.Request) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		if req.Header.Get("User-Agent") != "Gor" {
 			t.Error("Wrong header")
 		}
@@ -49,13 +36,14 @@ func TestHTTPOutput(t *testing.T) {
 		}
 
 		wg.Done()
-	})
+	}))
+	defer server.Close()
 
 	headers := HTTPHeaders{HTTPHeader{"User-Agent", "Gor"}}
 	methods := HTTPMethods{[]byte("GET"), []byte("PUT"), []byte("POST")}
 	Settings.modifierConfig = HTTPModifierConfig{headers: headers, methods: methods}
 
-	output := NewHTTPOutput(listener.Addr().String(), &HTTPOutputConfig{})
+	output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: true})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{output}
@@ -82,18 +70,19 @@ func TestHTTPOutputKeepOriginalHost(t *testing.T) {
 
 	input := NewTestInput()
 
-	listener := startHTTP(func(req *http.Request) {
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		if req.Host != "custom-host.com" {
 			t.Error("Wrong header", req.Host)
 		}
 
 		wg.Done()
-	})
+	}))
+	defer server.Close()
 
 	headers := HTTPHeaders{HTTPHeader{"Host", "custom-host.com"}}
 	Settings.modifierConfig = HTTPModifierConfig{headers: headers}
 
-	output := NewHTTPOutput(listener.Addr().String(), &HTTPOutputConfig{Debug: false, OriginalHost: true})
+	output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: false, OriginalHost: true})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{output}
@@ -140,13 +129,14 @@ func BenchmarkHTTPOutput(b *testing.B) {
 	wg := new(sync.WaitGroup)
 	quit := make(chan int)
 
-	listener := startHTTP(func(req *http.Request) {
+	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		time.Sleep(50 * time.Millisecond)
 		wg.Done()
-	})
+	}))
+	defer server.Close()
 
 	input := NewTestInput()
-	output := NewHTTPOutput(listener.Addr().String(), &HTTPOutputConfig{})
+	output := NewHTTPOutput(server.URL, &HTTPOutputConfig{})
 
 	Plugins.Inputs = []io.Reader{input}
 	Plugins.Outputs = []io.Writer{output}

output_tcp.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"encoding/hex"
 	"fmt"
 	"io"
 	"log"
@@ -47,7 +46,9 @@ func (o *TCPOutput) worker() {
 	defer conn.Close()
 
 	for {
-		_, err := conn.Write(<-o.buf)
+		conn.Write(<-o.buf)
+		_, err := conn.Write([]byte(payloadSeparator))
+
 		if err != nil {
 			log.Println("Worker failed on write, exitings and starting new worker")
 			go o.worker()
@@ -57,11 +58,15 @@ func (o *TCPOutput) worker() {
 }
 
 func (o *TCPOutput) Write(data []byte) (n int, err error) {
-	// Hex encoding always 2x number of bytes
-	encoded := make([]byte, len(data)*2+1)
-	hex.Encode(encoded, data)
-	encoded[len(encoded)-1] = '\n'
-	o.buf <- encoded
+	if !isOriginPayload(data) {
+		return len(data), nil
+	}
+
+	// We have to copy, because sending data in multiple threads
+	newBuf := make([]byte, len(data))
+	copy(newBuf, data)
+
+	o.buf <- newBuf
 
 	if Settings.outputTCPStats {
 		o.bufStats.Write(len(o.buf))

output_tcp_test.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"bufio"
-	"encoding/hex"
 	"io"
 	"log"
 	"net"
@@ -50,13 +49,10 @@ func startTCP(cb func([]byte)) net.Listener {
 			go func() {
 				reader := bufio.NewReader(conn)
 				scanner := bufio.NewScanner(reader)
+				scanner.Split(payloadScanner)
 
 				for scanner.Scan() {
-					encodedPayload := scanner.Bytes()
-					// Hex encoding always 2x number of bytes
-					decoded := make([]byte, len(encodedPayload)/2)
-					hex.Decode(decoded, encodedPayload)
-					cb(decoded)
+					cb(scanner.Bytes())
 				}
 			}()
 		}

plugins.go

@@ -4,6 +4,7 @@ import (
 	"io"
 	"reflect"
 	"strings"
+	"time"
 )
 
 // InOutPlugins struct for holding references to plugins
@@ -55,11 +56,15 @@ func registerPlugin(constructor interface{}, options ...interface{}) {
 		pluginWrapper = plugin
 	}
 
-	if _, ok := plugin.(io.Reader); ok {
+	_, isR := plugin.(io.Reader)
+	_, isW := plugin.(io.Writer)
+
+	// Some of the output can be Readers as well because return responses
+	if isR && !isW {
 		Plugins.Inputs = append(Plugins.Inputs, pluginWrapper.(io.Reader))
 	}
 
-	if _, ok := plugin.(io.Writer); ok {
+	if isW {
 		Plugins.Outputs = append(Plugins.Outputs, pluginWrapper.(io.Writer))
 	}
 }
@@ -75,7 +80,7 @@ func InitPlugins() {
 	}
 
 	for _, options := range Settings.inputRAW {
-		registerPlugin(NewRAWInput, options)
+		registerPlugin(NewRAWInput, options, time.Duration(0))
 	}
 
 	for _, options := range Settings.inputTCP {

proto/proto.go

@@ -55,7 +55,13 @@ func header(payload []byte, name []byte) (value []byte, headerStart, valueStart,
 	if payload[valueStart] == ' ' {          // Ignore empty space after ':'
 		valueStart++
 	}
-	headerEnd = valueStart + bytes.IndexByte(payload[valueStart:], '\r')
+
+	headerEnd = valueStart + bytes.IndexByte(payload[valueStart:], '\n')
+
+	if payload[headerEnd - 1] == '\r' {
+		headerEnd -= 1
+	}
+
 	value = payload[valueStart:headerEnd]
 
 	return
@@ -95,6 +101,12 @@ func AddHeader(payload, name, value []byte) []byte {
 	return byteutils.Insert(payload, mimeStart, header)
 }
 
+// Body returns request/response body
+func Body(payload []byte) []byte {
+	// 4 -> len(EMPTY_LINE)
+	return payload[MIMEHeadersEndPos(payload)+4:]
+}
+
 // Path takes payload and retuns request path: Split(firstLine, ' ')[1]
 func Path(payload []byte) []byte {
 	start := bytes.IndexByte(payload, ' ') + 1
@@ -202,3 +214,22 @@ func Method(payload []byte) []byte {
 func Status(payload []byte) []byte {
 	return Path(payload)
 }
+
+var httpMethods []string = []string{
+	"GET ", "OPTI", "HEAD", "POST", "PUT ", "DELE", "TRAC", "CONN",
+}
+
+func IsHTTPPayload(payload []byte) bool {
+	if len(payload) < 4 {
+		return false
+	}
+
+	method := string(payload[0:4])
+
+	for _, m := range httpMethods {
+		if method == m {
+			return true
+		}
+	}
+	return false
+}

proto/proto_test.go

@@ -9,18 +9,36 @@ func TestHeader(t *testing.T) {
 	var payload, val []byte
 	var headerStart int
 
+	// Value with space at start
 	payload = []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
 
 	if val = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("7")) {
 		t.Error("Should find header value")
 	}
 
+	// Value without space at start
 	payload = []byte("POST /post HTTP/1.1\r\nContent-Length:7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
 
 	if val = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("7")) {
 		t.Error("Should find header value without space after :")
 	}
 
+	// Value is empty
+	payload = []byte("GET /p HTTP/1.1\r\nCookie:\r\nHost: www.w3.org\r\n\r\n")
+
+	if val = Header(payload, []byte("Cookie")); len(val) > 0 {
+		t.Error("Should return empty value")
+	}
+
+	// Wrong delimeter
+	payload = []byte("GET /p HTTP/1.1\r\nCookie: 123\nHost: www.w3.org\r\n\r\n")
+
+	if val = Header(payload, []byte("Cookie")); !bytes.Equal(val, []byte("123")) {
+		t.Error("Should handle wrong header delimeter")
+	}
+
+
+	// Header not found
 	if _, headerStart, _, _ = header(payload, []byte("Not-Found")); headerStart != -1 {
 		t.Error("Should not found header")
 	}

protocol.go

@@ -0,0 +1,87 @@
+package main
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/hex"
+	"strconv"
+)
+
+const (
+	RequestPayload          = '1'
+	ResponsePayload         = '2'
+	ReplayedResponsePayload = '3'
+)
+
+func uuid() []byte {
+	b := make([]byte, 20)
+	rand.Read(b)
+
+	uuid := make([]byte, 40)
+	hex.Encode(uuid, b)
+
+	return uuid
+}
+
+var payloadSeparator = "\n🐵🙈🙉\n"
+
+func payloadScanner(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	if atEOF && len(data) == 0 {
+		return 0, nil, nil
+	}
+
+	if i := bytes.Index(data, []byte(payloadSeparator)); i >= 0 {
+		// We have a full newline-terminated line.
+		return i + len([]byte(payloadSeparator)), data[0:i], nil
+	}
+
+	if atEOF {
+		return len(data), data, nil
+	}
+	return 0, nil, nil
+}
+
+// Timing is request start or round-trip time, depending on payloadType
+func payloadHeader(payloadType byte, uuid []byte, timing int64) (header []byte) {
+	sTime := strconv.FormatInt(timing, 10)
+
+	//Example:
+	//  3 f45590522cd1838b4a0d5c5aab80b77929dea3b3 1231\n
+	// `+ 1` indicates space characters or end of line
+	header = make([]byte, 1+1+len(uuid)+1+len(sTime)+1)
+	header[0] = payloadType
+	header[1] = ' '
+	header[2+len(uuid)] = ' '
+	header[len(header)-1] = '\n'
+
+	copy(header[2:], uuid)
+	copy(header[3+len(uuid):], sTime)
+
+	return header
+}
+
+func payloadBody(payload []byte) []byte {
+	headerSize := bytes.IndexByte(payload, '\n')
+	return payload[headerSize+1:]
+}
+
+func payloadMeta(payload []byte) [][]byte {
+	headerSize := bytes.IndexByte(payload, '\n')
+        if headerSize < 0 {
+                headerSize = 0;
+	}
+	return bytes.Split(payload[:headerSize], []byte{' '})
+}
+
+func isOriginPayload(payload []byte) bool {
+	switch payload[0] {
+	case RequestPayload, ResponsePayload:
+		return true
+	default:
+		return false
+	}
+}
+
+func isRequestPayload(payload []byte) bool {
+	return payload[0] == RequestPayload
+}

raw_socket_listener/listener.go

@@ -17,7 +17,10 @@ import (
 	"encoding/binary"
 	"log"
 	"net"
+	"runtime/debug"
 	"strconv"
+	"strings"
+	"time"
 )
 
 // Listener handle traffic capture
@@ -31,93 +34,144 @@ type Listener struct {
 	// To get ACK of second message we need to compute its Seq and wait for them message
 	seqWithData map[uint32]uint32
 
+	respAliases map[uint32]*request
+
 	// Messages ready to be send to client
 	packetsChan chan *TCPPacket
 
 	// Messages ready to be send to client
 	messagesChan chan *TCPMessage
 
-	// Used for notifications about completed or expired messages
-	messageDelChan chan *TCPMessage
-
 	addr string // IP to listen
-	port int    // Port to listen
+	port uint16 // Port to listen
+
+	messageExpire time.Duration
+
+	captureResponse bool
+
+	conn net.PacketConn
+	quit chan bool
+}
+
+type request struct {
+	start time.Time
+	ack   uint32
 }
 
 // NewListener creates and initializes new Listener object
-func NewListener(addr string, port string) (rawListener *Listener) {
-	rawListener = &Listener{}
+func NewListener(addr string, port string, expire time.Duration, captureResponse bool) (l *Listener) {
+	l = &Listener{captureResponse: captureResponse}
 
-	rawListener.packetsChan = make(chan *TCPPacket, 10000)
-	rawListener.messagesChan = make(chan *TCPMessage, 10000)
-	rawListener.messageDelChan = make(chan *TCPMessage, 10000)
+	l.packetsChan = make(chan *TCPPacket, 10000)
+	l.messagesChan = make(chan *TCPMessage, 10000)
+	l.quit = make(chan bool)
 
-	rawListener.messages = make(map[string]*TCPMessage)
-	rawListener.ackAliases = make(map[uint32]uint32)
-	rawListener.seqWithData = make(map[uint32]uint32)
+	l.messages = make(map[string]*TCPMessage)
+	l.ackAliases = make(map[uint32]uint32)
+	l.seqWithData = make(map[uint32]uint32)
+	l.respAliases = make(map[uint32]*request)
 
-	rawListener.addr = addr
-	rawListener.port, _ = strconv.Atoi(port)
+	l.addr = addr
+	_port, _ := strconv.Atoi(port)
+	l.port = uint16(_port)
 
-	go rawListener.listen()
-	go rawListener.readRAWSocket()
+	if expire.Nanoseconds() == 0 {
+		expire = 2000 * time.Millisecond
+	}
+
+	l.messageExpire = expire
+
+	go l.listen()
+	go l.readRAWSocket()
 
 	return
 }
 
 func (t *Listener) listen() {
+	gcTicker := time.Tick(t.messageExpire / 2)
+
 	for {
 		select {
-		// If message ready for deletion it means that its also complete or expired by timeout
-		case message := <-t.messageDelChan:
-			t.messagesChan <- message
-			delete(t.ackAliases, message.Ack)
-			delete(t.messages, message.ID)
-
+		case <-t.quit:
+			t.conn.Close()
+			return
 		// We need to use channels to process each packet to avoid data races
 		case packet := <-t.packetsChan:
 			t.processTCPPacket(packet)
+
+		case <- gcTicker:
+			now := time.Now()
+
+			for _, message := range t.messages {
+				if now.Sub(message.Start) >= t.messageExpire {
+					t.dispatchMessage(message)
+				}
+			}
 		}
 	}
 }
+
+func (t *Listener) dispatchMessage(message *TCPMessage) {
+	delete(t.ackAliases, message.Ack)
+	delete(t.messages, message.ID)
+
+	if !message.IsIncoming {
+		delete(t.respAliases, message.Ack)
+
+		// Do not track responses which have no associated requests
+		if message.RequestAck == 0 {
+			return
+		}
+	}
+	t.messagesChan <- message
+}
+
 func (t *Listener) readRAWSocket() {
 	conn, e := net.ListenPacket("ip4:tcp", t.addr)
+	t.conn = conn
 
 	if e != nil {
 		log.Fatal(e)
 	}
 
-	defer conn.Close()
+	defer t.conn.Close()
+
+	buf := make([]byte, 64*1024) // 64kb
 
 	for {
-		buf := make([]byte, 64*1024) // 64kb
 		// Note: ReadFrom receive messages without IP header
-		n, addr, err := conn.ReadFrom(buf)
+		n, addr, err := t.conn.ReadFrom(buf)
 
 		if err != nil {
-			log.Println("Error:", err)
-			continue
+			if strings.HasSuffix(err.Error(), "closed network connection") {
+				return
+			} else {
+				log.Println("Raw listener error:", err)
+				continue
+			}
 		}
 
 		if n > 0 {
-			go t.parsePacket(addr, buf[:n])
+			if t.isValidPacket(buf[:n]) {
+				newBuf := make([]byte, n)
+				copy(newBuf, buf[:n])
+
+				go func(newBuf []byte) {
+					t.packetsChan <- ParseTCPPacket(addr, newBuf)
+				}(newBuf)
+			}
 		}
 	}
 }
 
-func (t *Listener) parsePacket(addr net.Addr, buf []byte) {
-	if t.isIncomingDataPacket(buf) {
-		t.packetsChan <- ParseTCPPacket(addr, buf)
-	}
-}
-
-func (t *Listener) isIncomingDataPacket(buf []byte) bool {
+func (t *Listener) isValidPacket(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
 	destPort := binary.BigEndian.Uint16(buf[2:4])
+	srcPort := binary.BigEndian.Uint16(buf[0:2])
 
 	// Because RAW_SOCKET can't be bound to port, we have to control it by ourself
-	if int(destPort) == t.port {
+	if destPort == t.port || (t.captureResponse && srcPort == t.port) {
 		// Get the 'data offset' (size of the TCP header in 32-bit words)
 		dataOffset := (buf[12] & 0xF0) >> 4
 
@@ -139,10 +193,17 @@ var bPOST = []byte("POST")
 //
 // For TCP message unique id is Acknowledgment number (see tcp_packet.go)
 func (t *Listener) processTCPPacket(packet *TCPPacket) {
-	defer func() { recover() }()
+	// Don't exit on panic
+	defer func() {
+		if r := recover(); r != nil {
+			log.Println("PANIC: pkg:", r, packet, string(debug.Stack()))
+		}
+	}()
 
 	var message *TCPMessage
 
+	isIncoming := packet.DestPort == t.port
+
 	if parentAck, ok := t.seqWithData[packet.Seq]; ok {
 		t.ackAliases[packet.Ack] = parentAck
 		delete(t.seqWithData, packet.Seq)
@@ -152,16 +213,29 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 		packet.Ack = alias
 	}
 
-	mID := packet.Addr.String() + strconv.Itoa(int(packet.SrcPort)) + strconv.Itoa(int(packet.Ack))
+	var responseRequest *request
+
+	if t.captureResponse && !isIncoming {
+		responseRequest, _ = t.respAliases[packet.Ack]
+	}
+
+	mID := packet.Addr.String() + strconv.Itoa(int(packet.DestPort)) + strconv.Itoa(int(packet.Ack))
+
 	message, ok := t.messages[mID]
 
 	if !ok {
-		// We sending messageDelChan channel, so message object can communicate with Listener and notify it if message completed
-		message = NewTCPMessage(mID, t.messageDelChan, packet.Ack)
+		message = NewTCPMessage(mID, packet.Ack, isIncoming)
 		t.messages[mID] = message
+
+		if !isIncoming && responseRequest != nil {
+			message.RequestStart = responseRequest.start
+			message.RequestAck = responseRequest.ack
+		}
 	}
 
-	if bytes.Equal(packet.Data[0:4], bPOST) {
+	// Handling Expect: 100-continue requests
+	if len(packet.Data) > 4 && bytes.Equal(packet.Data[0:4], bPOST) {
+		// reading last 20 bytes (not counting CRLF): last header value (if no body presented)
 		if bytes.Equal(packet.Data[len(packet.Data)-24:len(packet.Data)-4], bExpect100ContinueCheck) {
 			t.seqWithData[packet.Seq+uint32(len(packet.Data))] = packet.Ack
 
@@ -170,11 +244,33 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 		}
 	}
 
+	if t.captureResponse && isIncoming {
+		// If message have multiple packets, delete previous alias
+		if len(message.packets) > 0 {
+			delete(t.respAliases, message.ResponseAck)
+		}
+
+		responseAck := packet.Seq + uint32(len(packet.Data))
+		t.respAliases[responseAck] = &request{message.Start, message.Ack}
+		message.ResponseAck = responseAck
+	}
+
 	// Adding packet to message
-	message.packetsChan <- packet
+	message.AddPacket(packet)
+
+	// If message contains only single packet immediately dispatch it
+	if !message.IsMultipart() {
+		t.dispatchMessage(message)
+	}
 }
 
 // Receive TCP messages from the listener channel
 func (t *Listener) Receive() *TCPMessage {
 	return <-t.messagesChan
 }
+
+func (t *Listener) Close() {
+	close(t.quit)
+	t.conn.Close()
+	return
+}

raw_socket_listener/tcp_message.go

@@ -1,14 +1,15 @@
 package rawSocket
 
 import (
+	"bytes"
+	"crypto/sha1"
+	"encoding/hex"
+	"github.com/buger/gor/proto"
 	"log"
-	"sort"
+	"strconv"
 	"time"
 )
 
-// MsgExpire specify period that message should wait before it considered as finished
-const MsgExpire = 2000 * 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
 //
@@ -16,74 +17,48 @@ const MsgExpire = 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
-	Ack     uint32
+	ID           string // Message ID
+	Ack          uint32
+	ResponseAck  uint32
+	RequestStart time.Time
+	RequestAck   uint32
+	Start        time.Time
+	End          time.Time
+	IsIncoming   bool
+
 	packets []*TCPPacket
 
-	timer *time.Timer // Used for expire check
-
-	packetsChan chan *TCPPacket
-
 	delChan chan *TCPMessage
 }
 
 // NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted
-func NewTCPMessage(ID string, delChan chan *TCPMessage, Ack uint32) (msg *TCPMessage) {
-	msg = &TCPMessage{ID: ID, Ack: Ack}
-
-	msg.packetsChan = make(chan *TCPPacket)
-	msg.delChan = delChan // used for notifying that message completed or expired
-
-	// Every time we receive packet we reset this timer
-	msg.timer = time.AfterFunc(MsgExpire, msg.Timeout)
-
-	go msg.listen()
+func NewTCPMessage(ID string, Ack uint32, IsIncoming bool) (msg *TCPMessage) {
+	msg = &TCPMessage{ID: ID, Ack: Ack, IsIncoming: IsIncoming}
+	msg.Start = time.Now()
 
 	return
 }
 
-func (t *TCPMessage) listen() {
-	for {
-		select {
-		case packet, more := <-t.packetsChan:
-			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() {
-	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.packetsChan:
-		if ok {
-			t.AddPacket(packet)
-		} else {
-			return
-		}
-	default:
-		close(t.packetsChan)
-		t.delChan <- t // Notify RAWListener that message is ready to be send to replay server
-	}
-}
-
-// Bytes sorts packets in right orders and return message content
+// Bytes return message content
 func (t *TCPMessage) Bytes() (output []byte) {
-	sort.Sort(sortBySeq(t.packets))
-
-	for _, v := range t.packets {
-		output = append(output, v.Data...)
+	for _, p := range t.packets {
+		output = append(output, p.Data...)
 	}
 
 	return output
 }
 
+// Size returns total size of message
+func (t *TCPMessage) Size() (size int) {
+	size += len(proto.Body(t.packets[0].Data))
+
+	for _, p := range t.packets[1:] {
+		size += len(p.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) {
@@ -99,9 +74,80 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) {
 	if packetFound {
 		log.Println("Received packet with same sequence")
 	} else {
-		t.packets = append(t.packets, packet)
+		// Packets not always captured in same Seq order, and sometimes we need to prepend
+		if len(t.packets) == 0 || packet.Seq > t.packets[len(t.packets)-1].Seq {
+			t.packets = append(t.packets, packet)
+		} else {
+			t.packets = append([]*TCPPacket{packet}, t.packets...)
+		}
+
+		t.End = time.Now()
+	}
+}
+
+// isMultipart returns true if message contains from multiple tcp packets
+func (t *TCPMessage) IsMultipart() bool {
+	if len(t.packets) > 1 {
+		return true
 	}
 
-	// Reset message timeout timer
-	t.timer.Reset(MsgExpire)
+	payload := t.packets[0].Data
+
+	if len(payload) < 4 {
+		return true
+	}
+
+	m := payload[:4]
+
+	if t.IsIncoming {
+		// If one GET, OPTIONS, or HEAD request
+		if bytes.Equal(m, []byte("GET ")) || bytes.Equal(m, []byte("OPTI")) || bytes.Equal(m, []byte("HEAD")) {
+			return false
+		} else {
+			// Sometimes header comes after the body :(
+			if bytes.Equal(m, []byte("POST")) || bytes.Equal(m, []byte("PUT ")) || bytes.Equal(m, []byte("PATC")) {
+				if length := proto.Header(payload, []byte("Content-Length")); len(length) > 0 {
+					l, _ := strconv.Atoi(string(length))
+
+					// If content-length equal current body length
+					if l > 0 && l == t.Size() {
+						return false
+					}
+				}
+			}
+		}
+	} else {
+		if length := proto.Header(payload, []byte("Content-Length")); len(length) > 0 {
+			if length[0] == '0' {
+				return false
+			}
+
+			l, _ := strconv.Atoi(string(length))
+
+			// If content-length equal current body length
+			if l > 0 && l == t.Size() {
+				return false
+			}
+		}
+	}
+
+	return true
+}
+
+func (t *TCPMessage) UUID() []byte {
+	var key []byte
+
+	if t.IsIncoming {
+		key = strconv.AppendInt(key, t.Start.UnixNano(), 10)
+		key = strconv.AppendUint(key, uint64(t.Ack), 10)
+	} else {
+		key = strconv.AppendInt(key, t.RequestStart.UnixNano(), 10)
+		key = strconv.AppendUint(key, uint64(t.RequestAck), 10)
+	}
+
+	uuid := make([]byte, 40)
+	sha := sha1.Sum(key)
+	hex.Encode(uuid, sha[:20])
+
+	return uuid
 }

raw_socket_listener/tcp_packet.go

@@ -50,7 +50,6 @@ func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) {
 // Parse TCP Packet, inspired by: https://github.com/miekg/pcap/blob/master/packet.go
 func (t *TCPPacket) Parse() {
 	t.ParseBasic()
-	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])
@@ -59,6 +58,7 @@ func (t *TCPPacket) Parse() {
 
 // ParseBasic set of fields
 func (t *TCPPacket) ParseBasic() {
+	t.DestPort = binary.BigEndian.Uint16(t.Data[2:4])
 	t.SrcPort = binary.BigEndian.Uint16(t.Data[0:2])
 	t.Seq = binary.BigEndian.Uint32(t.Data[4:8])
 	t.Ack = binary.BigEndian.Uint32(t.Data[8:12])
@@ -69,6 +69,11 @@ func (t *TCPPacket) ParseBasic() {
 
 // String output for a TCP Packet
 func (t *TCPPacket) String() string {
+	maxLen := len(t.Data)
+	if maxLen > 500 {
+		maxLen = 500
+	}
+
 	return strings.Join([]string{
 		"Source port: " + strconv.Itoa(int(t.SrcPort)),
 		"Dest port:" + strconv.Itoa(int(t.DestPort)),
@@ -90,7 +95,7 @@ func (t *TCPPacket) String() string {
 		"Checksum:" + strconv.Itoa(int(t.Checksum)),
 
 		"Data size:" + strconv.Itoa(len(t.Data)),
-		"Data:" + string(t.Data),
+		"Data:" + string(t.Data[:maxLen]),
 	}, "\n")
 }
 

settings.go

@@ -3,14 +3,12 @@ package main
 import (
 	"flag"
 	"fmt"
-	"log"
 	"os"
+	"sync"
+	"time"
 )
 
-const (
-	// VERSION specifies Gor current version
-	VERSION = "0.9.8"
-)
+var VERSION string
 
 // MultiOption allows to specify multiple flags with same name and collects all values into array
 type MultiOption []string
@@ -45,6 +43,8 @@ type AppSettings struct {
 
 	inputRAW MultiOption
 
+	middleware string
+
 	inputHTTP  MultiOption
 	outputHTTP MultiOption
 
@@ -65,7 +65,7 @@ func init() {
 	flag.Usage = usage
 
 	flag.BoolVar(&Settings.verbose, "verbose", false, "Turn on more verbose output")
-	flag.BoolVar(&Settings.debug, "debug", false, "Turn on debug output, shows all itercepted traffic. Works only when with `verbose` flag")
+	flag.BoolVar(&Settings.debug, "debug", false, "Turn on debug output, shows all intercepted traffic. Works only when with `verbose` flag")
 	flag.BoolVar(&Settings.stats, "stats", false, "Turn on queue stats 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.")
@@ -82,6 +82,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.StringVar(&Settings.middleware, "middleware", "", "Used for modifying traffic using external command")
+
 	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")
@@ -102,10 +104,10 @@ func init() {
 	flag.Var(&Settings.modifierConfig.methods, "http-allow-method", "Whitelist of HTTP methods to replay. Anything else will be dropped:\n\tgor --input-raw :8080 --output-http staging.com --http-allow-method GET --http-allow-method OPTIONS")
 	flag.Var(&Settings.modifierConfig.methods, "output-http-method", "WARNING: `--output-http-method` DEPRECATED, use `--http-allow-method` instead")
 
-	flag.Var(&Settings.modifierConfig.urlRegexp, "http-allow-url", "A regexp to match requests against. Filter get matched agains full url with domain. Anything else will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-allow-url ^www.")
+	flag.Var(&Settings.modifierConfig.urlRegexp, "http-allow-url", "A regexp to match requests against. Filter get matched against full url with domain. Anything else will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-allow-url ^www.")
 	flag.Var(&Settings.modifierConfig.urlRegexp, "output-http-url-regexp", "WARNING: `--output-http-url-regexp` DEPRECATED, use `--http-allow-url` instead")
 
-	flag.Var(&Settings.modifierConfig.urlNegativeRegexp, "http-diallow-url", "A regexp to match requests against. Filter get matched agains full url with domain. Anything else will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-disallow-url ^www.")
+	flag.Var(&Settings.modifierConfig.urlNegativeRegexp, "http-disallow-url", "A regexp to match requests against. Filter get matched against full url with domain. Anything else will be forwarded:\n\t gor --input-raw :8080 --output-http staging.com --http-disallow-url ^www.")
 
 	flag.Var(&Settings.modifierConfig.urlRewrite, "http-rewrite-url", "Rewrite the request url based on a mapping:\n\tgor --input-raw :8080 --output-http staging.com --http-rewrite-url /v1/user/([^\\/]+)/ping:/v2/user/$1/ping")
 	flag.Var(&Settings.modifierConfig.urlRewrite, "output-http-rewrite-url", "WARNING: `--output-http-rewrite-url` DEPRECATED, use `--http-rewrite-url` instead")
@@ -121,10 +123,19 @@ func init() {
 	flag.Var(&Settings.modifierConfig.paramHashFilters, "http-param-limiter", "Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific GET param:\n\t gor --input-raw :8080 --output-http staging.com --http-param-limiter user_id:25%")
 }
 
+var previousDebugTime int64
+var debugMutex sync.Mutex
+
 // Debug gets called only if --verbose flag specified
 func Debug(args ...interface{}) {
 	if Settings.verbose {
-		fmt.Printf("[DEBUG][PID %d] ", os.Getpid())
-		log.Println(args...)
+		debugMutex.Lock()
+		now := time.Now()
+		diff := float64(now.UnixNano()-previousDebugTime) / 1000000
+		previousDebugTime = now.UnixNano()
+		debugMutex.Unlock()
+
+		fmt.Printf("[DEBUG][PID %d][%d][%fms] ", os.Getpid(), now.UnixNano(), diff)
+		fmt.Println(args...)
 	}
 }

test_input.go

@@ -3,6 +3,7 @@ package main
 import (
 	"crypto/rand"
 	"encoding/base64"
+	"time"
 )
 
 // TestInput used for testing purpose, it allows emitting requests on demand
@@ -20,9 +21,12 @@ func NewTestInput() (i *TestInput) {
 
 func (i *TestInput) Read(data []byte) (int, error) {
 	buf := <-i.data
-	copy(data, buf)
 
-	return len(buf), nil
+	header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano())
+	copy(data[0:len(header)], header)
+	copy(data[len(header):], buf)
+
+	return len(buf) + len(header), nil
 }
 
 // EmitGET emits GET request without headers