add id in raw message stats

the current UUID generator runs at:
```
BenchmarkMessageUUID 	13599342	        87.7 ns/op	      16 B/op	       0 allocs/op
```
 with the former version that was running on
```
BenchmarkMessageUUID 	 2818203	       427 ns/op	      96 B/op	       2 allocs/op
```

it fixes #842 #851

.travis.yml

@@ -3,7 +3,7 @@ dist: focal
 language: go
 go: 1.14
 
-script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.14)' && go test ./... -v -timeout 120s"
+script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.14)' && go test ./... -v -timeout 120s -race"
 
 before_install:
   - sudo apt-get install libpcap-dev -y

Makefile

@@ -130,7 +130,7 @@ bash:
 
 
 FPMCOMMON= \
-    --name goreplay \
+    --name gor \
     --description "GoReplay is an open-source network monitoring tool which can record your live traffic, and use it for shadowing, load testing, monitoring and detailed analysis." \
     -v $(VERSION) \
     --vendor "Leonid Bugaev" \

emitter.go

@@ -9,12 +9,21 @@ import (
 	"time"
 
 	"github.com/buger/goreplay/byteutils"
+	"github.com/buger/goreplay/stat"
 )
 
+// EmitterConfig configurations associated with emitters
+type EmitterConfig struct {
+	Stats                bool
+	SplitOutput          bool
+	RecognizeTCPSessions bool
+}
+
 // Emitter represents an abject to manage plugins communication
 type Emitter struct {
 	sync.WaitGroup
 	plugins *InOutPlugins
+	st      *stat.Writer
 }
 
 // NewEmitter creates and initializes new Emitter object.
@@ -28,6 +37,10 @@ func (e *Emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
 		Settings.CopyBufferSize = 5 << 20
 	}
 	e.plugins = plugins
+	if Settings.EmitterConfig.Stats {
+		e.st = initializeStatWriter("meta.csv", "type: GOR_STAT\ntitle: Messages Meta")
+		statsWriteRecord(e.st, []string{"type", "id", "timestamp", "latency"}, "[EMITTER]")
+	}
 
 	if middlewareCmd != "" {
 		middleware := NewMiddleware(middlewareCmd)
@@ -41,8 +54,8 @@ func (e *Emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
 		e.Add(1)
 		go func() {
 			defer e.Done()
-			if err := CopyMulty(middleware, plugins.Outputs...); err != nil {
-				Debug(2, fmt.Sprintf("[EMITTER] error during copy: %q", err))
+			if err := e.CopyMulty(middleware, plugins.Outputs...); err != nil {
+				Debug(1, fmt.Sprintf("[EMITTER] error during copy: %q", err))
 			}
 		}()
 	} else {
@@ -50,8 +63,8 @@ func (e *Emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
 			e.Add(1)
 			go func(in PluginReader) {
 				defer e.Done()
-				if err := CopyMulty(in, plugins.Outputs...); err != nil {
-					Debug(2, fmt.Sprintf("[EMITTER] error during copy: %q", err))
+				if err := e.CopyMulty(in, plugins.Outputs...); err != nil {
+					Debug(1, fmt.Sprintf("[EMITTER] error during copy: %q", err))
 				}
 			}(in)
 		}
@@ -70,10 +83,13 @@ func (e *Emitter) Close() {
 		e.Wait()
 	}
 	e.plugins.All = nil // avoid Close to make changes again
+	if e.st != nil {
+		e.st.Close()
+	}
 }
 
 // CopyMulty copies from 1 reader to multiple writers
-func CopyMulty(src PluginReader, writers ...PluginWriter) error {
+func (e *Emitter) CopyMulty(src PluginReader, writers ...PluginWriter) error {
 	wIndex := 0
 	modifier := NewHTTPModifier(&Settings.ModifierConfig)
 	filteredRequests := make(map[string]int64)
@@ -97,13 +113,16 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 				Debug(2, fmt.Sprintf("[EMITTER] Found malformed record %q from %q", msg.Meta, src))
 				continue
 			}
+			if e.st != nil {
+				statsWriteRecord(e.st, splitUnsafe(meta), "[EMITTER]")
+			}
 			requestID := byteutils.SliceToString(meta[1])
 			// start a subroutine only when necessary
 			if Settings.Verbose >= 3 {
 				Debug(3, "[EMITTER] input: ", byteutils.SliceToString(msg.Meta[:len(msg.Meta)-1]), " from: ", src)
 			}
 			if modifier != nil {
-				Debug(3, "[EMITTER] modifier:", requestID, "from:", src)
+				Debug(3, "[EMITTER] modifier: ", requestID, " from: ", src)
 				if isRequestPayload(msg.Meta) {
 					msg.Data = modifier.Rewrite(msg.Data)
 					// If modifier tells to skip request
@@ -175,3 +194,11 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 		}
 	}
 }
+
+func splitUnsafe(data [][]byte) []string {
+	s := make([]string, len(data), len(data))
+	for i := range data {
+		s[i] = byteutils.SliceToString(data[i])
+	}
+	return s
+}

gor_stat.go

@@ -1,64 +0,0 @@
-package main
-
-import (
-	"runtime"
-	"strconv"
-	"time"
-)
-
-type GorStat struct {
-	statName string
-	rateMs   int
-	latest   int
-	mean     int
-	max      int
-	count    int
-}
-
-func NewGorStat(statName string, rateMs int) (s *GorStat) {
-	s = new(GorStat)
-	s.statName = statName
-	s.rateMs = rateMs
-	s.latest = 0
-	s.mean = 0
-	s.max = 0
-	s.count = 0
-
-	if Settings.Stats {
-		go s.reportStats()
-	}
-	return
-}
-
-func (s *GorStat) Write(latest int) {
-	if Settings.Stats {
-		if latest > s.max {
-			s.max = latest
-		}
-		if latest != 0 {
-			s.mean = ((s.mean * s.count) + latest) / (s.count + 1)
-		}
-		s.latest = latest
-		s.count = s.count + 1
-	}
-}
-
-func (s *GorStat) Reset() {
-	s.latest = 0
-	s.max = 0
-	s.mean = 0
-	s.count = 0
-}
-
-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/(s.rateMs/1000.0)) + "," + strconv.Itoa(runtime.NumGoroutine())
-}
-
-func (s *GorStat) reportStats() {
-	Debug(0, "\n", s.statName+":latest,mean,max,count,count/second,gcount")
-	for {
-		Debug(0, "\n", s)
-		s.Reset()
-		time.Sleep(time.Duration(s.rateMs) * time.Millisecond)
-	}
-}

input_raw.go

@@ -12,6 +12,7 @@ import (
 	"github.com/buger/goreplay/capture"
 	"github.com/buger/goreplay/proto"
 	"github.com/buger/goreplay/size"
+	"github.com/buger/goreplay/stat"
 	"github.com/buger/goreplay/tcp"
 )
 
@@ -68,7 +69,7 @@ type RAWInputConfig struct {
 type RAWInput struct {
 	sync.Mutex
 	RAWInputConfig
-	messageStats   []tcp.Stats
+	st             *stat.Writer
 	listener       *capture.Listener
 	message        chan *tcp.Message
 	cancelListener context.CancelFunc
@@ -96,6 +97,13 @@ func NewRAWInput(address string, config RAWInputConfig) (i *RAWInput) {
 	}
 	i.host = host
 	i.port = uint16(port)
+	if i.Stats {
+		i.st = initializeStatWriter("input_raw_message.csv", "type: GOR_STAT\ntitle: raw messages info")
+		statsWriteRecord(i.st, []string{
+			"length", "lostdata", "source", "destination", "ipversion",
+			"timedout", "truncated", "id",
+		}, "[INPUT-RAW]")
+	}
 
 	i.listen(address)
 
@@ -108,30 +116,28 @@ func (i *RAWInput) PluginRead() (*Message, error) {
 	var msg Message
 	select {
 	case <-i.quit:
+		if i.st != nil {
+			i.st.Close()
+			i.st = nil
+		}
 		return nil, ErrorStopped
 	case msgTCP = <-i.message:
 		msg.Data = msgTCP.Data()
 	}
 	var msgType byte = ResponsePayload
+	timestamp := msgTCP.Start.UnixNano()
+	latency := msgTCP.End.UnixNano() - timestamp
 	if msgTCP.IsIncoming {
 		msgType = RequestPayload
 		if i.RealIPHeader != "" {
 			msg.Data = proto.SetHeader(msg.Data, []byte(i.RealIPHeader), []byte(msgTCP.SrcAddr))
 		}
 	}
-	msg.Meta = payloadHeader(msgType, msgTCP.UUID(), msgTCP.Start.UnixNano(), msgTCP.End.UnixNano()-msgTCP.Start.UnixNano())
+	id := msgTCP.UUID()
+	msg.Meta = payloadHeader(msgType, id, timestamp, latency)
 
-	// to be removed....
-	if msgTCP.Truncated {
-		Debug(2, "[INPUT-RAW] message truncated, increase copy-buffer-size")
-	}
-	// to be removed...
-	if msgTCP.TimedOut {
-		Debug(2, "[INPUT-RAW] message timeout reached, increase input-raw-expire")
-	}
-	if i.Stats {
-		stat := msgTCP.Stats
-		go i.addStats(stat)
+	if i.Stats && i.st != nil {
+		statsWriteRecord(i.st, getStats(msgTCP.Stats, id), "[INPUT-RAW]")
 	}
 	msgTCP = nil
 	return &msg, nil
@@ -149,8 +155,10 @@ func (i *RAWInput) listen(address string) {
 		log.Fatal(err)
 	}
 	pool := tcp.NewMessagePool(i.CopyBufferSize, i.Expire, Debug, i.handler)
-	pool.End = endHint
-	pool.Start = startHint
+	pool.MatchUUID(i.TrackResponse)
+	if i.Protocol == ProtocolHTTP {
+		pool.Start = http1StartHint
+	}
 	var ctx context.Context
 	ctx, i.cancelListener = context.WithCancel(context.Background())
 	errCh := i.listener.ListenBackground(ctx, pool.Handler)
@@ -171,13 +179,26 @@ func (i *RAWInput) String() string {
 }
 
 // GetStats returns the stats so far and reset the stats
-func (i *RAWInput) GetStats() []tcp.Stats {
-	i.Lock()
-	defer func() {
-		i.messageStats = []tcp.Stats{}
-		i.Unlock()
-	}()
-	return i.messageStats
+func getStats(stat tcp.Stats, id []byte) []string {
+	var s [8]string
+
+	s[0] = strconv.Itoa(stat.Length)    // length
+	s[1] = strconv.Itoa(stat.LostData)  // lostdata
+	s[2] = stat.SrcAddr                 // source
+	s[3] = stat.DstAddr                 // destination
+	s[4] = string(stat.IPversion + '0') // ipversion
+	if stat.TimedOut {                  // timedout
+		s[5] = "true"
+	} else {
+		s[5] = "false"
+	}
+	if stat.Truncated { // truncated
+		s[6] = "true"
+	} else {
+		s[6] = "false"
+	}
+	s[7] = string(id)
+	return s[:]
 }
 
 // Close closes the input raw listener
@@ -187,23 +208,10 @@ func (i *RAWInput) Close() error {
 	return nil
 }
 
-func (i *RAWInput) addStats(mStats tcp.Stats) {
-	i.Lock()
-	if len(i.messageStats) >= 10000 {
-		i.messageStats = []tcp.Stats{}
-	}
-	i.messageStats = append(i.messageStats, mStats)
-	i.Unlock()
-}
-
-func startHint(pckt *tcp.Packet) (isIncoming, isOutgoing bool) {
+func http1StartHint(pckt *tcp.Packet) (isIncoming, isOutgoing bool) {
 	isIncoming = proto.HasRequestTitle(pckt.Payload)
 	if isIncoming {
 		return
 	}
 	return false, proto.HasResponseTitle(pckt.Payload)
 }
-
-func endHint(m *tcp.Message) bool {
-	return proto.HasFullPayload(m.Data())
-}

middleware_test.go

@@ -1,225 +1,226 @@
 package main
 
-// import (
-// 	"bytes"
-// 	"crypto/rand"
-// 	"encoding/hex"
-// 	"io"
-// 	"net/http"
-// 	"net/http/httptest"
-// 	"strings"
-// 	"sync"
-// 	"testing"
-// 	"time"
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/hex"
+	"net/http"
+	"net/http/httptest"
+	"net/http/httputil"
+	"strings"
+	"sync"
+	"testing"
+	"time"
 
-// 	"github.com/buger/goreplay/capture"
-// 	"github.com/buger/goreplay/proto"
-// )
+	"github.com/buger/goreplay/capture"
+	"github.com/buger/goreplay/proto"
+)
 
-// type fakeServiceCb func(string, int, []byte)
+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
+// 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 {
+	activeTokens := make([]string, 0)
+	var mu sync.Mutex
 
-// 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-// 		mu.Lock()
-// 		defer mu.Unlock()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		mu.Lock()
+		defer mu.Unlock()
 
-// 		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)
+		switch req.URL.Path {
+		case "/token":
+			// Generate random token
+			tokenLength := 10
+			buf := make([]byte, tokenLength)
+			rand.Read(buf)
+			token := hex.EncodeToString(buf)
+			activeTokens = append(activeTokens, token)
 
-// 			w.Write([]byte(token))
+			w.Write([]byte(token))
 
-// 			cb(req.URL.Path, 200, []byte(token))
-// 		case "/secure":
-// 			token := req.URL.Query().Get("token")
-// 			token_found := false
+			cb(req.URL.Path, 200, []byte(token))
+		case "/secure":
+			token := req.URL.Query().Get("token")
+			tokenFound := false
 
-// 			for _, t := range active_tokens {
-// 				if t == token {
-// 					token_found = true
-// 					break
-// 				}
-// 			}
+			for _, t := range activeTokens {
+				if t == token {
+					tokenFound = 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))
-// 			}
-// 		}
+			if tokenFound {
+				w.WriteHeader(http.StatusAccepted)
+				cb(req.URL.Path, 202, nil)
+			} else {
+				w.WriteHeader(http.StatusForbidden)
+				cb(req.URL.Path, 403, nil)
+			}
+		}
 
-// 		wg.Done()
-// 	}))
+		wg.Done()
+	}))
 
-// 	return server
-// }
+	return server
+}
 
-// func TestFakeSecureService(t *testing.T) {
-// 	var resp, token []byte
+func TestFakeSecureService(t *testing.T) {
+	var resp, token []byte
 
-// 	wg := new(sync.WaitGroup)
+	wg := new(sync.WaitGroup)
 
-// 	server := NewFakeSecureService(wg, func(path string, status int, resp []byte) {
-// 	})
-// 	defer server.Close()
+	server := NewFakeSecureService(wg, func(path string, status int, resp []byte) {
+	})
+	defer server.Close()
 
-// 	wg.Add(3)
+	wg.Add(3)
 
-// 	client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true})
-// 	resp, _ = client.Get("/token")
-// 	token = proto.Body(resp)
+	client := NewHTTPClient(&HTTPOutputConfig{}).Client
+	rep, _ := client.Get(server.URL + "/token")
+	resp, _ = httputil.DumpResponse(rep, true)
+	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)))
-// 	}
+	// Right token
+	rep, _ = client.Get(server.URL + "/secure?token=" + string(token))
+	resp, _ = httputil.DumpResponse(rep, true)
+	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)))
-// 	}
+	// Wrong tokens forbidden
+	rep, _ = client.Get(server.URL + "/secure?token=wrong")
+	resp, _ = httputil.DumpResponse(rep, true)
+	if !bytes.Equal(proto.Status(resp), []byte("403")) {
+		t.Error("Wrong token should returns status 403:", string(proto.Status(resp)))
+	}
 
-// 	wg.Wait()
-// }
+	wg.Wait()
+}
 
-// func TestEchoMiddleware(t *testing.T) {
-// 	wg := new(sync.WaitGroup)
+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()
+	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()
+	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)
+	// Catch traffic from one service
+	fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
+	conf := RAWInputConfig{
+		Engine:        capture.EnginePcap,
+		Expire:        testRawExpire,
+		Protocol:      ProtocolHTTP,
+		TrackResponse: true,
+	}
+	input := NewRAWInput(fromAddr, conf)
 
-// 	// Catch traffic from one service
-// 	fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
-// 	conf := RAWInputConfig{
-// 		engine:        capture.EnginePcap,
-// 		expire:        testRawExpire,
-// 		protocol:      ProtocolHTTP,
-// 		trackResponse: true,
-// 	}
-// 	input := NewRAWInput(fromAddr, conf)
+	// And redirect to another
+	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{})
 
-// 	// And redirect to another
-// 	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: false})
+	plugins := &InOutPlugins{
+		Inputs:  []PluginReader{input},
+		Outputs: []PluginWriter{output},
+	}
+	plugins.All = append(plugins.All, input, output)
 
-// 	plugins := &InOutPlugins{
-// 		Inputs:  []io.Reader{input},
-// 		Outputs: []io.Writer{output},
-// 	}
-// 	plugins.All = append(plugins.All, input, output)
+	// Start Gor
+	emitter := NewEmitter()
+	emitter.Start(plugins, "echo -n && GOR_TEST=true && ./examples/middleware/echo.sh")
 
-// 	// Start Gor
-// 	emitter := NewEmitter(quit)
-// 	go emitter.Start(plugins, "echo -n && GOR_TEST=true && ./examples/middleware/echo.sh")
+	// Wait till middleware initialization
+	time.Sleep(100 * time.Millisecond)
 
-// 	// Wait till middleware initialization
-// 	time.Sleep(100 * time.Millisecond)
+	// Should receive 2 requests from original + 2 from replayed
+	client := NewHTTPClient(output.(*HTTPOutput).config).Client
 
-// 	// Should receive 2 requests from original + 2 from replayed
-// 	client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: false})
+	for i := 0; i < 10; i++ {
+		wg.Add(2)
+		// Request should be echoed
+		client.Get(to.URL + "/a")
+		time.Sleep(5 * time.Millisecond)
+		client.Get(to.URL + "/b")
+		time.Sleep(5 * time.Millisecond)
+	}
 
-// 	for i := 0; i < 10; i++ {
-// 		wg.Add(2)
-// 		// Request should be echoed
-// 		client.Get("/a")
-// 		time.Sleep(5 * time.Millisecond)
-// 		client.Get("/b")
-// 		time.Sleep(5 * time.Millisecond)
-// 	}
+	wg.Wait()
+	emitter.Close()
+}
 
-// 	wg.Wait()
-// 	emitter.Close()
-// }
+func TestTokenMiddleware(t *testing.T) {
+	var resp, token []byte
 
-// func TestTokenMiddleware(t *testing.T) {
-// 	var resp, token []byte
+	wg := new(sync.WaitGroup)
 
-// 	wg := new(sync.WaitGroup)
+	from := NewFakeSecureService(wg, func(path string, status int, tok []byte) {
+		time.Sleep(10 * time.Millisecond)
+	})
+	defer from.Close()
 
-// 	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")
+			}
+		}
 
-// 	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()
 
-// 		time.Sleep(10 * time.Millisecond)
-// 	})
-// 	defer to.Close()
+	Settings.Middleware = "echo -n && GOR_TEST=true && go run ./examples/middleware/token_modifier.go"
 
-// 	quit := make(chan int)
+	fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
+	conf := RAWInputConfig{
+		Engine:        capture.EnginePcap,
+		Expire:        testRawExpire,
+		Protocol:      ProtocolHTTP,
+		TrackResponse: true,
+	}
+	// Catch traffic from one service
+	input := NewRAWInput(fromAddr, conf)
 
-// 	Settings.middleware = "echo -n && GOR_TEST=true && go run ./examples/middleware/token_modifier.go"
+	// And redirect to another
+	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{})
 
-// 	fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
-// 	conf := RAWInputConfig{
-// 		engine:        capture.EnginePcap,
-// 		expire:        testRawExpire,
-// 		protocol:      ProtocolHTTP,
-// 		trackResponse: true,
-// 	}
-// 	// Catch traffic from one service
-// 	input := NewRAWInput(fromAddr, conf)
+	plugins := &InOutPlugins{
+		Inputs:  []PluginReader{input},
+		Outputs: []PluginWriter{output},
+	}
+	plugins.All = append(plugins.All, input, output)
 
-// 	// And redirect to another
-// 	output := NewHTTPOutput(to.URL, &HTTPOutputConfig{Debug: true})
+	// Start Gor
+	emitter := NewEmitter()
+	emitter.Start(plugins, Settings.Middleware)
 
-// 	plugins := &InOutPlugins{
-// 		Inputs:  []io.Reader{input},
-// 		Outputs: []io.Writer{output},
-// 	}
-// 	plugins.All = append(plugins.All, input, output)
+	// Should receive 2 requests from original + 2 from replayed
+	wg.Add(2)
 
-// 	// Start Gor
-// 	emitter := NewEmitter(quit)
-// 	go emitter.Start(plugins, Settings.middleware)
+	client := NewHTTPClient(&HTTPOutputConfig{}).Client
 
-// 	// Should receive 2 requests from original + 2 from replayed
-// 	wg.Add(2)
+	// Sending traffic to original service
+	rep, _ := client.Get(to.URL + "/token")
+	resp, _ = httputil.DumpResponse(rep, true)
+	token = proto.Body(resp)
 
-// 	client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: true})
+	rep, _ = client.Get(to.URL + "/secure?token=" + string(token))
+	resp, _ = httputil.DumpResponse(rep, true)
+	if !bytes.Equal(proto.Status(resp), []byte("202")) {
+		t.Error("Valid token should return 202:", proto.Status(resp))
+	}
 
-// 	// Sending traffic to original service
-// 	resp, _ = client.Get("/token")
-// 	token = proto.Body(resp)
-
-// 	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()
-// 	emitter.Close()
-// 	Settings.middleware = ""
-// }
+	wg.Wait()
+	emitter.Close()
+	Settings.Middleware = ""
+}

output_binary.go

@@ -30,7 +30,6 @@ type BinaryOutput struct {
 	needWorker    chan int
 	quit          chan struct{}
 	config        *BinaryOutputConfig
-	queueStats    *GorStat
 }
 
 // NewBinaryOutput constructor for BinaryOutput

output_http.go

@@ -32,12 +32,10 @@ type response struct {
 // HTTPOutputConfig struct for holding http output configuration
 type HTTPOutputConfig struct {
 	TrackResponses bool          `json:"output-http-track-response"`
-	Stats          bool          `json:"output-http-stats"`
 	OriginalHost   bool          `json:"output-http-original-host"`
 	RedirectLimit  int           `json:"output-http-redirect-limit"`
 	WorkersMin     int           `json:"output-http-workers-min"`
 	WorkersMax     int           `json:"output-http-workers"`
-	StatsMs        int           `json:"output-http-stats-ms"`
 	QueueLen       int           `json:"output-http-queue-len"`
 	ElasticSearch  string        `json:"output-http-elasticsearch"`
 	Timeout        time.Duration `json:"output-http-timeout"`
@@ -54,7 +52,6 @@ type HTTPOutputConfig struct {
 type HTTPOutput struct {
 	activeWorkers int32
 	config        *HTTPOutputConfig
-	queueStats    *GorStat
 	elasticSearch *ESPlugin
 	client        *HTTPClient
 	stopWorker    chan struct{}
@@ -105,9 +102,6 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) PluginReadWriter {
 	}
 	o.config = config
 	o.stop = make(chan bool)
-	if o.config.Stats {
-		o.queueStats = NewGorStat("output_http", o.config.StatsMs)
-	}
 
 	o.queue = make(chan *Message, o.config.QueueLen)
 	if o.config.TrackResponses {
@@ -179,9 +173,6 @@ func (o *HTTPOutput) PluginWrite(msg *Message) (n int, err error) {
 	case o.queue <- msg:
 	}
 
-	if o.config.Stats {
-		o.queueStats.Write(len(o.queue))
-	}
 	if len(o.queue) > 0 {
 		// try to start a new worker to serve
 		if atomic.LoadInt32(&o.activeWorkers) < int32(o.config.WorkersMax) {

output_tcp.go

@@ -15,9 +15,9 @@ type TCPOutput struct {
 	address     string
 	limit       int
 	buf         []chan *Message
-	bufStats    *GorStat
 	config      *TCPOutputConfig
 	workerIndex uint32
+	stop        chan struct{}
 }
 
 // TCPOutputConfig tcp output configuration
@@ -35,10 +35,7 @@ func NewTCPOutput(address string, config *TCPOutputConfig) PluginWriter {
 
 	o.address = address
 	o.config = config
-
-	if Settings.OutputTCPStats {
-		o.bufStats = NewGorStat("output_tcp", 5000)
-	}
+	o.stop = make(chan struct{})
 
 	// create X buffers and send the buffer index to the worker
 	o.buf = make([]chan *Message, o.config.Workers)
@@ -72,7 +69,12 @@ func (o *TCPOutput) worker(bufferIndex int) {
 	defer conn.Close()
 
 	for {
-		msg := <-o.buf[bufferIndex]
+		var msg *Message
+		select {
+		case <-o.stop:
+			return
+		case msg = <-o.buf[bufferIndex]:
+		}
 		if _, err = conn.Write(msg.Meta); err == nil {
 			if _, err = conn.Write(msg.Data); err == nil {
 				_, err = conn.Write(payloadSeparatorAsBytes)
@@ -109,10 +111,6 @@ func (o *TCPOutput) PluginWrite(msg *Message) (n int, err error) {
 	bufferIndex := o.getBufferIndex(msg.Data)
 	o.buf[bufferIndex] <- msg
 
-	if Settings.OutputTCPStats {
-		o.bufStats.Write(len(o.buf[bufferIndex]))
-	}
-
 	return len(msg.Data) + len(msg.Meta), nil
 }
 
@@ -129,3 +127,9 @@ func (o *TCPOutput) connect(address string) (conn net.Conn, err error) {
 func (o *TCPOutput) String() string {
 	return fmt.Sprintf("TCP output %s, limit: %d", o.address, o.limit)
 }
+
+// Close closes the data channel so that data
+func (o *TCPOutput) Close() error {
+	close(o.stop)
+	return nil
+}

settings.go

@@ -3,7 +3,10 @@ package main
 import (
 	"flag"
 	"fmt"
+	"log"
 	"os"
+	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 )
@@ -27,12 +30,10 @@ func (h *MultiOption) Set(value string) error {
 // AppSettings is the struct of main configuration
 type AppSettings struct {
 	Verbose   int           `json:"verbose"`
-	Stats     bool          `json:"stats"`
 	ExitAfter time.Duration `json:"exit-after"`
 
-	SplitOutput          bool   `json:"split-output"`
-	RecognizeTCPSessions bool   `json:"recognize-tcp-sessions"`
-	Pprof                string `json:"http-pprof"`
+	EmitterConfig
+	Pprof string `json:"http-pprof"`
 
 	InputDummy   MultiOption `json:"input-dummy"`
 	OutputDummy  MultiOption
@@ -43,7 +44,6 @@ type AppSettings struct {
 	InputTCPConfig  TCPInputConfig
 	OutputTCP       MultiOption `json:"output-tcp"`
 	OutputTCPConfig TCPOutputConfig
-	OutputTCPStats  bool `json:"output-tcp-stats"`
 
 	InputFile        MultiOption `json:"input-file"`
 	InputFileLoop    bool        `json:"input-file-loop"`
@@ -80,11 +80,10 @@ func usage() {
 	os.Exit(2)
 }
 
-func init() {
+func initFlags() {
 	flag.Usage = usage
 	flag.StringVar(&Settings.Pprof, "http-pprof", "", "Enable profiling. Starts  http server on specified port, exposing special /debug/pprof endpoint. Example: `:8181`")
 	flag.IntVar(&Settings.Verbose, "verbose", 0, "set the level of verbosity, if greater than zero then it will turn on debug output")
-	flag.BoolVar(&Settings.Stats, "stats", false, "Turn on queue stats output")
 
 	if DEMO == "" {
 		flag.DurationVar(&Settings.ExitAfter, "exit-after", 0, "exit after specified duration")
@@ -92,6 +91,7 @@ func init() {
 		Settings.ExitAfter = 5 * time.Minute
 	}
 
+	flag.BoolVar(&Settings.EmitterConfig.Stats, "meta-stats", false, "Turn on stats for meta data.")
 	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.BoolVar(&Settings.RecognizeTCPSessions, "recognize-tcp-sessions", false, "[PRO] If turned on http output will create separate worker for each TCP session. Splitting output will session based as well.")
 
@@ -109,7 +109,6 @@ func init() {
 	flag.BoolVar(&Settings.OutputTCPConfig.SkipVerify, "output-tcp-skip-verify", false, "Don't verify hostname on TLS secure connection.")
 	flag.BoolVar(&Settings.OutputTCPConfig.Sticky, "output-tcp-sticky", false, "Use Sticky connection. Request/Response with same ID will be sent to the same connection.")
 	flag.IntVar(&Settings.OutputTCPConfig.Workers, "output-tcp-workers", 10, "Number of parallel tcp connections, default is 10")
-	flag.BoolVar(&Settings.OutputTCPStats, "output-tcp-stats", false, "Report TCP output queue stats to console every 5 seconds.")
 
 	flag.Var(&Settings.InputFile, "input-file", "Read requests from file: \n\tgor --input-file ./requests.gor --output-http staging.com")
 	flag.BoolVar(&Settings.InputFileLoop, "input-file-loop", false, "Loop input files, useful for performance testing.")
@@ -140,7 +139,7 @@ func init() {
 	flag.Var(&Settings.BufferSize, "input-raw-buffer-size", "Controls size of the OS buffer which holds packets until they dispatched. Default value depends by system: in Linux around 2MB. If you see big package drop, increase this value.")
 	flag.BoolVar(&Settings.Promiscuous, "input-raw-promisc", false, "enable promiscuous mode")
 	flag.BoolVar(&Settings.Monitor, "input-raw-monitor", false, "enable RF monitor mode")
-	flag.BoolVar(&Settings.Stats, "input-raw-stats", false, "enable stats generator on raw TCP messages")
+	flag.BoolVar(&Settings.RAWInputConfig.Stats, "input-raw-stats", false, "enable stats generator on raw TCP messages")
 
 	flag.StringVar(&Settings.Middleware, "middleware", "", "Used for modifying traffic using external command")
 
@@ -153,13 +152,9 @@ func init() {
 	flag.IntVar(&Settings.OutputHTTPConfig.QueueLen, "output-http-queue-len", 1000, "Number of requests that can be queued for output, if all workers are busy. default = 1000")
 	flag.BoolVar(&Settings.OutputHTTPConfig.SkipVerify, "output-http-skip-verify", false, "Don't verify hostname on TLS secure connection.")
 	flag.DurationVar(&Settings.OutputHTTPConfig.WorkerTimeout, "output-http-worker-timeout", 2*time.Second, "Duration to rollback idle workers.")
-
 	flag.IntVar(&Settings.OutputHTTPConfig.RedirectLimit, "output-http-redirects", 0, "Enable how often redirects should be followed.")
 	flag.DurationVar(&Settings.OutputHTTPConfig.Timeout, "output-http-timeout", 5*time.Second, "Specify HTTP request/response timeout. By default 5s. Example: --output-http-timeout 30s")
 	flag.BoolVar(&Settings.OutputHTTPConfig.TrackResponses, "output-http-track-response", false, "If turned on, HTTP output responses will be set to all outputs like stdout, file and etc.")
-
-	flag.BoolVar(&Settings.OutputHTTPConfig.Stats, "output-http-stats", false, "Report http output queue stats to console every N milliseconds. See output-http-stats-ms")
-	flag.IntVar(&Settings.OutputHTTPConfig.StatsMs, "output-http-stats-ms", 5000, "Report http output queue stats to console every N milliseconds. default: 5000")
 	flag.BoolVar(&Settings.OutputHTTPConfig.OriginalHost, "http-original-host", false, "Normally gor replaces the Host http header with the host supplied with --output-http.  This option disables that behavior, preserving the original Host header.")
 	flag.StringVar(&Settings.OutputHTTPConfig.ElasticSearch, "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'")
 	/* outputHTTPConfig */
@@ -234,3 +229,32 @@ func Debug(level int, args ...interface{}) {
 		fmt.Fprintln(os.Stderr, args...)
 	}
 }
+
+func initGorDir() {
+	var err error
+	p := os.Getenv("GORDIR")
+	if p == "" {
+		if p, err = os.UserHomeDir(); err != nil {
+			goto errLog
+		}
+		if !strings.Contains(p, "root") {
+			p = filepath.Join(p, ".gor")
+		} else {
+			p = ".gor"
+		}
+	}
+	if err = os.MkdirAll(p, os.ModeDir|os.ModePerm); err != nil {
+		goto errLog
+	}
+	if err = os.Setenv("GORDIR", p); err != nil {
+		goto errLog
+	}
+	return
+errLog:
+	log.Fatalf("init GORDIR error:%q", err)
+}
+
+func init() {
+	initGorDir()
+	initFlags()
+}

stat/csv/writer.go

@@ -0,0 +1,47 @@
+package csv
+
+import (
+	"encoding/csv"
+	"os"
+	"strings"
+)
+
+// Writer check encoding/csv.Writer
+type Writer struct {
+	*csv.Writer
+	f *os.File
+}
+
+// NewWriter returns an csv writer with file opened
+// error returned will be associated with opening the file
+func NewWriter(path string) (*Writer, error) {
+	var wr Writer
+	f, err := os.Create(path)
+	if err != nil {
+		return nil, err
+	}
+	wr.f = f
+	wr.Writer = csv.NewWriter(f)
+	return &wr, nil
+}
+
+// WriteHeader will write infos about this docs in form of comments
+// it uses "#" to indicate that this should be ignored by the reader
+func (wr *Writer) WriteHeader(hdr string) error {
+	_, err := wr.f.WriteString(formatHeader(hdr))
+	return err
+}
+
+// Close the underlying file
+func (wr *Writer) Close() error {
+	return wr.f.Close()
+}
+
+func formatHeader(hdr string) string {
+	if !strings.HasSuffix(hdr, "\n") {
+		hdr += "\n"
+	}
+	// add comments before every line
+	hdr = strings.ReplaceAll("#"+hdr, "\n", "\n#")
+	return hdr[:len(hdr)-1]
+}

stat/csv/writer_test.go

@@ -0,0 +1,23 @@
+package csv
+
+import (
+	"testing"
+)
+
+func TestFormatHeader(t *testing.T) {
+	m := map[string]string{
+		"hello":                       "#hello\n",
+		"":                            "#\n",
+		"hello\n":                     "#hello\n",
+		"hello\nhello\n":              "#hello\n#hello\n",
+		"hello\nhello":                "#hello\n#hello\n",
+		"hello\nhello\nhello":         "#hello\n#hello\n#hello\n",
+		"hello\nhello\nhello\n":       "#hello\n#hello\n#hello\n",
+		"hello\nhello\r\n\r\nhello\n": "#hello\n#hello\r\n#\r\n#hello\n",
+	}
+	for k, v := range m {
+		if vl := formatHeader(k); vl != v {
+			t.Errorf("expected %q to equal %q", vl, v)
+		}
+	}
+}

stat/writer.go

@@ -0,0 +1,124 @@
+package stat
+
+import (
+	"errors"
+	"sync"
+	"time"
+)
+
+// ErrNotImplemented returned when impls requested are not implemnted
+var ErrNotImplemented = errors.New("method not implemented")
+
+// RecordWriter represents a stat writer it should have all this method.
+// for rationale  behind this interface check encoding/csv or
+type RecordWriter interface {
+	Write(record []string) error
+	Flush()
+	Error() error
+}
+
+const flushSpan = 10
+
+// Writer represents a single instance of goreplay stats writer
+type Writer struct {
+	name    string
+	records [flushSpan][]string // a temporary stats holder
+	off     int
+	quit    chan struct{}
+	mu      sync.Mutex
+	w       RecordWriter
+	err     error
+}
+
+// NewWriter returns a new instance of stats associated with the writer "wr"
+// and "name" as the stats identity
+func NewWriter(w RecordWriter, name string) (wr *Writer) {
+	wr = new(Writer)
+	wr.name = name
+	wr.w = w
+	wr.quit = make(chan struct{})
+	return
+}
+
+// WriteHeader writes header of the documents to the underlying writer.
+// Header can be any kind of data like json unmarshal or file info.
+// underlying writer should sets a different way to represents the header.
+func (wr *Writer) WriteHeader(header string) error {
+	if wHdr, ok := wr.w.(interface{ WriteHeader(header string) error }); ok {
+		return wHdr.WriteHeader(header)
+	}
+	return ErrNotImplemented
+}
+
+// SetReader sets another source of records reader to use along with (*Writer).Write([]string)
+// rd function will be called every "rate" duration time
+func (wr *Writer) SetReader(rd func() []string, rate time.Duration) {
+	go func() {
+		timer := time.NewTicker(rate)
+		defer timer.Stop()
+		for {
+			select {
+			case <-wr.quit:
+				return
+			case <-timer.C:
+				err := wr.Write(rd())
+				if err != nil {
+					return
+				}
+			}
+		}
+	}()
+}
+
+func (wr *Writer) Write(record []string) error {
+	wr.mu.Lock()
+	defer wr.mu.Unlock()
+	if wr.err != nil {
+		return wr.err
+	}
+	if len(record) > 0 {
+		wr.records[wr.off] = record
+		wr.off++
+	}
+	return wr.flush(wr.off >= flushSpan)
+}
+
+func (wr *Writer) String() string {
+	return "stats: " + wr.name
+}
+
+// Error reports any error that has occurred during a previous Write or Flush.
+func (wr *Writer) Error() error {
+	wr.mu.Lock()
+	wr.mu.Unlock()
+	return wr.err
+}
+
+// Close closes and flush all data of the stat
+func (wr *Writer) Close() error {
+	wr.mu.Lock()
+	defer wr.mu.Unlock()
+	close(wr.quit)
+	wr.err = wr.flush(true)
+	if cl, ok := wr.w.(interface{ Close() error }); ok {
+		wr.err = cl.Close()
+	}
+	return wr.err
+}
+
+func (wr *Writer) flush(f bool) error {
+	if !f {
+		return nil
+	}
+	i := 0
+	for i < wr.off {
+		if wr.err = wr.w.Write(wr.records[i]); wr.err != nil {
+			break
+		}
+		i++
+	}
+	wr.off -= i
+	wr.w.Flush()
+	wr.err = wr.w.Error()
+	return wr.err
+}

stat/writer_test.go

@@ -0,0 +1,33 @@
+package stat
+
+import "testing"
+
+type testWriter struct {
+	f  func([]string)
+	fl func()
+}
+
+func (t testWriter) WriteHeader(_ string) error { return nil }
+
+func (t testWriter) Flush() { t.fl() }
+
+func (t testWriter) Error() error { return nil }
+
+func (t testWriter) Write(r []string) error {
+	t.f(r)
+	return nil
+}
+
+func TestTempSize(t *testing.T) {
+	count := 0
+	tW := testWriter{f: func(_ []string) { count++ }, fl: func() {
+		if count != flushSpan {
+			t.Errorf("should flush after %d records", flushSpan)
+		}
+	}}
+	s := NewWriter(tW, "")
+	defer s.Close()
+	for i := 0; i < flushSpan; i++ {
+		s.Write([]string{"1"})
+	}
+}

stats.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"log"
+	"os"
+	"path/filepath"
+
+	"github.com/buger/goreplay/stat"
+	"github.com/buger/goreplay/stat/csv"
+)
+
+func statsWriteRecord(st *stat.Writer, record []string, plugin string) {
+	if st == nil {
+		return
+	}
+	if st.Error() != nil {
+		Debug(1, plugin, " stats closed with error: ", st.Error())
+		st.Close()
+		st = nil
+		return
+	}
+	st.Write(record)
+}
+
+func initializeStatWriter(file string, hdr string) *stat.Writer {
+	csvWr, err := csv.NewWriter(filepath.Join(os.Getenv("GORDIR"), file))
+	if err != nil {
+		log.Fatalf("init meta stat writer error: %q", err)
+	}
+	st := stat.NewWriter(csvWr, "meta")
+	if hdr != "" {
+		st.WriteHeader("type: GOR_STATS\ntitle: message meta")
+	}
+	return st
+}

tcp/tcp_message.go

@@ -1,8 +1,7 @@
 package tcp
 
 import (
-	"bytes"
-	"crypto/sha1"
+	"encoding/binary"
 	"encoding/hex"
 	"fmt"
 	"sort"
@@ -31,7 +30,7 @@ type Stats struct {
 type Message struct {
 	packets []*Packet
 	done    chan bool
-	buf     bytes.Buffer
+	pool    *MessagePool
 	Stats
 }
 
@@ -45,35 +44,46 @@ func NewMessage(srcAddr, dstAddr string, ipVersion uint8) (m *Message) {
 	return
 }
 
-// UUID the unique id of a TCP session it is not granted to be unique!
+// UUID returns the UUID of a TCP request and its response.
 func (m *Message) UUID() []byte {
-	var src, dst string
-	if m.IsIncoming {
-		src = m.SrcAddr
-		dst = m.DstAddr
-	} else {
-		src = m.DstAddr
-		dst = m.SrcAddr
+	var key uint64
+	pckt := m.packets[0]
+
+	// check if response or request have generated the ID before.
+	if m.pool.uuids != nil {
+		lst := len(pckt.SrcIP) - 4
+		if m.IsIncoming {
+			key = uint64(pckt.SrcPort)<<48 | uint64(pckt.DstPort)<<32 |
+				uint64(_uint32(pckt.SrcIP[lst:]))
+		} else {
+			key = uint64(pckt.DstPort)<<48 | uint64(pckt.SrcPort)<<32 |
+				uint64(_uint32(pckt.DstIP[lst:]))
+		}
+		if uuidHex, ok := m.pool.uuids[key]; ok {
+			delete(m.pool.uuids, key)
+			return uuidHex
+		}
 	}
 
-	length := len(src) + len(dst)
-	uuid := make([]byte, length)
-	copy(uuid, src)
-	copy(uuid[len(src):], dst)
-	sha := sha1.Sum(uuid)
-	uuid = make([]byte, 40)
-	hex.Encode(uuid, sha[:])
-
-	return uuid
+	id := make([]byte, 12, 12)
+	binary.BigEndian.PutUint32(id, pckt.Seq)
+	tStamp := m.End.UnixNano()
+	binary.BigEndian.PutUint64(id[4:], uint64(tStamp))
+	uuidHex := make([]byte, 24, 24)
+	hex.Encode(uuidHex[:], id[:])
+	if m.pool.uuids != nil {
+		if len(m.pool.uuids) >= 1000 {
+			m.pool.cleanUUIDs()
+		}
+		m.pool.uuids[key] = uuidHex
+	}
+	return uuidHex
 }
 
 func (m *Message) add(pckt *Packet) {
 	m.Length += len(pckt.Payload)
 	m.LostData += int(pckt.Lost)
 	m.packets = append(m.packets, pckt)
-	if len(pckt.Payload) > 0 {
-		m.buf.Write(pckt.Payload)
-	}
 	m.End = pckt.Timestamp
 }
 
@@ -84,7 +94,16 @@ func (m *Message) Packets() []*Packet {
 
 // Data returns data in this message
 func (m *Message) Data() []byte {
-	return m.buf.Bytes()
+	buf := make([]byte, m.Length, m.Length)
+	var n, nn int
+	for _, p := range m.packets {
+		if len(p.Payload) < 1 {
+			continue
+		}
+		nn = copy(buf[n:], p.Payload)
+		n += nn
+	}
+	return buf
 }
 
 // Sort a helper to sort packets
@@ -104,7 +123,7 @@ type Debugger func(int, ...interface{})
 type HintEnd func(*Message) bool
 
 // HintStart hints the pool to start the reassembling the message, see MessagePool.Start
-// when set, it will be used instead of checking SYN flag
+// when set, it will be called after checking SYN flag
 type HintStart func(*Packet) (IsIncoming, IsOutgoing bool)
 
 // MessagePool holds data of all tcp messages in progress(still receiving/sending packets).
@@ -114,6 +133,7 @@ type MessagePool struct {
 	debug         Debugger
 	maxSize       size.Size // maximum message size, default 5mb
 	pool          map[uint64]*Message
+	uuids         map[uint64][]byte
 	handler       Handler
 	messageExpire time.Duration // the maximum time to wait for the final packet, minimum is 100ms
 	End           HintEnd
@@ -141,19 +161,18 @@ func NewMessagePool(maxSize size.Size, messageExpire time.Duration, debugger Deb
 func (pool *MessagePool) Handler(packet gopacket.Packet) {
 	var in, out bool
 	pckt, err := ParsePacket(packet)
-	if err != nil || pckt == nil {
+	if err != nil {
 		go pool.say(4, fmt.Sprintf("error decoding packet(%dBytes):%s\n", packet.Metadata().CaptureLength, err))
 		return
 	}
+	if pckt == nil {
+		return
+	}
 	pool.Lock()
 	defer pool.Unlock()
-	lst := 3
-	if pckt.Version == 6 {
-		lst = 15
-	}
+	lst := len(pckt.SrcIP) - 4
 	key := uint64(pckt.SrcPort)<<48 | uint64(pckt.DstPort)<<32 |
-		uint64(pckt.SrcIP[lst])<<24 | uint64(pckt.SrcIP[lst-1])<<16 |
-		uint64(pckt.SrcIP[lst-2])<<8 | uint64(pckt.SrcIP[lst-3])
+		uint64(_uint32(pckt.SrcIP[lst:]))
 	m, ok := pool.pool[key]
 	if pckt.RST {
 		if ok {
@@ -161,8 +180,7 @@ func (pool *MessagePool) Handler(packet gopacket.Packet) {
 			<-m.done
 		}
 		key = uint64(pckt.DstPort)<<48 | uint64(pckt.SrcPort)<<32 |
-			uint64(pckt.DstIP[lst])<<24 | uint64(pckt.DstIP[lst-1])<<16 |
-			uint64(pckt.DstIP[lst-2])<<8 | uint64(pckt.DstIP[lst-3])
+			uint64(_uint32(pckt.DstIP[lst:]))
 		m, ok = pool.pool[key]
 		if ok {
 			m.done <- true
@@ -188,10 +206,34 @@ func (pool *MessagePool) Handler(packet gopacket.Packet) {
 	m.IsIncoming = in
 	pool.pool[key] = m
 	m.Start = pckt.Timestamp
+	m.pool = pool
 	go pool.dispatch(key, m)
 	pool.addPacket(m, pckt)
 }
 
+// MatchUUID instructs the pool to use same UUID for request and responses
+// this function should be called at initial stage of the pool
+func (pool *MessagePool) MatchUUID(match bool) {
+	if match {
+		pool.uuids = make(map[uint64][]byte)
+		return
+	}
+	pool.uuids = nil
+}
+
+// run GC on UUID map
+func (pool *MessagePool) cleanUUIDs() {
+	var tStamp int64
+	now := time.Now().UnixNano()
+	for k, v := range pool.uuids {
+		// there is a timestamp wrapped in every ID
+		tStamp = int64(binary.BigEndian.Uint64(v[4:]))
+		if time.Duration(now-tStamp) > pool.messageExpire {
+			delete(pool.uuids, k)
+		}
+	}
+}
+
 func (pool *MessagePool) dispatch(key uint64, m *Message) {
 	select {
 	case <-m.done:
@@ -217,10 +259,20 @@ func (pool *MessagePool) addPacket(m *Message, pckt *Packet) {
 	}
 	m.add(pckt)
 	switch {
+
+	// if one of this cases matches, we dispatch the message
 	case trunc >= 0:
 	case pckt.FIN:
-	case pool.End != nil && pool.End(m):
 	default:
+		if pool.End != nil {
+			if pool.End(m) {
+				break
+			}
+		} else if m.Length > 0 && len(pckt.Payload) < 1 {
+			// generally, data are chunked and sent sequentially in packets, if one packets is empty we assume
+			// that it's just an acknowledgement packet after the final chunk.
+			break
+		}
 		return
 	}
 	m.done <- true
@@ -233,3 +285,7 @@ func (pool *MessagePool) say(level int, args ...interface{}) {
 		pool.debug(level, args...)
 	}
 }
+
+func _uint32(b []byte) uint32 {
+	return binary.BigEndian.Uint32(b)
+}

tcp/tcp_test.go

@@ -51,9 +51,6 @@ func TestMessageParserWithHint(t *testing.T) {
 	pool.Start = func(pckt *Packet) (bool, bool) {
 		return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
 	}
-	pool.End = func(m *Message) bool {
-		return proto.HasFullPayload(m.Data())
-	}
 	packets := GetPackets(1, 30, nil)
 	packets[0].Data()[14:][20:][13] = 2  // SYN flag
 	packets[10].Data()[14:][20:][13] = 2 // SYN flag
@@ -75,8 +72,8 @@ func TestMessageParserWithHint(t *testing.T) {
 		return
 	case m = <-mssg:
 	}
-	if len(m.packets) != 7 {
-		t.Errorf("expected to have 7 packets got %d", len(m.packets))
+	if len(m.packets) != 8 {
+		t.Errorf("expected to have 8 packets got %d", len(m.packets))
 	}
 	if !bytes.HasSuffix(m.Data(), []byte("\n7\r\nNetwork\r\n0\r\n\r\n")) {
 		t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))
@@ -88,8 +85,8 @@ func TestMessageParserWithHint(t *testing.T) {
 		return
 	case m = <-mssg:
 	}
-	if len(m.packets) != 7 {
-		t.Errorf("expected to have 7 packets got %d", len(m.packets))
+	if len(m.packets) != 8 {
+		t.Errorf("expected to have 8 packets got %d", len(m.packets))
 	}
 	if !bytes.HasSuffix(m.Data(), []byte("Network")) {
 		t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("Network"))
@@ -101,8 +98,8 @@ func TestMessageParserWithHint(t *testing.T) {
 		return
 	case m = <-mssg:
 	}
-	if len(m.packets) != 6 {
-		t.Errorf("expected to have 6 packets got %d", len(m.packets))
+	if len(m.packets) != 2 {
+		t.Errorf("expected to have 2 packets got %d", len(m.packets))
 	}
 	if !bytes.HasSuffix(m.Data(), []byte("Content-Length: 0\r\n\r")) {
 		t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("Content-Length: 0\r\n\r"))
@@ -191,16 +188,47 @@ func TestMessageTimeoutReached(t *testing.T) {
 }
 
 func TestMessageUUID(t *testing.T) {
-	m1 := &Message{}
-	m1.IsIncoming = true
-	m1.SrcAddr = "src"
-	m1.DstAddr = "dst"
-	m2 := &Message{}
-	m2.SrcAddr = "dst"
-	m2.DstAddr = "src"
-	if string(m1.UUID()) != string(m2.UUID()) {
-		t.Errorf("expected %s, to equal %s", m1.UUID(), m2.UUID())
+	packets := GetPackets(1, 10, nil)
+	packets[0].Data()[14:][20:][13] = 2 // SYN flag
+	packets[4].Data()[14:][20:][13] = 1 // FIN flag
+	packets[5].Data()[14:][20:][13] = 2 // SYN flag
+	packets[9].Data()[14:][20:][13] = 1 // FIN flag
+	var uuid, uuid1 []byte
+	pool := NewMessagePool(0, 0, nil, func(msg *Message) {
+		if len(uuid) == 0 {
+			uuid = msg.UUID()
+			return
+		}
+		uuid1 = msg.UUID()
+	})
+	pool.MatchUUID(true)
+	for _, p := range packets {
+		pool.Handler(p)
 	}
+
+	if string(uuid) != string(uuid1) {
+		t.Errorf("expected %s, to equal %s", uuid, uuid1)
+	}
+}
+
+func BenchmarkMessageUUID(b *testing.B) {
+	packets := GetPackets(1, 5, nil)
+	packets[0].Data()[14:][20:][13] = 2 // SYN flag
+	packets[4].Data()[14:][20:][13] = 1 // FIN flag
+	var uuid []byte
+	var msg *Message
+	pool := NewMessagePool(0, 0, nil, func(m *Message) {
+		msg = m
+	})
+	pool.MatchUUID(true)
+	for _, p := range packets {
+		pool.Handler(p)
+	}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		uuid = msg.UUID()
+	}
+	_ = uuid
 }
 
 func BenchmarkPacketParseAndSort(b *testing.B) {
@@ -235,7 +263,7 @@ func BenchmarkMessageParserWithoutHint(b *testing.B) {
 }
 
 func BenchmarkMessageParserWithHint(b *testing.B) {
-	var buf [1002][]byte
+	var buf [1003][]byte
 	var chunk = []byte("1e\r\n111111111111111111111111111111\r\n")
 	buf[0] = []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n")
 	for i := 1; i < 1000; i++ {
@@ -246,14 +274,12 @@ func BenchmarkMessageParserWithHint(b *testing.B) {
 	for i := 0; i < len(buf); i++ {
 		packets[i] = GetPackets(uint32(i+10), 1, buf[i])[0]
 	}
+	packets[1002] = GetPackets(1020, 1, nil)[0]
 	var mssg = make(chan *Message, 1)
 	pool := NewMessagePool(1<<30, time.Second*10, nil, func(m *Message) { mssg <- m })
 	pool.Start = func(pckt *Packet) (bool, bool) {
 		return false, proto.HasResponseTitle(pckt.Payload)
 	}
-	pool.End = func(m *Message) bool {
-		return proto.HasFullPayload(m.Data())
-	}
 	b.ResetTimer()
 	b.ReportMetric(float64(len(packets)), "packets/op")
 	b.ReportMetric(float64(1000), "chunks/op")