Fix setting a version

Ensure that docker images gets build in x64 arch, and use nfpm instead of fpm

Dockerfile.dev

@@ -1,4 +1,4 @@
-FROM golang:1.15
+FROM golang:1.17
 
 RUN apt-get update && apt-get install ruby vim-common -y
 

Makefile

@@ -23,36 +23,38 @@ FPMCOMMON= \
     -s dir \
     -C /tmp/gor-build \
 
-release: vendor release-x64 release-mac release-windows
+.PHONY: vendor
+
+release: release-x64 release-mac release-windows
 
 vendor:
 	go mod vendor
 
-release-bin:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+release-bin: vendor
+	docker run --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
 
-release-bin-mac:
-	GOOS=darwin go build -o $(BIN_NAME) $(MAC_LDFLAGS)
+release-bin-mac: vendor
+	GOOS=darwin go build -mod=vendor -o $(BIN_NAME) $(MAC_LDFLAGS)
 
-release-x64:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+release-bin-windows: vendor
+	docker run -it --rm -v `pwd`:$(SOURCE_PATH) -w $(SOURCE_PATH) -e CGO_ENABLED=1 docker.elastic.co/beats-dev/golang-crossbuild:1.16.4-main --build-cmd "make VERSION=$(VERSION) build" -p "windows/amd64"
+
+release-x64: release-bin
 	tar -czf gor_$(VERSION)$(PREFIX)_x64.tar.gz $(BIN_NAME)
 	mkdir -p /tmp/gor-build
 	mv ./$(BIN_NAME) /tmp/gor-build/$(BIN_NAME)
 	cd /tmp/gor-build
 	rm -f goreplay_$(VERSION)_amd64.deb
 	rm -f goreplay-$(VERSION)-1.x86_64.rpm
-	fpm $(FPMCOMMON) -a amd64 -t deb ./=/usr/local/bin
-	fpm $(FPMCOMMON) -a amd64 -t rpm ./=/usr/local/bin
+	nfpm pkg --packager deb --target ./
+	nfpm pkg --packager rpm --target ./
 	rm -rf /tmp/gor-build
 
-release-x86:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+release-x86: release-bin-x86
 	tar -czf gor_$(VERSION)$(PREFIX)_x86.tar.gz $(BIN_NAME)
 	rm $(BIN_NAME)
 
-release-mac:
-	go build -mod=vendor -o $(BIN_NAME) $(MAC_LDFLAGS)
+release-mac: release-bin-mac
 	tar -czf gor_$(VERSION)$(PREFIX)_mac.tar.gz $(BIN_NAME)
 	mkdir -p /tmp/gor-build
 	mv ./$(BIN_NAME) /tmp/gor-build/$(BIN_NAME)
@@ -61,18 +63,18 @@ release-mac:
 	fpm $(FPMCOMMON) -a amd64 -t osxpkg ./=/usr/local/bin
 	rm -rf /tmp/gor-build
 
-release-windows:
-	docker run -it --rm \
-	  -v `pwd`:/go/src/github.com/buger/goreplay \
-	  -w /go/src/github.com/buger/goreplay \
-	  -e CGO_ENABLED=1 \
-	  docker.elastic.co/beats-dev/golang-crossbuild:1.16.4-main \
-	  --build-cmd "make VERSION=$(VERSION) build" \
-	  -p "windows/amd64"
+release-windows: release-bin-windows
 	mv ./gor ./gor.exe
 	zip gor-$(VERSION)$(PREFIX)_windows.zip ./gor.exe
 	rm -rf ./gor.exe
 
+clean:
+	rm -rf *.pkg
+	rm -rf *.zip
+	rm -rf *.gz
+	rm -rf *.deb
+	rm -rf *.rpm
+
 build:
 	go build -mod=vendor -o $(BIN_NAME) $(LDFLAGS)
 
@@ -80,7 +82,7 @@ install:
 	go install $(MAC_LDFLAGS)
 
 build-env:
-	docker build -t $(CONTAINER) -f Dockerfile.dev .
+	docker buildx build --platform linux/amd64 -t $(CONTAINER) -f Dockerfile.dev .
 
 profile:
 	go build && ./$(BIN_NAME) --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

capture/capture.go

@@ -357,8 +357,9 @@ func http1EndHint(m *tcp.Message) bool {
 	if m.MissingChunk() {
 		return false
 	}
-
-	return proto.HasFullPayload(m, m.PacketData()...)
+	
+	req, res := http1StartHint(m.Packets()[0])
+	return proto.HasFullPayload(m, m.PacketData()...) && (req || res)
 }
 
 func (l *Listener) read() {
@@ -397,10 +398,12 @@ func (l *Listener) read() {
 					return
 				case <-timer.C:
 					if h, ok := hndl.handler.(PcapStatProvider); ok {
-						s, _ := h.Stats()
-						stats.Add("packets_received", int64(s.PacketsReceived))
-						stats.Add("packets_dropped", int64(s.PacketsDropped))
-						stats.Add("packets_if_dropped", int64(s.PacketsIfDropped))
+						s, err := h.Stats()
+						if err == nil {
+							stats.Add("packets_received", int64(s.PacketsReceived))
+							stats.Add("packets_dropped", int64(s.PacketsDropped))
+							stats.Add("packets_if_dropped", int64(s.PacketsIfDropped))
+						}
 					}
 				default:
 					data, ci, err := hndl.handler.ReadPacketData()
@@ -520,6 +523,9 @@ func (l *Listener) activatePcapFile() (err error) {
 		handle.Close()
 		return fmt.Errorf("BPF filter error: %q, filter: %s", e, l.BPFFilter)
 	}
+
+	fmt.Println("BPF Filter:", l.BPFFilter)
+
 	l.Handles["pcap_file"] = packetHandle{
 		handler: handle,
 	}
@@ -569,33 +575,33 @@ func (l *Listener) setInterfaces() (err error) {
 	}
 
 	for _, pi := range pifis {
+		if isDevice(l.host, pi) {
+			l.Interfaces = []pcap.Interface{pi}
+			return
+		}
+
 		var ni net.Interface
 		for _, i := range ifis {
+			if i.Name == pi.Name {
+				ni = i
+				break
+			}
+
 			addrs, _ := i.Addrs()
 			for _, a := range addrs {
 				for _, pa := range pi.Addresses {
-					if strings.HasPrefix(a.String(), pa.IP.String()) {
+					if a.String() == pa.IP.String() {
 						ni = i
 						break
 					}
 				}
 			}
-
-			if len(addrs) == 0 && i.Name == pi.Name {
-				ni = i
-				break
-			}
 		}
 
 		if ni.Flags&net.FlagLoopback != 0 {
 			l.loopIndex = ni.Index
 		}
 
-		if isDevice(l.host, pi) {
-			l.Interfaces = []pcap.Interface{pi}
-			return
-		}
-
 		if runtime.GOOS != "windows" {
 			if len(pi.Addresses) == 0 {
 				continue

input_file.go

@@ -63,6 +63,9 @@ type fileInputReader struct {
 	s3        bool
 	queue     payloadQueue
 	readDepth int
+	dryRun    bool
+	path      string
+	stats     *expvar.Map
 }
 
 func (f *fileInputReader) parse(init chan struct{}) error {
@@ -70,8 +73,11 @@ func (f *fileInputReader) parse(init chan struct{}) error {
 	var buffer bytes.Buffer
 	var initialized bool
 
+	lineNum := 0
+
 	for {
 		line, err := f.reader.ReadBytes('\n')
+		lineNum++
 
 		if err != nil {
 			if err != io.EOF {
@@ -92,6 +98,23 @@ func (f *fileInputReader) parse(init chan struct{}) error {
 			asBytes := buffer.Bytes()
 			meta := payloadMeta(asBytes)
 
+			malformed := false
+			if len(meta) < 3 {
+				malformed = true
+			} else {
+				if len(meta[0]) != 1 || len(meta[1]) != 24 || len(meta[2]) != 19 {
+					malformed = true
+				}
+			}
+
+			if malformed {
+				f.stats.Add("malformed", 1)
+
+				Debug(1, fmt.Sprintf("Found malformed record, file: %s, line %d", f.path, lineNum))
+				buffer = bytes.Buffer{}
+				continue
+			}
+
 			timestamp, _ := strconv.ParseInt(string(meta[2]), 10, 64)
 			data := asBytes[:len(asBytes)-1]
 
@@ -112,7 +135,9 @@ func (f *fileInputReader) parse(init chan struct{}) error {
 					initialized = true
 				}
 
-				time.Sleep(100 * time.Millisecond)
+				if !f.dryRun {
+					time.Sleep(100 * time.Millisecond)
+				}
 			}
 
 			buffer = bytes.Buffer{}
@@ -133,7 +158,9 @@ func (f *fileInputReader) wait() {
 			return
 		}
 
-		time.Sleep(100 * time.Millisecond)
+		if !f.dryRun {
+			time.Sleep(100 * time.Millisecond)
+		}
 	}
 
 	return
@@ -149,7 +176,7 @@ func (f *fileInputReader) Close() error {
 	return nil
 }
 
-func newFileInputReader(path string, readDepth int) *fileInputReader {
+func newFileInputReader(path string, readDepth int, stats *expvar.Map, dryRun bool) *fileInputReader {
 	var file io.ReadCloser
 	var err error
 
@@ -164,7 +191,7 @@ func newFileInputReader(path string, readDepth int) *fileInputReader {
 		return nil
 	}
 
-	r := &fileInputReader{file: file, closed: 0, readDepth: readDepth}
+	r := &fileInputReader{path: path, file: file, closed: 0, readDepth: readDepth, stats: stats, dryRun: dryRun}
 	if strings.HasSuffix(path, ".gz") {
 		gzReader, err := gzip.NewReader(file)
 		if err != nil {
@@ -262,7 +289,7 @@ func (i *FileInput) init() (err error) {
 	i.readers = make([]*fileInputReader, len(matches))
 
 	for idx, p := range matches {
-		i.readers[idx] = newFileInputReader(p, i.readDepth)
+		i.readers[idx] = newFileInputReader(p, i.readDepth, i.stats, i.dryRun)
 	}
 
 	i.stats.Add("reader_count", int64(len(matches)))
@@ -398,8 +425,9 @@ func (i *FileInput) emit() {
 	Debug(0, fmt.Sprintf("[INPUT-FILE] FileInput: end of file '%s'\n", i.path))
 
 	if i.dryRun {
-		fmt.Printf("Records found: %v\nFiles processed: %v\nBytes processed: %v\nMax wait: %v\nMin wait: %v\nFirst wait: %v\nIt will take `%v` to replay at current speed.\nFound %v records with out of order timestamp\n",
+		fmt.Printf("Records found: %v\nMalformed records: %v\nFiles processed: %v\nBytes processed: %v\nMax wait: %v\nMin wait: %v\nFirst wait: %v\nIt will take `%v` to replay at current speed.\nFound %v records with out of order timestamp\n",
 			i.stats.Get("total_counter"),
+			i.stats.Get("malformed"),
 			i.stats.Get("reader_count"),
 			i.stats.Get("total_bytes"),
 			i.stats.Get("max_wait"),

nfpm.yaml

@@ -0,0 +1,22 @@
+# nfpm example config file
+#
+# check https://nfpm.goreleaser.com/configuration for detailed usage
+#
+name: "GoReplay"
+arch: "amd64"
+platform: "linux"
+version: "v1.3.1"
+section: "default"
+priority: "extra"
+provides:
+- goreplay
+maintainer: "Leonid Bugaev <hello@goreplay.org>"
+description: |
+ GoReplay is the simplest and safest way to test your app using real traffic before you put it into production. 
+vendor: "GoReplay"
+homepage: "https://goreplay.org"
+license: "AGPL"
+contents:
+- src: ./
+  dst: /usr/local/bin
+  type: dir

output_file.go

@@ -195,7 +195,6 @@ func (o *FileOutput) filename() string {
 
 				if nextChunk {
 					fileIndex++
-					o.currentFileSize = 0
 				}
 			}
 
@@ -309,6 +308,8 @@ func (o *FileOutput) closeLocked() error {
 	}
 
 	o.closed = true
+	o.currentFileSize = 0
+
 	return nil
 }
 

proto/proto.go

@@ -19,6 +19,7 @@ package proto
 import (
 	"bufio"
 	"bytes"
+
 	_ "fmt"
 	"net/http"
 	"net/textproto"
@@ -77,9 +78,12 @@ func header(payload []byte, name []byte) (value []byte, headerStart, headerEnd,
 		headerEnd += headerStart
 		colonIndex = bytes.IndexByte(payload[headerStart:headerEnd], ':')
 		if colonIndex == -1 {
-			break
+			// Malformed header, skip, most likely packet with partial headers
+			headerStart = headerEnd + 1
+			continue
 		}
 		colonIndex += headerStart
+
 		if bytes.EqualFold(payload[headerStart:colonIndex], name) {
 			valueStart = colonIndex + 1
 			valueEnd = headerEnd - 2
@@ -448,34 +452,45 @@ type ProtocolStateSetter interface {
 	ProtocolState() interface{}
 }
 
-type httpProto struct {
-	body         int // body index
-	headerStart  int
-	headerParsed bool // we checked necessary headers
-	hasFullBody  bool // all chunks has been parsed
-	isChunked    bool // Transfer-Encoding: chunked
-	bodyLen      int  // Content-Length's value
-	hasTrailer   bool // Trailer header?
+type HTTPState struct {
+	Body           int // body index
+	HeaderStart    int
+	HeaderEnd      int
+	HeaderParsed   bool // we checked necessary headers
+	HasFullPayload bool // all chunks has been parsed
+	IsChunked      bool // Transfer-Encoding: chunked
+	BodyLen        int  // Content-Length's value
+	HasTrailer     bool // Trailer header?
+	Continue100    bool
 }
 
 // HasFullPayload checks if this message has full or valid payloads and returns true.
 // Message param is optional but recommended on cases where 'data' is storing
 // partial-to-full stream of bytes(packets).
 func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
-	var state *httpProto
+	var state *HTTPState
 	if m != nil {
-		state, _ = m.ProtocolState().(*httpProto)
+		state, _ = m.ProtocolState().(*HTTPState)
 	}
 	if state == nil {
-		state = new(httpProto)
+		state = new(HTTPState)
 		if m != nil {
 			m.SetProtocolState(state)
 		}
 	}
-	if state.headerStart < 1 {
+
+	// Http Packets can only start with a few things, check if this is one of them
+	if len(payloads) == 0 {
+		return false
+	}
+	if !HasRequestTitle(payloads[0]) && !HasResponseTitle(payloads[0]) {
+		return false
+	}
+
+	if state.HeaderStart < 1 {
 		for _, data := range payloads {
-			state.headerStart = MIMEHeadersStartPos(data)
-			if state.headerStart < 0 {
+			state.HeaderStart = MIMEHeadersStartPos(data)
+			if state.HeaderStart < 0 {
 				return false
 			} else {
 				break
@@ -483,7 +498,7 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
 		}
 	}
 
-	if state.body < 1 {
+	if state.Body < 1 || state.HeaderEnd < 1 {
 		var pos int
 		for _, data := range payloads {
 			endPos := MIMEHeadersEndPos(data)
@@ -491,32 +506,42 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
 				pos += len(data)
 			} else {
 				pos += endPos
+				state.HeaderEnd = pos
 			}
 
 			if endPos > 0 {
-				state.body = pos
+				state.Body = pos
 				break
 			}
 		}
 	}
-	if !state.headerParsed {
+
+	if state.HeaderEnd < 1 {
+		return false
+	}
+
+	if !state.HeaderParsed {
 		var pos int
 		for _, data := range payloads {
 			chunked := Header(data, []byte("Transfer-Encoding"))
 
 			if len(chunked) > 0 && bytes.Index(data, []byte("chunked")) > 0 {
-				state.isChunked = true
+				state.IsChunked = true
 				// trailers are generally not allowed in non-chunks body
-				state.hasTrailer = len(Header(data, []byte("Trailer"))) > 0
+				state.HasTrailer = len(Header(data, []byte("Trailer"))) > 0
 			} else {
 				contentLen := Header(data, []byte("Content-Length"))
-				state.bodyLen, _ = atoI(contentLen, 10)
+				state.BodyLen, _ = atoI(contentLen, 10)
 			}
 
 			pos += len(data)
 
-			if state.bodyLen > 0 || pos >= state.body {
-				state.headerParsed = true
+			if string(Header(data, []byte("Expect"))) == "100-continue" {
+				state.Continue100 = true
+			}
+
+			if state.BodyLen > 0 || pos >= state.Body {
+				state.HeaderParsed = true
 				break
 			}
 		}
@@ -526,22 +551,22 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
 	for _, data := range payloads {
 		bodyLen += len(data)
 	}
-	bodyLen -= state.body
+	bodyLen -= state.Body
 
-	if state.isChunked {
+	if state.IsChunked {
 		// check chunks
 		if bodyLen < 1 {
 			return false
 		}
 
 		// check trailer headers
-		if state.hasTrailer {
+		if state.HasTrailer {
 			if bytes.HasSuffix(payloads[len(payloads)-1], []byte("\r\n\r\n")) {
 				return true
 			}
 		} else {
 			if bytes.HasSuffix(payloads[len(payloads)-1], []byte("0\r\n\r\n")) {
-				state.hasFullBody = true
+				state.HasFullPayload = true
 				return true
 			}
 		}
@@ -550,7 +575,7 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
 	}
 
 	// check for content-length header
-	return state.bodyLen == bodyLen
+	return state.BodyLen == bodyLen
 }
 
 // this works with positive integers

proto/proto_test.go

@@ -56,6 +56,12 @@ func TestHeader(t *testing.T) {
 	if val = Header(payload, []byte("host")); !bytes.Equal(val, []byte("www.w3.org")) {
 		t.Error("Should find lower case 1 word header")
 	}
+
+	payload = []byte("GT\r\nContent-Length: 10\r\n\r\n")
+
+	if val = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("10")) {
+		t.Error("Should find in partial payload")
+	}
 }
 
 func TestMIMEHeadersEndPos(t *testing.T) {
@@ -479,6 +485,14 @@ func TestHasFullPayload(t *testing.T) {
 	if got != expected {
 		t.Errorf("expected %v to equal %v", got, expected)
 	}
+
+	// check with trailer and no header
+	m = "Content-Type: text/plain\r\nContent-Length: 23\r\n\r\nMozillaDeveloperNetwork"
+	got = HasFullPayload(nil, []byte(m))
+	expected = false
+	if got != expected {
+		t.Errorf("expected %v to equal %v", got, expected)
+	}
 }
 
 func BenchmarkHasFullPayload(b *testing.B) {

tcp/tcp_message.go

@@ -7,9 +7,10 @@ import (
 	"net"
 	"reflect"
 	"sort"
-	"sync"
 	"time"
 	"unsafe"
+
+	"github.com/buger/goreplay/proto"
 )
 
 // TCPProtocol is a number to indicate type of protocol
@@ -62,10 +63,10 @@ type Stats struct {
 
 // Message is the representation of a tcp message
 type Message struct {
-	packets  []*Packet
-	parser   *MessageParser
-	feedback interface{}
-	Idx      uint16
+	packets          []*Packet
+	parser           *MessageParser
+	feedback         interface{}
+	continueAdjusted bool
 	Stats
 }
 
@@ -168,6 +169,13 @@ func (m *Message) Data() []byte {
 		tmp, _ = copySlice(tmp, len(packetData[0]), packetData[1:]...)
 	}
 
+	// Remove Expect header, since its replay not fully supported
+	if state, ok := m.feedback.(*proto.HTTPState); ok {
+		if state.Continue100 {
+			tmp = proto.DeleteHeader(tmp, []byte("Expect"))
+		}
+	}
+
 	return tmp
 }
 
@@ -200,8 +208,7 @@ type HintStart func(*Packet) (IsRequest, IsOutgoing bool)
 // MessageParser holds data of all tcp messages in progress(still receiving/sending packets).
 // message is identified by its source port and dst port, and last 4bytes of src IP.
 type MessageParser struct {
-	m  []map[uint64]*Message
-	mL []sync.RWMutex
+	m map[uint64]*Message
 
 	messageExpire  time.Duration // the maximum time to wait for the final packet, minimum is 100ms
 	allowIncompete bool
@@ -229,24 +236,18 @@ func NewMessageParser(messages chan *Message, ports []uint16, ips []net.IP, mess
 	parser.packets = make(chan *PcapPacket, 10000)
 
 	if messages == nil {
-		messages = make(chan *Message, 100)
+		messages = make(chan *Message, 1000)
 	}
 	parser.messages = messages
+
+	parser.m = make(map[uint64]*Message)
 	parser.ticker = time.NewTicker(time.Millisecond * 100)
 	parser.close = make(chan struct{}, 1)
 
 	parser.ports = ports
 	parser.ips = ips
 
-	for i := 0; i < 10; i++ {
-		parser.m = append(parser.m, make(map[uint64]*Message))
-		parser.mL = append(parser.mL, sync.RWMutex{})
-	}
-
-	for i := 0; i < 10; i++ {
-		go parser.wait(i)
-	}
-
+	go parser.wait()
 	return parser
 }
 
@@ -258,7 +259,7 @@ func (parser *MessageParser) PacketHandler(packet *PcapPacket) {
 	parser.packets <- packet
 }
 
-func (parser *MessageParser) wait(index int) {
+func (parser *MessageParser) wait() {
 	var (
 		now time.Time
 	)
@@ -267,7 +268,7 @@ func (parser *MessageParser) wait(index int) {
 		case pckt := <-parser.packets:
 			parser.processPacket(parser.parsePacket(pckt))
 		case now = <-parser.ticker.C:
-			parser.timer(now, index)
+			parser.timer(now)
 		case <-parser.close:
 			parser.ticker.Stop()
 			// parser.Close should wait for this function to return
@@ -281,7 +282,9 @@ func (parser *MessageParser) wait(index int) {
 func (parser *MessageParser) parsePacket(pcapPkt *PcapPacket) *Packet {
 	pckt, err := ParsePacket(pcapPkt.Data, pcapPkt.LType, pcapPkt.LTypeLen, pcapPkt.Ci, false)
 	if err != nil {
-		stats.Add("packet_error", 1)
+		if _, empty := err.(EmptyPacket); !empty {
+			stats.Add("packet_error", 1)
+		}
 		return nil
 	}
 
@@ -307,28 +310,19 @@ func (parser *MessageParser) processPacket(pckt *Packet) {
 
 	// Trying to build unique hash, but there is small chance of collision
 	// No matter if it is request or response, all packets in the same message have same
-	mID := pckt.MessageID()
-	mIDX := pckt.SrcPort % 10
-
-	parser.mL[mIDX].Lock()
-	m, ok := parser.m[mIDX][mID]
-	if !ok {
-		parser.mL[mIDX].Unlock()
-
-		mIDX = pckt.DstPort % 10
-		parser.mL[mIDX].Lock()
-		m, ok = parser.m[mIDX][mID]
-
-		if !ok {
-			parser.mL[mIDX].Unlock()
-		}
-	}
-
+	m, ok := parser.m[pckt.MessageID()]
 	switch {
 	case ok:
+		if m.Direction == DirUnknown {
+			if in, out := parser.Start(pckt); in || out {
+				if in {
+					m.Direction = DirIncoming
+				} else {
+					m.Direction = DirOutcoming
+				}
+			}
+		}
 		parser.addPacket(m, pckt)
-
-		parser.mL[mIDX].Unlock()
 		return
 	case pckt.Direction == DirUnknown && parser.Start != nil:
 		if in, out := parser.Start(pckt); in || out {
@@ -340,25 +334,14 @@ func (parser *MessageParser) processPacket(pckt *Packet) {
 		}
 	}
 
-	if pckt.Direction == DirIncoming {
-		mIDX = pckt.SrcPort % 10
-	} else {
-		mIDX = pckt.DstPort % 10
-	}
-
-	parser.mL[mIDX].Lock()
-
 	m = new(Message)
 	m.Direction = pckt.Direction
-
-	parser.m[mIDX][mID] = m
-
-	m.Idx = mIDX
+	m.SrcAddr = pckt.SrcIP.String()
+	m.DstAddr = pckt.DstIP.String()
+	parser.m[pckt.MessageID()] = m
 	m.Start = pckt.Timestamp
 	m.parser = parser
 	parser.addPacket(m, pckt)
-
-	parser.mL[mIDX].Unlock()
 }
 
 func (parser *MessageParser) addPacket(m *Message, pckt *Packet) bool {
@@ -371,12 +354,40 @@ func (parser *MessageParser) addPacket(m *Message, pckt *Packet) bool {
 	if parser.End != nil {
 		if parser.End(m) {
 			parser.Emit(m)
+			return true
 		}
+
+		parser.Fix100Continue(m)
 	}
 
 	return true
 }
 
+func (parser *MessageParser) Fix100Continue(m *Message) {
+	// Only adjust a message once
+	if state, ok := m.feedback.(*proto.HTTPState); ok && state.Continue100 && !m.continueAdjusted {
+		delete(parser.m, m.packets[0].MessageID())
+
+		// Shift Ack by given offset
+		// Size of "HTTP/1.1 100 Continue\r\n\r\n" message
+		for _, p := range m.packets {
+			p.messageID = 0
+			p.Ack += 25
+		}
+
+		// If next section was aready approved and received, merge messages
+		if next, found := parser.m[m.packets[0].MessageID()]; found {
+			for _, p := range next.packets {
+				parser.addPacket(m, p)
+			}
+		}
+
+		// Re-add (or override) again with new message and ID
+		parser.m[m.packets[0].MessageID()] = m
+		m.continueAdjusted = true
+	}
+}
+
 func (parser *MessageParser) Read() *Message {
 	m := <-parser.messages
 	return m
@@ -385,7 +396,7 @@ func (parser *MessageParser) Read() *Message {
 func (parser *MessageParser) Emit(m *Message) {
 	stats.Add("message_count", 1)
 
-	delete(parser.m[m.Idx], m.packets[0].MessageID())
+	delete(parser.m, m.packets[0].MessageID())
 
 	parser.messages <- m
 }
@@ -396,14 +407,13 @@ func GetUnexportedField(field reflect.Value) interface{} {
 
 var failMsg int
 
-func (parser *MessageParser) timer(now time.Time, index int) {
+func (parser *MessageParser) timer(now time.Time) {
 	packetLen = 0
-	parser.mL[index].Lock()
 
 	packetQueueLen.Set(int64(len(parser.packets)))
-	messageQueueLen.Set(int64(len(parser.m[index])))
+	messageQueueLen.Set(int64(len(parser.m)))
 
-	for _, m := range parser.m[index] {
+	for id, m := range parser.m {
 		if now.Sub(m.End) > parser.messageExpire {
 			m.TimedOut = true
 			stats.Add("message_timeout_count", 1)
@@ -412,11 +422,9 @@ func (parser *MessageParser) timer(now time.Time, index int) {
 				parser.Emit(m)
 			}
 
-			delete(parser.m[index], m.packets[0].MessageID())
+			delete(parser.m, id)
 		}
 	}
-
-	parser.mL[index].Unlock()
 }
 
 func (parser *MessageParser) Close() error {

tcp/tcp_packet.go

@@ -169,7 +169,16 @@ func (pckt *Packet) parse(data []byte, lType, lTypeLen int, cp *gopacket.Capture
 		return ErrHdrLength("TCP opts")
 	}
 
-	if !allowEmpty && len(ndata[dOf:]) == 0 {
+	// There are case when packet have padding but dOf shows its not
+	empty := true
+	for i := 0; i < len(ndata[dOf:]); i++ {
+		if ndata[dOf:][i] != 0 {
+			empty = false
+			break
+		}
+	}
+
+	if !allowEmpty && empty {
 		return EmptyPacket("")
 	}
 

tcp/tcp_test.go

@@ -213,15 +213,16 @@ func TestMessageParserWithoutHint(t *testing.T) {
 }
 
 func TestMessageTimeoutReached(t *testing.T) {
-	var data [63 << 10]byte
+	const size = 63 << 11
+	var data [size >> 1]byte
 	packets := GetPackets(true, 1, 2, data[:])
 	p := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true)
 	p.processPacket(packets[0])
-	time.Sleep(time.Millisecond * 100)
+	time.Sleep(time.Millisecond * 200)
 	p.processPacket(packets[1])
 	m := p.Read()
-	if m.Length != 63<<10 {
-		t.Errorf("expected %d to equal %d", m.Length, 63<<10)
+	if m.Length != size {
+		t.Errorf("expected %d to equal %d", m.Length, size)
 	}
 	if !m.TimedOut {
 		t.Error("expected message to be timeout")

version.go

@@ -1,4 +1,4 @@
 package main
 
 // VERSION the current version of goreplay
-const VERSION = "1.3.0"
+var VERSION = "1.3.0"