Merge pull request #992 from buger/fix/989-duplicate

Issue was introduced while fixing windows https://github.com/buger/goreplay/commit/c9274ac92a6f021240d82682002240cfceaecd5e

Added exception for Windows, which by default allows interfaces without IPs.
Interface name check moved higher, so if interface namee or IP match, rest of check will be ignored.

Additionally windows npcap loopback mechanism can now be picked by specifying 127.0.0.1 or loopback IP.

Fix #989

.gitignore

@@ -1,3 +1,4 @@
+vendor
 *.swp
 *.gor
 *.rpm
@@ -5,8 +6,9 @@
 *.deb
 *.pkg
 *.exe
-
+*.pprof
 *.out
+hey
 
 *.bin
 

Makefile

@@ -23,16 +23,19 @@ FPMCOMMON= \
     -s dir \
     -C /tmp/gor-build \
 
-release: release-x64 release-mac release-windows
+release: vendor 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 -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+	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-mac:
 	GOOS=darwin go build -o $(BIN_NAME) $(MAC_LDFLAGS)
 
 release-x64:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i $(CONTAINER) go build -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+	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)
 	tar -czf gor_$(VERSION)$(PREFIX)_x64.tar.gz $(BIN_NAME)
 	mkdir -p /tmp/gor-build
 	mv ./$(BIN_NAME) /tmp/gor-build/$(BIN_NAME)
@@ -44,12 +47,12 @@ release-x64:
 	rm -rf /tmp/gor-build
 
 release-x86:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i $(CONTAINER) go build -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+	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)
 	tar -czf gor_$(VERSION)$(PREFIX)_x86.tar.gz $(BIN_NAME)
 	rm $(BIN_NAME)
 
 release-mac:
-	go build -o $(BIN_NAME) $(MAC_LDFLAGS)
+	go build -mod=vendor -o $(BIN_NAME) $(MAC_LDFLAGS)
 	tar -czf gor_$(VERSION)$(PREFIX)_mac.tar.gz $(BIN_NAME)
 	mkdir -p /tmp/gor-build
 	mv ./$(BIN_NAME) /tmp/gor-build/$(BIN_NAME)
@@ -71,7 +74,7 @@ release-windows:
 	rm -rf ./gor.exe
 
 build:
-	go build -o $(BIN_NAME) $(LDFLAGS)
+	go build -mod=vendor -o $(BIN_NAME) $(LDFLAGS)
 
 install:
 	go install $(MAC_LDFLAGS)

capture/af_packet.go

@@ -0,0 +1,32 @@
+// +build !linux
+
+package capture
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/google/gopacket"
+)
+
+func newAfpacketHandle(device string, snaplen int, block_size int, num_blocks int,
+	useVLAN bool, timeout time.Duration) (*afpacketHandle, error) {
+	return nil, fmt.Errorf("Not implemented")
+}
+
+func afpacketComputeSize(targetSizeMb int, snaplen int, pageSize int) (
+	frameSize int, blockSize int, numBlocks int, err error) {
+	return 0, 0, 0, fmt.Errorf("Not implemented")
+}
+
+type afpacketHandle struct{}
+
+// ReadPacketData satisfies PacketDataSource interface
+func (h *afpacketHandle) ReadPacketData() (data []byte, ci gopacket.CaptureInfo, err error) {
+	return nil, gopacket.CaptureInfo{}, fmt.Errorf("Not implemented")
+}
+
+// SetBPFFilter translates a BPF filter string into BPF RawInstruction and applies them.
+func (h *afpacketHandle) SetBPFFilter(filter string, snaplen int) (err error) {
+	return fmt.Errorf("Not implemented")
+}

capture/af_packet_linux.go

@@ -0,0 +1,115 @@
+// +build linux
+
+package capture
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/afpacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcap"
+	"golang.org/x/net/bpf"
+
+	_ "github.com/google/gopacket/layers"
+)
+
+type afpacketHandle struct {
+	TPacket *afpacket.TPacket
+}
+
+func newAfpacketHandle(device string, snaplen int, block_size int, num_blocks int,
+	useVLAN bool, timeout time.Duration) (*afpacketHandle, error) {
+
+	h := &afpacketHandle{}
+	var err error
+
+	if device == "any" {
+		h.TPacket, err = afpacket.NewTPacket(
+			afpacket.OptFrameSize(snaplen),
+			afpacket.OptBlockSize(block_size),
+			afpacket.OptNumBlocks(num_blocks),
+			afpacket.OptAddVLANHeader(false),
+			afpacket.OptPollTimeout(timeout),
+			afpacket.SocketRaw,
+			afpacket.TPacketVersion3)
+	} else {
+		h.TPacket, err = afpacket.NewTPacket(
+			afpacket.OptInterface(device),
+			afpacket.OptFrameSize(snaplen),
+			afpacket.OptBlockSize(block_size),
+			afpacket.OptNumBlocks(num_blocks),
+			afpacket.OptAddVLANHeader(false),
+			afpacket.OptPollTimeout(timeout),
+			afpacket.SocketRaw,
+			afpacket.TPacketVersion3)
+	}
+	return h, err
+}
+
+// ReadPacketData satisfies PacketDataSource interface
+func (h *afpacketHandle) ReadPacketData() (data []byte, ci gopacket.CaptureInfo, err error) {
+	return h.TPacket.ReadPacketData()
+}
+
+// SetBPFFilter translates a BPF filter string into BPF RawInstruction and applies them.
+func (h *afpacketHandle) SetBPFFilter(filter string, snaplen int) (err error) {
+	pcapBPF, err := pcap.CompileBPFFilter(layers.LinkTypeEthernet, snaplen, filter)
+	if err != nil {
+		return err
+	}
+	bpfIns := []bpf.RawInstruction{}
+	for _, ins := range pcapBPF {
+		bpfIns2 := bpf.RawInstruction{
+			Op: ins.Code,
+			Jt: ins.Jt,
+			Jf: ins.Jf,
+			K:  ins.K,
+		}
+		bpfIns = append(bpfIns, bpfIns2)
+	}
+	if h.TPacket.SetBPF(bpfIns); err != nil {
+		return err
+	}
+	return h.TPacket.SetBPF(bpfIns)
+}
+
+// LinkType returns ethernet link type.
+func (h *afpacketHandle) LinkType() layers.LinkType {
+	return layers.LinkTypeEthernet
+}
+
+// Close will close afpacket source.
+func (h *afpacketHandle) Close() {
+	h.TPacket.Close()
+}
+
+// SocketStats prints received, dropped, queue-freeze packet stats.
+func (h *afpacketHandle) SocketStats() (as afpacket.SocketStats, asv afpacket.SocketStatsV3, err error) {
+	return h.TPacket.SocketStats()
+}
+
+// afpacketComputeSize computes the block_size and the num_blocks in such a way that the
+// allocated mmap buffer is close to but smaller than target_size_mb.
+// The restriction is that the block_size must be divisible by both the
+// frame size and page size.
+func afpacketComputeSize(targetSizeMb int, snaplen int, pageSize int) (
+	frameSize int, blockSize int, numBlocks int, err error) {
+
+	if snaplen < pageSize {
+		frameSize = pageSize / (pageSize / snaplen)
+	} else {
+		frameSize = (snaplen/pageSize + 1) * pageSize
+	}
+
+	// 128 is the default from the gopacket library so just use that
+	blockSize = frameSize * 128
+	numBlocks = (targetSizeMb * 1024 * 1024) / blockSize
+
+	if numBlocks == 0 {
+		return 0, 0, 0, fmt.Errorf("Interface buffersize is too small")
+	}
+
+	return frameSize, blockSize, numBlocks, nil
+}

capture/capture.go

@@ -3,6 +3,7 @@ package capture
 import (
 	"context"
 	"errors"
+	"expvar"
 	"fmt"
 	"io"
 	"log"
@@ -14,6 +15,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/buger/goreplay/proto"
 	"github.com/buger/goreplay/size"
 	"github.com/buger/goreplay/tcp"
 
@@ -22,9 +24,20 @@ import (
 	"github.com/google/gopacket/pcap"
 )
 
+var stats *expvar.Map
+
+func init() {
+	stats = expvar.NewMap("raw")
+	stats.Init()
+}
+
 // PacketHandler is a function that is used to handle packets
 type PacketHandler func(*tcp.Packet)
 
+type PcapStatProvider interface {
+	Stats() (*pcap.Stats, error)
+}
+
 // PcapOptions options that can be set on a pcap capture handle,
 // these options take effect on inactive pcap handles
 type PcapOptions struct {
@@ -42,14 +55,18 @@ type Listener struct {
 	sync.Mutex
 	Transport  string       // transport layer default to tcp
 	Activate   func() error // function is used to activate the engine. it must be called before reading packets
-	Handles    map[string]gopacket.ZeroCopyPacketDataSource
+	Handles    map[string]packetHandle
 	Interfaces []pcap.Interface
 	loopIndex  int
 	Reading    chan bool // this channel is closed when the listener has started reading packets
 	PcapOptions
-	Engine        EngineType
-	ports         []uint16 // src or/and dst ports
-	trackResponse bool
+	Engine          EngineType
+	ports           []uint16 // src or/and dst ports
+	trackResponse   bool
+	expiry          time.Duration
+	allowIncomplete bool
+	messages        chan *tcp.Message
+	protocol        tcp.TCPProtocol
 
 	host string // pcap file name or interface (name, hardware addr, index or ip address)
 
@@ -57,6 +74,11 @@ type Listener struct {
 	quit      chan struct{}
 }
 
+type packetHandle struct {
+	handler gopacket.PacketDataSource
+	ips     []net.IP
+}
+
 // EngineType ...
 type EngineType uint8
 
@@ -65,6 +87,7 @@ const (
 	EnginePcap EngineType = 1 << iota
 	EnginePcapFile
 	EngineRawSocket
+	EngineAFPacket
 )
 
 // Set is here so that EngineType can implement flag.Var
@@ -74,8 +97,10 @@ func (eng *EngineType) Set(v string) error {
 		*eng = EnginePcap
 	case "pcap_file":
 		*eng = EnginePcapFile
-	case "raw_socket", "af_packet":
+	case "raw_socket":
 		*eng = EngineRawSocket
+	case "af_packet":
+		*eng = EngineAFPacket
 	default:
 		return fmt.Errorf("invalid engine %s", v)
 	}
@@ -90,6 +115,8 @@ func (eng *EngineType) String() (e string) {
 		e = "libpcap"
 	case EngineRawSocket:
 		e = "raw_socket"
+	case EngineAFPacket:
+		e = "af_packet"
 	default:
 		e = ""
 	}
@@ -99,7 +126,7 @@ func (eng *EngineType) String() (e string) {
 // NewListener creates and initialize a new Listener. if transport or/and engine are invalid/unsupported
 // is "tcp" and "pcap", are assumed. l.Engine and l.Transport can help to get the values used.
 // if there is an error it will be associated with getting network interfaces
-func NewListener(host string, ports []uint16, transport string, engine EngineType, trackResponse bool) (l *Listener, err error) {
+func NewListener(host string, ports []uint16, transport string, engine EngineType, protocol tcp.TCPProtocol, trackResponse bool, expiry time.Duration, allowIncomplete bool) (l *Listener, err error) {
 	l = &Listener{}
 
 	l.host = host
@@ -112,11 +139,16 @@ func NewListener(host string, ports []uint16, transport string, engine EngineTyp
 	if transport != "" {
 		l.Transport = transport
 	}
-	l.Handles = make(map[string]gopacket.ZeroCopyPacketDataSource)
+	l.Handles = make(map[string]packetHandle)
 	l.trackResponse = trackResponse
 	l.closeDone = make(chan struct{})
 	l.quit = make(chan struct{})
 	l.Reading = make(chan bool)
+	l.expiry = expiry
+	l.allowIncomplete = allowIncomplete
+	l.protocol = protocol
+	l.messages = make(chan *tcp.Message, 10000)
+
 	switch engine {
 	default:
 		l.Engine = EnginePcap
@@ -124,6 +156,9 @@ func NewListener(host string, ports []uint16, transport string, engine EngineTyp
 	case EngineRawSocket:
 		l.Engine = EngineRawSocket
 		l.Activate = l.activateRawSocket
+	case EngineAFPacket:
+		l.Engine = EngineAFPacket
+		l.Activate = l.activateAFPacket
 	case EnginePcapFile:
 		l.Engine = EnginePcapFile
 		l.Activate = l.activatePcapFile
@@ -146,8 +181,8 @@ func (l *Listener) SetPcapOptions(opts PcapOptions) {
 // Listen listens for packets from the handles, and call handler on every packet received
 // until the context done signal is sent or there is unrecoverable error on all handles.
 // this function must be called after activating pcap handles
-func (l *Listener) Listen(ctx context.Context, handler PacketHandler) (err error) {
-	l.read(handler)
+func (l *Listener) Listen(ctx context.Context) (err error) {
+	l.read()
 	done := ctx.Done()
 	select {
 	case <-done:
@@ -160,11 +195,11 @@ func (l *Listener) Listen(ctx context.Context, handler PacketHandler) (err error
 }
 
 // ListenBackground is like listen but can run concurrently and signal error through channel
-func (l *Listener) ListenBackground(ctx context.Context, handler PacketHandler) chan error {
+func (l *Listener) ListenBackground(ctx context.Context) chan error {
 	err := make(chan error, 1)
 	go func() {
 		defer close(err)
-		if e := l.Listen(ctx, handler); err != nil {
+		if e := l.Listen(ctx); err != nil {
 			err <- e
 		}
 	}()
@@ -201,26 +236,11 @@ func (l *Listener) Filter(ifi pcap.Interface) (filter string) {
 		filter = fmt.Sprintf("%s or %s", filter, responseFilter)
 	}
 
+	// filter = fmt.Sprintf("((((ip[2:2] - ((ip[0]&0xf)<<2)) - ((tcp[12]&0xf0)>>2)) != 0)) and (%s)", filter)
+
 	return
 }
 
-// PcapDumpHandler returns a handler to write packet data in PCAP
-// format, See http://wiki.wireshark.org/Development/LibpcapFileFormathandler.
-// if link layer is invalid Ethernet is assumed
-// func PcapDumpHandler(file *os.File, link layers.LinkType) (handler func(packet *tcp.Packet) error, err error) {
-// 	if link.String() == "" {
-// 		link = layers.LinkTypeEthernet
-// 	}
-// 	w := NewWriterNanos(file)
-// 	err = w.WriteFileHeader(64<<10, link)
-// 	if err != nil {
-// 		return nil, err
-// 	}
-// 	return func(packet *tcp.Packet) error {
-// 		return w.WritePacket(*packet.Info, packet.Data)
-// 	}, nil
-// }
-
 // PcapHandle returns new pcap Handle from dev on success.
 // this function should be called after setting all necessary options for this listener
 func (l *Listener) PcapHandle(ifi pcap.Interface) (handle *pcap.Handle, err error) {
@@ -230,9 +250,11 @@ func (l *Listener) PcapHandle(ifi pcap.Interface) (handle *pcap.Handle, err erro
 		return nil, fmt.Errorf("inactive handle error: %q, interface: %q", err, ifi.Name)
 	}
 	defer inactive.CleanUp()
-	if l.TimestampType != "" {
+
+	if l.TimestampType != "" && l.TimestampType != "go" {
 		var ts pcap.TimestampSource
 		ts, err = pcap.TimestampSourceFromString(l.TimestampType)
+		fmt.Println("Setting custom Timestamp Source. Supported values: `go`, ", inactive.SupportedTimestamps())
 		err = inactive.SetTimestampSource(ts)
 		if err != nil {
 			return nil, fmt.Errorf("%q: supported timestamps: %q, interface: %q", err, inactive.SupportedTimestamps(), ifi.Name)
@@ -275,7 +297,7 @@ func (l *Listener) PcapHandle(ifi pcap.Interface) (handle *pcap.Handle, err erro
 		}
 	}
 	if l.BufferTimeout == 0 {
-		l.BufferTimeout = pcap.BlockForever
+		l.BufferTimeout = 2000 * time.Millisecond
 	}
 	err = inactive.SetTimeout(l.BufferTimeout)
 	if err != nil {
@@ -285,12 +307,16 @@ func (l *Listener) PcapHandle(ifi pcap.Interface) (handle *pcap.Handle, err erro
 	if err != nil {
 		return nil, fmt.Errorf("PCAP Activate device error: %q, interface: %q", err, ifi.Name)
 	}
-	l.BPFFilter = l.Filter(ifi)
-	fmt.Println("Interface:", ifi.Name, ". BPF Filter:", l.BPFFilter)
-	err = handle.SetBPFFilter(l.BPFFilter)
+
+	bpfFilter := l.BPFFilter
+	if bpfFilter == "" {
+		bpfFilter = l.Filter(ifi)
+	}
+	fmt.Println("Interface:", ifi.Name, ". BPF Filter:", bpfFilter)
+	err = handle.SetBPFFilter(bpfFilter)
 	if err != nil {
 		handle.Close()
-		return nil, fmt.Errorf("BPF filter error: %q%s, interface: %q", err, l.BPFFilter, ifi.Name)
+		return nil, fmt.Errorf("BPF filter error: %q%s, interface: %q", err, bpfFilter, ifi.Name)
 	}
 	return
 }
@@ -304,7 +330,9 @@ func (l *Listener) SocketHandle(ifi pcap.Interface) (handle Socket, err error) {
 	if err = handle.SetPromiscuous(l.Promiscuous || l.Monitor); err != nil {
 		return nil, fmt.Errorf("promiscuous mode error: %q, interface: %q", err, ifi.Name)
 	}
-	l.BPFFilter = l.Filter(ifi)
+	if l.BPFFilter == "" {
+		l.BPFFilter = l.Filter(ifi)
+	}
 	fmt.Println("BPF Filter: ", l.BPFFilter)
 	if err = handle.SetBPFFilter(l.BPFFilter); err != nil {
 		handle.Close()
@@ -314,16 +342,39 @@ func (l *Listener) SocketHandle(ifi pcap.Interface) (handle Socket, err error) {
 	return
 }
 
-func (l *Listener) read(handler PacketHandler) {
+func http1StartHint(pckt *tcp.Packet) (isRequest, isResponse bool) {
+	if proto.HasRequestTitle(pckt.Payload) {
+		return true, false
+	}
+
+	if proto.HasResponseTitle(pckt.Payload) {
+		return false, true
+	}
+
+	// No request or response detected
+	return false, false
+}
+
+func http1EndHint(m *tcp.Message) bool {
+	if m.MissingChunk() {
+		return false
+	}
+
+	return proto.HasFullPayload(m, m.PacketData()...)
+}
+
+func (l *Listener) read() {
 	l.Lock()
 	defer l.Unlock()
 	for key, handle := range l.Handles {
-		go func(key string, hndl gopacket.ZeroCopyPacketDataSource) {
+		go func(key string, hndl packetHandle) {
+			runtime.LockOSThread()
+
 			defer l.closeHandles(key)
 			linkSize := 14
 			linkType := int(layers.LinkTypeEthernet)
-			if _, ok := hndl.(*pcap.Handle); ok {
-				linkType = int(hndl.(*pcap.Handle).LinkType())
+			if _, ok := hndl.handler.(*pcap.Handle); ok {
+				linkType = int(hndl.handler.(*pcap.Handle).LinkType())
 				linkSize, ok = pcapLinkTypeLength(linkType)
 				if !ok {
 					if os.Getenv("GORDEBUG") != "0" {
@@ -333,17 +384,39 @@ func (l *Listener) read(handler PacketHandler) {
 				}
 			}
 
+			messageParser := tcp.NewMessageParser(l.messages, l.ports, hndl.ips, l.expiry, l.allowIncomplete)
+
+			if l.protocol == tcp.ProtocolHTTP {
+				messageParser.Start = http1StartHint
+				messageParser.End = http1EndHint
+			}
+
+			timer := time.NewTicker(1 * time.Second)
+
 			for {
 				select {
 				case <-l.quit:
 					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))
+					}
 				default:
-					data, ci, err := hndl.ZeroCopyReadPacketData()
+					data, ci, err := hndl.handler.ReadPacketData()
 					if err == nil {
-						pckt, err := tcp.ParsePacket(data, linkType, linkSize, &ci)
-						if err == nil {
-							handler(pckt)
+						if l.TimestampType == "go" {
+							ci.Timestamp = time.Now()
 						}
+
+						messageParser.PacketHandler(&tcp.PcapPacket{
+							Data:     data,
+							LType:    linkType,
+							LTypeLen: linkSize,
+							Ci:       &ci,
+						})
 						continue
 					}
 					if enext, ok := err.(pcap.NextError); ok && enext == pcap.NextErrorTimeoutExpired {
@@ -369,15 +442,18 @@ func (l *Listener) read(handler PacketHandler) {
 	close(l.Reading)
 }
 
+func (l *Listener) Messages() chan *tcp.Message {
+	return l.messages
+}
+
 func (l *Listener) closeHandles(key string) {
 	l.Lock()
 	defer l.Unlock()
 	if handle, ok := l.Handles[key]; ok {
-		if _, ok = handle.(Socket); ok {
-			handle.(Socket).Close()
-		} else {
-			handle.(*pcap.Handle).Close()
+		if c, ok := handle.handler.(io.Closer); ok {
+			c.Close()
 		}
+
 		delete(l.Handles, key)
 		if len(l.Handles) == 0 {
 			close(l.closeDone)
@@ -395,7 +471,10 @@ func (l *Listener) activatePcap() error {
 			msg += ("\n" + e.Error())
 			continue
 		}
-		l.Handles[ifi.Name] = handle
+		l.Handles[ifi.Name] = packetHandle{
+			handler: handle,
+			ips:     interfaceIPs(ifi),
+		}
 	}
 	if len(l.Handles) == 0 {
 		return fmt.Errorf("pcap handles error:%s", msg)
@@ -416,7 +495,10 @@ func (l *Listener) activateRawSocket() error {
 			msg += ("\n" + e.Error())
 			continue
 		}
-		l.Handles[ifi.Name] = handle
+		l.Handles[ifi.Name] = packetHandle{
+			handler: handle,
+			ips:     interfaceIPs(ifi),
+		}
 	}
 	if len(l.Handles) == 0 {
 		return fmt.Errorf("raw socket handles error:%s", msg)
@@ -440,10 +522,46 @@ func (l *Listener) activatePcapFile() (err error) {
 		handle.Close()
 		return fmt.Errorf("BPF filter error: %q, filter: %s", e, l.BPFFilter)
 	}
-	l.Handles["pcap_file"] = handle
+	l.Handles["pcap_file"] = packetHandle{
+		handler: handle,
+	}
 	return
 }
 
+func (l *Listener) activateAFPacket() error {
+	szFrame, szBlock, numBlocks, err := afpacketComputeSize(32, 32<<10, os.Getpagesize())
+	if err != nil {
+		return err
+	}
+
+	var msg string
+	for _, ifi := range l.Interfaces {
+		handle, err := newAfpacketHandle(ifi.Name, szFrame, szBlock, numBlocks, false, pcap.BlockForever)
+
+		if err != nil {
+			msg += ("\n" + err.Error())
+			continue
+		}
+
+		if l.BPFFilter == "" {
+			l.BPFFilter = l.Filter(ifi)
+		}
+		fmt.Println("Interface:", ifi.Name, ". BPF Filter:", l.BPFFilter)
+		handle.SetBPFFilter(l.BPFFilter, 64<<10)
+
+		l.Handles[ifi.Name] = packetHandle{
+			handler: handle,
+			ips:     interfaceIPs(ifi),
+		}
+	}
+
+	if len(l.Handles) == 0 {
+		return fmt.Errorf("pcap handles error:%s", msg)
+	}
+
+	return nil
+}
+
 func (l *Listener) setInterfaces() (err error) {
 	var pifis []pcap.Interface
 	pifis, err = pcap.FindAllDevs()
@@ -451,38 +569,67 @@ func (l *Listener) setInterfaces() (err error) {
 	if err != nil {
 		return
 	}
+
 	for _, pi := range pifis {
 		var ni net.Interface
 		for _, i := range ifis {
-			if i.Name == pi.Name {
+			addrs, _ := i.Addrs()
+			for _, a := range addrs {
+				for _, pa := range pi.Addresses {
+					if strings.HasPrefix(a.String(), pa.IP.String()) {
+						ni = i
+						break
+					}
+				}
+			}
+
+			if len(addrs) == 0 && i.Name == pi.Name {
 				ni = i
 				break
 			}
 		}
 
-		if net.Flags(pi.Flags)&net.FlagLoopback != 0 {
+		if ni.Flags&net.FlagLoopback != 0 {
 			l.loopIndex = ni.Index
 		}
-		if net.Flags(pi.Flags)&net.FlagUp == 0 {
-			continue
-		}
+
 		if isDevice(l.host, pi) {
 			l.Interfaces = []pcap.Interface{pi}
 			return
 		}
-		for _, addr := range pi.Addresses {
-			if addr.IP.String() == l.host {
-				l.Interfaces = []pcap.Interface{pi}
-				return
+
+		if runtime.GOOS != "windows" {
+			if len(pi.Addresses) == 0 {
+				continue
+			}
+
+			if ni.Flags&net.FlagUp == 0 {
+				continue
 			}
 		}
+
+		l.Interfaces = append(l.Interfaces, pi)
 	}
-	l.Interfaces = pifis
 	return
 }
 
 func isDevice(addr string, ifi pcap.Interface) bool {
-	return addr == ifi.Name
+	// Windows npcap loopback have no IPs
+	if addr == "127.0.0.1" && ifi.Name == `\Device\NPF_Loopback` {
+		return true
+	}
+
+	if addr == ifi.Name {
+		return true
+	}
+
+	for _, _addr := range ifi.Addresses {
+		if _addr.IP.String() == addr {
+			return true
+		}
+	}
+
+	return false
 }
 
 func interfaceAddresses(ifi pcap.Interface) []string {
@@ -493,6 +640,14 @@ func interfaceAddresses(ifi pcap.Interface) []string {
 	return hosts
 }
 
+func interfaceIPs(ifi pcap.Interface) []net.IP {
+	var ips []net.IP
+	for _, addr := range ifi.Addresses {
+		ips = append(ips, addr.IP)
+	}
+	return ips
+}
+
 func listenAll(addr string) bool {
 	switch addr {
 	case "", "0.0.0.0", "[::]", "::":

capture/capture_test.go

@@ -1,313 +0,0 @@
-package capture
-
-import (
-	"context"
-	"encoding/binary"
-	"io/ioutil"
-	"net"
-	"os"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/google/gopacket/layers"
-	"github.com/google/gopacket/pcap"
-)
-
-var LoopBack = func() net.Interface {
-	ifis, _ := net.Interfaces()
-	for _, v := range ifis {
-		if v.Flags&net.FlagLoopback != 0 {
-			return v
-		}
-	}
-	return ifis[0]
-}()
-
-func TestSetInterfaces(t *testing.T) {
-	l := &Listener{}
-	l.host = "127.0.0.1"
-	l.setInterfaces()
-	if len(l.Interfaces) != 1 {
-		t.Error("expected a single interface")
-	}
-	l.host = LoopBack.HardwareAddr.String()
-	l.setInterfaces()
-	if l.Interfaces[0].Name != LoopBack.Name && len(l.Interfaces) != 1 {
-		t.Error("interface should be loop back interface")
-	}
-	l.host = ""
-	l.setInterfaces()
-	if len(l.Interfaces) < 1 {
-		t.Error("should get all interfaces")
-	}
-}
-
-func TestBPFFilter(t *testing.T) {
-	l := &Listener{}
-	l.host = "127.0.0.1"
-	l.Transport = "tcp"
-	l.setInterfaces()
-	filter := l.Filter(l.Interfaces[0])
-	if filter != "(tcp dst portrange 0-65535 and host 127.0.0.1)" {
-		t.Error("wrong filter", filter)
-	}
-	l.port = 8000
-	l.trackResponse = true
-	filter = l.Filter(l.Interfaces[0])
-	if filter != "(tcp port 8000 and host 127.0.0.1)" {
-		t.Error("wrong filter")
-	}
-}
-
-func TestPcapDump(t *testing.T) {
-	f, err := ioutil.TempFile("", "pcap_file")
-	if err != nil {
-		t.Error(err)
-	}
-	h, _ := PcapDumpHandler(f, layers.LinkTypeLoop)
-	packets := Packets(1, 5, 5, 4)
-	for i := 0; i < len(packets); i++ {
-		if i == 1 {
-			// change dst port
-			binary.BigEndian.PutUint16(packets[i].TransLayer[2:], 8001)
-		}
-		if i == 4 {
-			packets[i].Info.CaptureLength = 40
-		}
-		h(packets[i])
-	}
-	name := f.Name()
-	f.Close()
-	testPcapDumpEngine(name, t)
-}
-
-func testPcapDumpEngine(f string, t *testing.T) {
-	defer os.Remove(f)
-	l, err := NewListener(f, 8000, "", EnginePcapFile, true)
-	err = l.Activate()
-	if err != nil {
-		t.Errorf("expected error to be nil, got %q", err)
-		return
-	}
-	pckts := 0
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	err = l.Listen(ctx, func(packet *Packet) {
-		if packet.Info.CaptureLength != 57 {
-			t.Errorf("expected packet length to be %d, got %d", 57, packet.Info.CaptureLength)
-		}
-		pckts++
-	})
-
-	if err != nil {
-		t.Errorf("expected error to be nil, got %q", err)
-	}
-	if pckts != 3 {
-		t.Errorf("expected %d packets, got %d packets", 3, pckts)
-	}
-}
-
-func TestPcapHandler(t *testing.T) {
-	l, err := NewListener(LoopBack.Name, 8000, "", EnginePcap, true)
-	if err != nil {
-		t.Errorf("expected error to be nil, got %v", err)
-		return
-	}
-	err = l.Activate()
-	if err != nil {
-		t.Errorf("expected error to be nil, got %v", err)
-		return
-	}
-	defer l.Handles[LoopBack.Name].(*pcap.Handle).Close()
-	if err != nil {
-		t.Errorf("expected error to be nil, got %v", err)
-		return
-	}
-	for i := 0; i < 5; i++ {
-		_, _ = net.Dial("tcp", "127.0.0.1:8000")
-	}
-	sts, _ := l.Handles[LoopBack.Name].(*pcap.Handle).Stats()
-	if sts.PacketsReceived < 5 {
-		t.Errorf("expected >=5 packets got %d", sts.PacketsReceived)
-	}
-}
-
-// func TestSocketHandler(t *testing.T) {
-// 	l, err := NewListener(LoopBack.Name, 8000, "", EngineRawSocket, true)
-// 	err = l.Activate()
-// 	if err != nil {
-// 		return
-// 	}
-// 	defer l.Handles[LoopBack.Name].(*SockRaw).Close()
-// 	if err != nil {
-// 		t.Errorf("expected error to be nil, got %v", err)
-// 		return
-// 	}
-// 	for i := 0; i < 5; i++ {
-// 		_, _ = net.Dial("tcp", "127.0.0.1:8000")
-// 	}
-// 	sts, _ := l.Handles[LoopBack.Name].(*SockRaw).Stats()
-// 	if sts.Packets < 5 {
-// 		t.Errorf("expected >=5 packets got %d", sts.Packets)
-// 	}
-// }
-
-func BenchmarkPcapDump(b *testing.B) {
-	f, err := ioutil.TempFile("", "pcap_file")
-	if err != nil {
-		b.Error(err)
-		return
-	}
-	now := time.Now()
-	defer os.Remove(f.Name())
-	h, _ := PcapDumpHandler(f, layers.LinkTypeLoop)
-	packets := Packets(1, b.N, 5, 4)
-	for i := 0; i < len(packets); i++ {
-		h(packets[i])
-	}
-	f.Close()
-	b.Logf("%d packets in %s", b.N, time.Since(now))
-}
-
-func BenchmarkPcapFile(b *testing.B) {
-	f, err := ioutil.TempFile("", "pcap_file")
-	if err != nil {
-		b.Error(err)
-		return
-	}
-	defer os.Remove(f.Name())
-	h, _ := PcapDumpHandler(f, layers.LinkTypeLoop)
-	packets := Packets(1, b.N, 5, 4)
-	for i := 0; i < len(packets); i++ {
-		h(packets[i])
-	}
-	name := f.Name()
-	f.Close()
-	b.ResetTimer()
-	var l *Listener
-	l, err = NewListener(name, 8000, "", EnginePcapFile, true)
-	if err != nil {
-		b.Error(err)
-		return
-	}
-	err = l.Activate()
-	if err != nil {
-		b.Error(err)
-		return
-	}
-	now := time.Now()
-	pckts := 0
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	if err = l.Listen(ctx, func(packet *Packet) {
-		if packet.Info.CaptureLength != 49 {
-			b.Errorf("expected packet length to be %d, got %d", 49, packet.Info.CaptureLength)
-		}
-		pckts++
-	}); err != nil {
-		b.Error(err)
-	}
-	b.Logf("%d/%d packets in %s", pckts, b.N, time.Since(now))
-}
-
-// used to benchmark sock engine
-var buf [1024]byte
-
-func init() {
-	for i := 0; i < len(buf); i++ {
-		buf[i] = 0xff
-	}
-}
-
-func handler(n, counter *int32) PacketHandler {
-	return func(p *Packet) {
-		nn := int32(len(p.Data))
-		atomic.AddInt32(n, nn)
-		atomic.AddInt32(counter, 1)
-	}
-}
-
-func BenchmarkPcap(b *testing.B) {
-	var err error
-	n := new(int32)
-	counter := new(int32)
-	l, err := NewListener(LoopBack.Name, 8000, "", EnginePcap, false)
-	if err != nil {
-		b.Error(err)
-		return
-	}
-	l.PcapOptions.BPFFilter = "udp dst port 8000 and host 127.0.0.1"
-	err = l.Activate()
-	if err != nil {
-		b.Error(err)
-		return
-	}
-	errCh := l.ListenBackground(context.Background(), handler(n, counter))
-	select {
-	case <-l.Reading:
-	case err = <-errCh:
-		b.Error(err)
-		return
-	}
-	var conn net.Conn
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		b.StopTimer()
-		conn, err = net.Dial("udp", "127.0.0.1:8000")
-		if err != nil {
-			b.Error(err)
-			return
-		}
-		b.StartTimer()
-		_, err = conn.Write(buf[:])
-		if err != nil {
-			b.Error(err)
-			return
-		}
-	}
-	b.ReportMetric(float64(atomic.LoadInt32(n)), "buf")
-	b.ReportMetric(float64(atomic.LoadInt32(counter)), "packets")
-}
-
-func BenchmarkRawSocket(b *testing.B) {
-	var err error
-	n := new(int32)
-	counter := new(int32)
-	l, err := NewListener(LoopBack.Name, 8000, "", EngineRawSocket, false)
-	if err != nil {
-		b.Error(err)
-		return
-	}
-	l.PcapOptions.BPFFilter = "udp dst port 8000 and host 127.0.0.1"
-	err = l.Activate()
-	if err != nil {
-		b.Error(err)
-		return
-	}
-	errCh := l.ListenBackground(context.Background(), handler(n, counter))
-	select {
-	case <-l.Reading:
-	case err = <-errCh:
-		b.Error(err)
-		return
-	}
-	var conn net.Conn
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		b.StopTimer()
-		conn, err = net.Dial("udp", "127.0.0.1:8000")
-		if err != nil {
-			b.Error(err)
-			return
-		}
-		b.StartTimer()
-		_, err = conn.Write(buf[:])
-		if err != nil {
-			b.Error(err)
-			return
-		}
-	}
-	b.ReportMetric(float64(atomic.LoadInt32(n)), "buf")
-	b.ReportMetric(float64(atomic.LoadInt32(counter)), "packets")
-}

capture/packet_test.go

@@ -1,197 +0,0 @@
-package capture
-
-import (
-	"encoding/binary"
-	"errors"
-	"testing"
-	"time"
-
-	"github.com/google/gopacket"
-	"github.com/google/gopacket/layers"
-)
-
-func generateHeader4(seq uint32, length uint16) []byte {
-	hdr := make([]byte, 4+24+24, 4+24+24)
-	binary.BigEndian.PutUint32(hdr, uint32(layers.ProtocolFamilyIPv4))
-
-	ip := hdr[4:]
-	ip[0] = 4<<4 | 6
-	binary.BigEndian.PutUint16(ip[2:4], length+24+24)
-	ip[9] = uint8(layers.IPProtocolTCP)
-	copy(ip[12:16], []byte{127, 0, 0, 1})
-	copy(ip[16:], []byte{127, 0, 0, 1})
-
-	// set tcp header
-	tcp := ip[24:]
-	tcp[12] = 6 << 4
-	binary.BigEndian.PutUint16(tcp, 5535)
-	binary.BigEndian.PutUint16(tcp[2:], 8000)
-	binary.BigEndian.PutUint32(tcp[4:], seq)
-	return hdr
-}
-
-func generateHeader6(seq uint32, length uint16) []byte {
-	hdr := make([]byte, 4+40+32+24, 4+40+32+24)
-	binary.BigEndian.PutUint32(hdr, uint32(layers.ProtocolFamilyIPv6Linux))
-
-	ip := hdr[4:]
-	ip[0] = 6 << 4
-	binary.BigEndian.PutUint16(ip[4:], length+32+24)
-	var ipAddr [16]byte
-	ipAddr[15] = 0x01
-	copy(ip[8:], ip[:])
-	copy(ip[24:], ip[:])
-	copy(ip[40:], []byte{
-		// net-layer (IPv6-Opts)
-		0x2b, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-		uint8(layers.IPProtocolTCP), 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
-	})
-
-	// set tcp header
-	tcp := ip[40+32:]
-	tcp[12] = 6 << 4
-	binary.BigEndian.PutUint16(tcp, 5535)
-	binary.BigEndian.PutUint16(tcp[2:], 8000)
-	binary.BigEndian.PutUint32(tcp[4:], seq)
-	return hdr
-}
-
-func Packets(start uint32, _len int, length uint16, version byte) []*Packet {
-	var packets = make([]*Packet, _len)
-	for i := start; i < start+uint32(_len); i++ {
-		var h []byte
-		if version == 4 {
-			h = generateHeader4(i, length)
-		} else {
-			h = generateHeader6(i, length)
-		}
-		d := append(h, make([]byte, int(length))...)
-		ci := &gopacket.CaptureInfo{Length: len(d), CaptureLength: len(d), Timestamp: time.Now()}
-		packets[i-start] = NewPacket(d, int(layers.LinkTypeLoop), 4, ci)
-	}
-	return packets
-}
-
-func TestIPv4Packet(t *testing.T) {
-	pckt := packet(append(generateHeader4(1024, 10), make([]byte, 10)...))
-	if pckt.Err != nil {
-		t.Error(pckt)
-		return
-	}
-	if err := packet(pckt.Data[:2]).Err; !errors.Is(err, ErrHdrLength("Link")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("Link"), err)
-		return
-	}
-	if err := packet(pckt.Data[:20]).Err; !errors.Is(err, ErrHdrLength("IPv4")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv4"), err)
-		return
-	}
-	if err := packet(pckt.Data[:27]).Err; !errors.Is(err, ErrHdrLength("IPv4 opts")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv4 opts"), err)
-		return
-	}
-	if err := packet(pckt.Data[:40]).Err; !errors.Is(err, ErrHdrLength("TCP")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err)
-		return
-	}
-	if err := packet(pckt.Data[:50]).Err; !errors.Is(err, ErrHdrLength("TCP opts")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err)
-		return
-	}
-	pckt.TransLayer[12] = 0x10
-	if err := packet(pckt.Data[:50]).Err; !errors.Is(err, ErrHdrInvalid("TCP's data offset")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrInvalid("TCP's data offset"), err)
-		return
-	}
-	pckt.TransLayer[12] = 0x60
-	if err := packet(pckt.Data[:28]).Err; !errors.Is(err, ErrHdrMissing("TCP")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrMissing("TCP"), err)
-		return
-	}
-	pckt.NetLayer[9] = 0x02
-	if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("TCP")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrExpected("TCP"), err)
-		return
-	}
-	pckt.NetLayer[9] = 0x06
-	pckt.NetLayer[0] = 0x44
-	if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrInvalid("IPv4's IHL")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrInvalid("IPv4's IHL"), err)
-		return
-	}
-	pckt.NetLayer[0] = 0x56
-	if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("IPv4 or IPv6")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrExpected("IPv4 or IPv6"), err)
-		return
-	}
-}
-
-func TestIPv6Packet(t *testing.T) {
-	pckt := packet(append(generateHeader6(1024, 10), make([]byte, 10)...))
-	if pckt.Err != nil {
-		t.Error(pckt)
-		return
-	}
-	if err := packet(pckt.Data[:4]).Err; !errors.Is(err, ErrHdrMissing("IPv4 or IPv6")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrMissing("IPv4 or IPv6"), err)
-		return
-	}
-	if err := packet(pckt.Data[:40]).Err; !errors.Is(err, ErrHdrLength("IPv6")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv6"), err)
-		return
-	}
-	if err := packet(pckt.Data[:52]).Err; !errors.Is(err, ErrHdrLength("IPv6 opts")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv6 opts"), err)
-		return
-	}
-	if err := packet(pckt.Data[:80]).Err; !errors.Is(err, ErrHdrLength("TCP")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err)
-		return
-	}
-	if err := packet(pckt.Data[:98]).Err; !errors.Is(err, ErrHdrLength("TCP opts")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err)
-		return
-	}
-	pckt.TransLayer[12] = 0x10
-	if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrInvalid("TCP's data offset")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrInvalid("TCP's data offset"), err)
-		return
-	}
-	pckt.TransLayer[12] = 0x60
-	if err := packet(pckt.Data[:76]).Err; !errors.Is(err, ErrHdrMissing("TCP")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrMissing("TCP"), err)
-		return
-	}
-	pckt.NetLayer[56] = 0x02
-	if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("TCP")) {
-		t.Errorf("should fail with %q, got %q", ErrHdrExpected("TCP"), err)
-		return
-	}
-	pckt.NetLayer[56] = 0x06
-}
-
-func packet(data []byte) *Packet {
-	return NewPacket(data, int(layers.LinkTypeLoop), 4, &gopacket.CaptureInfo{})
-}
-
-func BenchmarkNewPacketIPv4(b *testing.B) {
-	data := append(generateHeader4(1204, 10), make([]byte, 10)...)
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		if err := packet(data).Err; err != nil {
-			b.Error(err)
-		}
-	}
-}
-
-func BenchmarkNewPacketIPv6(b *testing.B) {
-	data := append(generateHeader6(1024, 10), make([]byte, 10)...)
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		if err := packet(data).Err; err != nil {
-			b.Error(err)
-		}
-	}
-}

capture/sock_linux.go

@@ -1,3 +1,5 @@
+// +build linux
+
 package capture
 
 import (
@@ -58,7 +60,7 @@ func NewSocket(pifi pcap.Interface) (*SockRaw, error) {
 	}
 
 	if !found {
-		return nil, fmt.Errorf("Can't find matching interface")
+		return nil, fmt.Errorf("can't find matching interface")
 	}
 
 	// sock create
@@ -124,7 +126,7 @@ func NewSocket(pifi pcap.Interface) (*SockRaw, error) {
 }
 
 // ReadPacketData implements gopacket.PacketDataSource.
-func (sock *SockRaw) ZeroCopyReadPacketData() (buf []byte, ci gopacket.CaptureInfo, err error) {
+func (sock *SockRaw) ReadPacketData() (buf []byte, ci gopacket.CaptureInfo, err error) {
 	sock.mu.Lock()
 	defer sock.mu.Unlock()
 	var tpHdr *unix.Tpacket2Hdr

capture/socket.go

@@ -8,7 +8,7 @@ import (
 
 // Socket is any interface that defines the behaviors of Socket
 type Socket interface {
-	ZeroCopyReadPacketData() ([]byte, gopacket.CaptureInfo, error)
+	ReadPacketData() ([]byte, gopacket.CaptureInfo, error)
 	WritePacketData([]byte) error
 	SetBPFFilter(string) error
 	SetPromiscuous(bool) error

emitter.go

@@ -152,7 +152,7 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 				}
 			} else {
 				for _, dst := range writers {
-					if _, err := dst.PluginWrite(msg); err != nil {
+					if _, err := dst.PluginWrite(msg); err != nil && err != io.ErrClosedPipe {
 						return err
 					}
 				}

go.mod

@@ -8,13 +8,13 @@ require (
 	github.com/aws/aws-sdk-go v1.33.2
 	github.com/bitly/go-hostpool v0.1.0 // indirect
 	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
-	github.com/google/gopacket v1.1.18
+	github.com/google/gopacket v1.1.20-0.20210429153827-3eaba0894325
 	github.com/klauspost/compress v1.10.10 // indirect
 	github.com/mattbaird/elastigo v0.0.0-20170123220020-2fe47fd29e4b
 	github.com/pierrec/lz4 v2.5.2+incompatible // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 // indirect
 	github.com/smartystreets/goconvey v1.6.4 // indirect
 	github.com/stretchr/testify v1.5.1
-	golang.org/x/net v0.0.0-20200707034311-ab3426394381 // indirect
+	golang.org/x/net v0.0.0-20200707034311-ab3426394381
 	golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd
 )

go.sum

@@ -29,8 +29,8 @@ github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEW
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/gopacket v1.1.18 h1:lum7VRA9kdlvBi7/v2p7/zcbkduHaCH/SVVyurs7OpY=
-github.com/google/gopacket v1.1.18/go.mod h1:UdDNZ1OO62aGYVnPhxT1U6aI7ukYtA/kB8vaU0diBUM=
+github.com/google/gopacket v1.1.20-0.20210429153827-3eaba0894325 h1:YmIcZ5Var3BAQ64AW98Iiys5Ih4fiU0xK41+8isC5Ec=
+github.com/google/gopacket v1.1.20-0.20210429153827-3eaba0894325/go.mod h1:riddUzxTSBpJXk3qBHtYr4qOhFhT6k/1c0E3qkQjQpA=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE=
@@ -69,20 +69,29 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE=
+github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
+github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
+golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190405154228-4b34438f7a67/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200217220822-9197077df867/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd h1:xhmwyvizuTgC2qz7ZlMluP20uW+C3Rm0FD/WLDX8884=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -90,6 +99,8 @@ golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

gor.go

@@ -3,15 +3,16 @@
 package main
 
 import (
+	"expvar"
 	"flag"
+	"fmt"
 	"log"
 	"net/http"
 	"net/http/httputil"
-	_ "net/http/pprof"
+	httppptof "net/http/pprof"
 	"os"
 	"os/signal"
 	"runtime"
-	_ "runtime/debug"
 	"runtime/pprof"
 	"syscall"
 	"time"
@@ -22,6 +23,35 @@ var (
 	memprofile = flag.String("memprofile", "", "write memory profile to this file")
 )
 
+func init() {
+	var defaultServeMux http.ServeMux
+	http.DefaultServeMux = &defaultServeMux
+
+	http.HandleFunc("/debug/vars", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json; charset=utf-8")
+		fmt.Fprintf(w, "{\n")
+		first := true
+		expvar.Do(func(kv expvar.KeyValue) {
+			if kv.Key == "memstats" || kv.Key == "cmdline" {
+				return
+			}
+
+			if !first {
+				fmt.Fprintf(w, ",\n")
+			}
+			first = false
+			fmt.Fprintf(w, "%q: %s", kv.Key, kv.Value)
+		})
+		fmt.Fprintf(w, "\n}\n")
+	})
+
+	http.HandleFunc("/debug/pprof/", httppptof.Index)
+	http.HandleFunc("/debug/pprof/cmdline", httppptof.Cmdline)
+	http.HandleFunc("/debug/pprof/profile", httppptof.Profile)
+	http.HandleFunc("/debug/pprof/symbol", httppptof.Symbol)
+	http.HandleFunc("/debug/pprof/trace", httppptof.Trace)
+}
+
 func loggingMiddleware(addr string, next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.URL.Path == "/loop" {

input_file.go

@@ -4,9 +4,12 @@ import (
 	"bufio"
 	"bytes"
 	"compress/gzip"
+	"container/heap"
 	"errors"
+	"expvar"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -20,27 +23,68 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3"
 )
 
-type fileInputReader struct {
-	reader    *bufio.Reader
+type filePayload struct {
 	data      []byte
-	file      io.ReadCloser
 	timestamp int64
-	closed    int32 // Value of 0 indicates that the file is still open.
-	s3        bool
 }
 
-func (f *fileInputReader) parseNext() error {
+// An IntHeap is a min-heap of ints.
+type payloadQueue struct {
+	sync.RWMutex
+	s []*filePayload
+}
+
+func (h payloadQueue) Len() int           { return len(h.s) }
+func (h payloadQueue) Less(i, j int) bool { return h.s[i].timestamp < h.s[j].timestamp }
+func (h payloadQueue) Swap(i, j int)      { h.s[i], h.s[j] = h.s[j], h.s[i] }
+
+func (h *payloadQueue) Push(x interface{}) {
+	// Push and Pop use pointer receivers because they modify the slice's length,
+	// not just its contents.
+	h.s = append(h.s, x.(*filePayload))
+}
+
+func (h *payloadQueue) Pop() interface{} {
+	old := h.s
+	n := len(old)
+	x := old[n-1]
+	h.s = old[0 : n-1]
+	return x
+}
+
+func (h payloadQueue) Idx(i int) *filePayload {
+	return h.s[i]
+}
+
+type fileInputReader struct {
+	reader    *bufio.Reader
+	file      io.ReadCloser
+	closed    int32 // Value of 0 indicates that the file is still open.
+	s3        bool
+	queue     payloadQueue
+	readDepth int
+}
+
+func (f *fileInputReader) parse(init chan struct{}) error {
 	payloadSeparatorAsBytes := []byte(payloadSeparator)
 	var buffer bytes.Buffer
+	var initialized bool
+
 	for {
 		line, err := f.reader.ReadBytes('\n')
 
 		if err != nil {
 			if err != io.EOF {
 				Debug(1, err)
-			} else {
-				f.Close()
 			}
+
+			f.Close()
+
+			if !initialized {
+				close(init)
+				initialized = true
+			}
+
 			return err
 		}
 
@@ -48,21 +92,51 @@ func (f *fileInputReader) parseNext() error {
 			asBytes := buffer.Bytes()
 			meta := payloadMeta(asBytes)
 
-			f.timestamp, _ = strconv.ParseInt(string(meta[2]), 10, 64)
-			f.data = asBytes[:len(asBytes)-1]
+			timestamp, _ := strconv.ParseInt(string(meta[2]), 10, 64)
+			data := asBytes[:len(asBytes)-1]
 
-			return nil
+			f.queue.Lock()
+			heap.Push(&f.queue, &filePayload{
+				timestamp: timestamp,
+				data:      data,
+			})
+			f.queue.Unlock()
+
+			for {
+				if f.queue.Len() < f.readDepth {
+					break
+				}
+
+				if !initialized {
+					close(init)
+					initialized = true
+				}
+
+				time.Sleep(100 * time.Millisecond)
+			}
+
+			buffer = bytes.Buffer{}
+			continue
 		}
 
 		buffer.Write(line)
 	}
-
 }
 
-func (f *fileInputReader) ReadPayload() []byte {
-	defer f.parseNext()
+func (f *fileInputReader) wait() {
+	for {
+		if atomic.LoadInt32(&f.closed) == 1 {
+			return
+		}
 
-	return f.data
+		if f.queue.Len() > 0 {
+			return
+		}
+
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	return
 }
 
 // Close closes this plugin
@@ -75,7 +149,7 @@ func (f *fileInputReader) Close() error {
 	return nil
 }
 
-func newFileInputReader(path string) *fileInputReader {
+func newFileInputReader(path string, readDepth int) *fileInputReader {
 	var file io.ReadCloser
 	var err error
 
@@ -90,7 +164,7 @@ func newFileInputReader(path string) *fileInputReader {
 		return nil
 	}
 
-	r := &fileInputReader{file: file, closed: 0}
+	r := &fileInputReader{file: file, closed: 0, readDepth: readDepth}
 	if strings.HasSuffix(path, ".gz") {
 		gzReader, err := gzip.NewReader(file)
 		if err != nil {
@@ -102,7 +176,11 @@ func newFileInputReader(path string) *fileInputReader {
 		r.reader = bufio.NewReader(file)
 	}
 
-	r.parseNext()
+	heap.Init(&r.queue)
+
+	init := make(chan struct{})
+	go r.parse(init)
+	<-init
 
 	return r
 }
@@ -116,16 +194,25 @@ type FileInput struct {
 	readers     []*fileInputReader
 	speedFactor float64
 	loop        bool
+	readDepth   int
+	dryRun      bool
+	maxWait     time.Duration
+
+	stats *expvar.Map
 }
 
 // NewFileInput constructor for FileInput. Accepts file path as argument.
-func NewFileInput(path string, loop bool) (i *FileInput) {
+func NewFileInput(path string, loop bool, readDepth int, maxWait time.Duration, dryRun bool) (i *FileInput) {
 	i = new(FileInput)
 	i.data = make(chan []byte, 1000)
 	i.exit = make(chan bool)
 	i.path = path
 	i.speedFactor = 1
 	i.loop = loop
+	i.readDepth = readDepth
+	i.stats = expvar.NewMap("file-" + path)
+	i.dryRun = dryRun
+	i.maxWait = maxWait
 
 	if err := i.init(); err != nil {
 		return
@@ -165,7 +252,6 @@ func (i *FileInput) init() (err error) {
 	} else if matches, err = filepath.Glob(i.path); err != nil {
 		Debug(0, "[INPUT-FILE] Wrong file pattern", i.path, err)
 		return
-
 	}
 
 	if len(matches) == 0 {
@@ -176,9 +262,11 @@ func (i *FileInput) init() (err error) {
 	i.readers = make([]*fileInputReader, len(matches))
 
 	for idx, p := range matches {
-		i.readers[idx] = newFileInputReader(p)
+		i.readers[idx] = newFileInputReader(p, i.readDepth)
 	}
 
+	i.stats.Add("reader_count", int64(len(matches)))
+
 	return nil
 }
 
@@ -189,6 +277,7 @@ func (i *FileInput) PluginRead() (*Message, error) {
 	case <-i.exit:
 		return nil, ErrorStopped
 	case buf := <-i.data:
+		i.stats.Add("read_from", 1)
 		msg.Meta, msg.Data = payloadMetaWithBody(buf)
 		return &msg, nil
 	}
@@ -201,11 +290,17 @@ func (i *FileInput) String() string {
 // Find reader with smallest timestamp e.g next payload in row
 func (i *FileInput) nextReader() (next *fileInputReader) {
 	for _, r := range i.readers {
-		if r == nil || atomic.LoadInt32(&r.closed) != 0 {
+		if r == nil {
 			continue
 		}
 
-		if next == nil || r.timestamp < next.timestamp {
+		r.wait()
+
+		if r.queue.Len() == 0 {
+			continue
+		}
+
+		if next == nil || r.queue.Idx(0).timestamp < next.queue.Idx(0).timestamp {
 			next = r
 			continue
 		}
@@ -217,6 +312,11 @@ func (i *FileInput) nextReader() (next *fileInputReader) {
 func (i *FileInput) emit() {
 	var lastTime int64 = -1
 
+	var maxWait, firstWait, minWait int64
+	minWait = math.MaxInt64
+
+	i.stats.Add("negative_wait", 0)
+
 	for {
 		select {
 		case <-i.exit:
@@ -236,17 +336,48 @@ func (i *FileInput) emit() {
 			}
 		}
 
+		reader.queue.RLock()
+		payload := heap.Pop(&reader.queue).(*filePayload)
+		i.stats.Add("total_counter", 1)
+		i.stats.Add("total_bytes", int64(len(payload.data)))
+		reader.queue.RUnlock()
+
 		if lastTime != -1 {
-			diff := reader.timestamp - lastTime
-			lastTime = reader.timestamp
+			diff := payload.timestamp - lastTime
+
+			if firstWait == 0 {
+				firstWait = diff
+			}
 
 			if i.speedFactor != 1 {
 				diff = int64(float64(diff) / i.speedFactor)
 			}
 
-			time.Sleep(time.Duration(diff))
+			if i.maxWait > 0 && diff > int64(i.maxWait) {
+				diff = int64(i.maxWait)
+			}
+
+			if diff >= 0 {
+				lastTime = payload.timestamp
+
+				if !i.dryRun {
+					time.Sleep(time.Duration(diff))
+				}
+
+				i.stats.Add("total_wait", diff)
+
+				if diff > maxWait {
+					maxWait = diff
+				}
+
+				if diff < minWait {
+					minWait = diff
+				}
+			} else {
+				i.stats.Add("negative_wait", 1)
+			}
 		} else {
-			lastTime = reader.timestamp
+			lastTime = payload.timestamp
 		}
 
 		// Recheck if we have exited since last check.
@@ -254,12 +385,30 @@ func (i *FileInput) emit() {
 		case <-i.exit:
 			return
 		default:
-			i.data <- reader.ReadPayload()
+			if !i.dryRun {
+				i.data <- payload.data
+			}
 		}
 	}
 
+	i.stats.Set("first_wait", time.Duration(firstWait))
+	i.stats.Set("max_wait", time.Duration(maxWait))
+	i.stats.Set("min_wait", time.Duration(minWait))
+
 	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",
+			i.stats.Get("total_counter"),
+			i.stats.Get("reader_count"),
+			i.stats.Get("total_bytes"),
+			i.stats.Get("max_wait"),
+			i.stats.Get("min_wait"),
+			i.stats.Get("first_wait"),
+			time.Duration(i.stats.Get("total_wait").(*expvar.Int).Value()),
+			i.stats.Get("negative_wait"),
+		)
+	}
 }
 
 // Close closes this plugin

input_file_test.go

@@ -104,7 +104,7 @@ func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) {
 	file2.Write([]byte(payloadSeparator))
 	file2.Close()
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
 
 	for i := '1'; i <= '4'; i++ {
 		msg, _ := input.PluginRead()
@@ -130,7 +130,7 @@ func TestInputFileRequestsWithLatency(t *testing.T) {
 	file.Write([]byte("1 3 250000000\nrequest3"))
 	file.Write([]byte(payloadSeparator))
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100, 0, false)
 
 	start := time.Now().UnixNano()
 	for i := 0; i < 3; i++ {
@@ -170,7 +170,7 @@ func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) {
 	file2.Write([]byte(payloadSeparator))
 	file2.Close()
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
 
 	for i := '1'; i <= '4'; i++ {
 		msg, _ := input.PluginRead()
@@ -198,7 +198,7 @@ func TestInputFileLoop(t *testing.T) {
 	file.Write([]byte(payloadSeparator))
 	file.Close()
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100, 0, false)
 
 	// Even if we have just 2 requests in file, it should indifinitly loop
 	for i := 0; i < 1000; i++ {
@@ -226,7 +226,7 @@ func TestInputFileCompressed(t *testing.T) {
 	name2 := output2.file.Name()
 	output2.Close()
 
-	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
+	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
 	for i := 0; i < 2000; i++ {
 		input.PluginRead()
 	}
@@ -326,7 +326,7 @@ func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile {
 func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback) (err error) {
 	wg := new(sync.WaitGroup)
 
-	input := NewFileInput(captureFile.Name(), false)
+	input := NewFileInput(captureFile.Name(), false, 100, 0, false)
 	output := NewTestOutput(func(msg *Message) {
 		callback(msg)
 		wg.Done()

input_raw.go

@@ -16,53 +16,20 @@ import (
 	"github.com/buger/goreplay/tcp"
 )
 
-// TCPProtocol is a number to indicate type of protocol
-type TCPProtocol uint8
-
-const (
-	// ProtocolHTTP ...
-	ProtocolHTTP TCPProtocol = iota
-	// ProtocolBinary ...
-	ProtocolBinary
-)
-
-// Set is here so that TCPProtocol can implement flag.Var
-func (protocol *TCPProtocol) Set(v string) error {
-	switch v {
-	case "", "http":
-		*protocol = ProtocolHTTP
-	case "binary":
-		*protocol = ProtocolBinary
-	default:
-		return fmt.Errorf("unsupported protocol %s", v)
-	}
-	return nil
-}
-
-func (protocol *TCPProtocol) String() string {
-	switch *protocol {
-	case ProtocolBinary:
-		return "binary"
-	case ProtocolHTTP:
-		return "http"
-	default:
-		return ""
-	}
-}
-
 // RAWInputConfig represents configuration that can be applied on raw input
 type RAWInputConfig struct {
 	capture.PcapOptions
-	Expire         time.Duration      `json:"input-raw-expire"`
-	CopyBufferSize size.Size          `json:"copy-buffer-size"`
-	Engine         capture.EngineType `json:"input-raw-engine"`
-	TrackResponse  bool               `json:"input-raw-track-response"`
-	Protocol       TCPProtocol        `json:"input-raw-protocol"`
-	RealIPHeader   string             `json:"input-raw-realip-header"`
-	Stats          bool               `json:"input-raw-stats"`
-	quit           chan bool          // Channel used only to indicate goroutine should shutdown
-	host           string
-	ports          []uint16
+	Expire          time.Duration      `json:"input-raw-expire"`
+	CopyBufferSize  size.Size          `json:"copy-buffer-size"`
+	Engine          capture.EngineType `json:"input-raw-engine"`
+	TrackResponse   bool               `json:"input-raw-track-response"`
+	Protocol        tcp.TCPProtocol    `json:"input-raw-protocol"`
+	RealIPHeader    string             `json:"input-raw-realip-header"`
+	Stats           bool               `json:"input-raw-stats"`
+	AllowIncomplete bool               `json:"input-raw-allow-incomplete"`
+	quit            chan bool          // Channel used only to indicate goroutine should shutdown
+	host            string
+	ports           []uint16
 }
 
 // RAWInput used for intercepting traffic for given address
@@ -71,7 +38,7 @@ type RAWInput struct {
 	RAWInputConfig
 	messageStats   []tcp.Stats
 	listener       *capture.Listener
-	message        chan *tcp.Message
+	messageParser  *tcp.MessageParser
 	cancelListener context.CancelFunc
 	closed         bool
 }
@@ -80,7 +47,6 @@ type RAWInput struct {
 func NewRAWInput(address string, config RAWInputConfig) (i *RAWInput) {
 	i = new(RAWInput)
 	i.RAWInputConfig = config
-	i.message = make(chan *tcp.Message, 10000)
 	i.quit = make(chan bool)
 
 	host, _ports, err := net.SplitHostPort(address)
@@ -117,11 +83,12 @@ func (i *RAWInput) PluginRead() (*Message, error) {
 	select {
 	case <-i.quit:
 		return nil, ErrorStopped
-	case msgTCP = <-i.message:
+	case msgTCP = <-i.listener.Messages():
 		msg.Data = msgTCP.Data()
 	}
+
 	var msgType byte = ResponsePayload
-	if msgTCP.IsRequest {
+	if msgTCP.Direction == tcp.DirIncoming {
 		msgType = RequestPayload
 		if i.RealIPHeader != "" {
 			msg.Data = proto.SetHeader(msg.Data, []byte(i.RealIPHeader), []byte(msgTCP.SrcAddr))
@@ -141,14 +108,13 @@ func (i *RAWInput) PluginRead() (*Message, error) {
 		stat := msgTCP.Stats
 		go i.addStats(stat)
 	}
-	msgTCP.Finalize()
 	msgTCP = nil
 	return &msg, nil
 }
 
 func (i *RAWInput) listen(address string) {
 	var err error
-	i.listener, err = capture.NewListener(i.host, i.ports, "", i.Engine, i.TrackResponse)
+	i.listener, err = capture.NewListener(i.host, i.ports, "", i.Engine, i.Protocol, i.TrackResponse, i.Expire, i.AllowIncomplete)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -157,15 +123,10 @@ func (i *RAWInput) listen(address string) {
 	if err != nil {
 		log.Fatal(err)
 	}
-	parser := tcp.NewMessageParser(i.CopyBufferSize, i.Expire, Debug, i.messageEmitter)
 
-	if i.Protocol == ProtocolHTTP {
-		parser.Start = http1StartHint
-		parser.End = http1EndHint
-	}
 	var ctx context.Context
 	ctx, i.cancelListener = context.WithCancel(context.Background())
-	errCh := i.listener.ListenBackground(ctx, parser.PacketHandler)
+	errCh := i.listener.ListenBackground(ctx)
 	<-i.listener.Reading
 	Debug(1, i)
 	go func() {
@@ -174,10 +135,6 @@ func (i *RAWInput) listen(address string) {
 	}()
 }
 
-func (i *RAWInput) messageEmitter(m *tcp.Message) {
-	i.message <- m
-}
-
 func (i *RAWInput) String() string {
 	return fmt.Sprintf("Intercepting traffic from: %s:%s", i.host, strings.Join(strings.Fields(fmt.Sprint(i.ports)), ","))
 }
@@ -213,24 +170,3 @@ func (i *RAWInput) addStats(mStats tcp.Stats) {
 	i.messageStats = append(i.messageStats, mStats)
 	i.Unlock()
 }
-
-func http1StartHint(pckt *tcp.Packet) (isRequest, isResponse bool) {
-	if proto.HasRequestTitle(pckt.Payload) {
-		return true, false
-	}
-
-	if proto.HasResponseTitle(pckt.Payload) {
-		return false, true
-	}
-
-	// No request or response detected
-	return false, false
-}
-
-func http1EndHint(m *tcp.Message) bool {
-	if m.MissingChunk() {
-		return false
-	}
-
-	return proto.HasFullPayload(m, m.PacketData()...)
-}

input_raw_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/buger/goreplay/capture"
 	"github.com/buger/goreplay/proto"
+	"github.com/buger/goreplay/tcp"
 )
 
 const testRawExpire = time.Millisecond * 200
@@ -43,7 +44,7 @@ func TestRAWInputIPv4(t *testing.T) {
 	conf := RAWInputConfig{
 		Engine:        capture.EnginePcap,
 		Expire:        0,
-		Protocol:      ProtocolHTTP,
+		Protocol:      tcp.ProtocolHTTP,
 		TrackResponse: true,
 		RealIPHeader:  "X-Real-IP",
 	}
@@ -58,6 +59,7 @@ func TestRAWInputIPv4(t *testing.T) {
 		} else {
 			respCounter++
 		}
+
 		wg.Done()
 	})
 
@@ -71,14 +73,18 @@ func TestRAWInputIPv4(t *testing.T) {
 	emitter := NewEmitter()
 	defer emitter.Close()
 	go emitter.Start(plugins, Settings.Middleware)
+
+	// time.Sleep(time.Second)
 	for i := 0; i < 1; i++ {
 		wg.Add(2)
 		_, err = http.Get(addr)
+
 		if err != nil {
 			t.Error(err)
 			return
 		}
 	}
+
 	wg.Wait()
 	const want = 10
 	if reqCounter != respCounter && reqCounter != want {
@@ -108,7 +114,7 @@ func TestRAWInputNoKeepAlive(t *testing.T) {
 	conf := RAWInputConfig{
 		Engine:        capture.EnginePcap,
 		Expire:        testRawExpire,
-		Protocol:      ProtocolHTTP,
+		Protocol:      tcp.ProtocolHTTP,
 		TrackResponse: true,
 	}
 	input := NewRAWInput(":"+port, conf)
@@ -173,7 +179,7 @@ func TestRAWInputIPv6(t *testing.T) {
 	var respCounter, reqCounter int64
 	conf := RAWInputConfig{
 		Engine:        capture.EnginePcap,
-		Protocol:      ProtocolHTTP,
+		Protocol:      tcp.ProtocolHTTP,
 		TrackResponse: true,
 	}
 	input := NewRAWInput(originAddr, conf)
@@ -228,10 +234,11 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
 
 	originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
 	conf := RAWInputConfig{
-		Engine:        capture.EnginePcap,
-		Expire:        time.Second,
-		Protocol:      ProtocolHTTP,
-		TrackResponse: true,
+		Engine:          capture.EnginePcap,
+		Expire:          time.Second,
+		Protocol:        tcp.ProtocolHTTP,
+		TrackResponse:   true,
+		AllowIncomplete: true,
 	}
 	input := NewRAWInput(originAddr, conf)
 
@@ -309,7 +316,7 @@ func BenchmarkRAWInputWithReplay(b *testing.B) {
 	conf := RAWInputConfig{
 		Engine:        capture.EnginePcap,
 		Expire:        testRawExpire,
-		Protocol:      ProtocolHTTP,
+		Protocol:      tcp.ProtocolHTTP,
 		TrackResponse: true,
 	}
 	input := NewRAWInput(originAddr, conf)

output_dummy.go

@@ -24,6 +24,7 @@ func (i *DummyOutput) PluginWrite(msg *Message) (int, error) {
 	n += nn
 	nn, err = os.Stdout.Write(payloadSeparatorAsBytes)
 	n += nn
+
 	return n, err
 }
 

output_file.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"runtime/debug"
@@ -19,6 +20,21 @@ import (
 	"github.com/buger/goreplay/size"
 )
 
+var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
+var instanceID string
+
+func init() {
+	instanceID = randSeq(8)
+}
+
+func randSeq(n int) string {
+	b := make([]rune, n)
+	for i := range b {
+		b[i] = letters[rand.Intn(len(letters))]
+	}
+	return string(b)
+}
+
 var dateFileNameFuncs = map[string]func(*FileOutput) string{
 	"%Y":  func(o *FileOutput) string { return time.Now().Format("2006") },
 	"%m":  func(o *FileOutput) string { return time.Now().Format("01") },
@@ -29,6 +45,7 @@ var dateFileNameFuncs = map[string]func(*FileOutput) string{
 	"%NS": func(o *FileOutput) string { return fmt.Sprint(time.Now().Nanosecond()) },
 	"%r":  func(o *FileOutput) string { return string(o.currentID) },
 	"%t":  func(o *FileOutput) string { return string(o.payloadType) },
+	"%i":  func(o *FileOutput) string { return instanceID },
 }
 
 // FileOutputConfig ...
@@ -45,17 +62,17 @@ type FileOutputConfig struct {
 // FileOutput output plugin
 type FileOutput struct {
 	sync.RWMutex
-	pathTemplate   string
-	currentName    string
-	file           *os.File
-	QueueLength    int
-	chunkSize      int
-	writer         io.Writer
-	requestPerFile bool
-	currentID      []byte
-	payloadType    []byte
-	closed         bool
-	totalFileSize  size.Size
+	pathTemplate    string
+	currentName     string
+	file            *os.File
+	QueueLength     int
+	writer          io.Writer
+	requestPerFile  bool
+	currentID       []byte
+	payloadType     []byte
+	closed          bool
+	currentFileSize int
+	totalFileSize   size.Size
 
 	config *FileOutputConfig
 }
@@ -157,7 +174,7 @@ func (o *FileOutput) filename() string {
 
 		if o.currentName == "" ||
 			((o.config.QueueLimit > 0 && o.QueueLength >= o.config.QueueLimit) ||
-				(o.config.SizeLimit > 0 && o.chunkSize >= int(o.config.SizeLimit))) {
+				(o.config.SizeLimit > 0 && o.currentFileSize >= int(o.config.SizeLimit))) {
 			nextChunk = true
 		}
 
@@ -178,6 +195,7 @@ func (o *FileOutput) filename() string {
 
 				if nextChunk {
 					fileIndex++
+					o.currentFileSize = 0
 				}
 			}
 
@@ -236,6 +254,7 @@ func (o *FileOutput) PluginWrite(msg *Message) (n int, err error) {
 	n += nn
 
 	o.totalFileSize += size.Size(n)
+	o.currentFileSize += n
 	o.QueueLength++
 
 	if Settings.OutputFileConfig.OutputFileMaxSize > 0 && o.totalFileSize >= Settings.OutputFileConfig.OutputFileMaxSize {
@@ -264,7 +283,7 @@ func (o *FileOutput) flush() {
 		}
 
 		if stat, err := o.file.Stat(); err == nil {
-			o.chunkSize = int(stat.Size())
+			o.currentFileSize = int(stat.Size())
 		} else {
 			Debug(0, "[OUTPUT-HTTP] error accessing file size", err)
 		}

output_file_test.go

@@ -39,7 +39,7 @@ func TestFileOutput(t *testing.T) {
 	emitter.Close()
 
 	var counter int64
-	input2 := NewFileInput("/tmp/test_requests.gor", false)
+	input2 := NewFileInput("/tmp/test_requests.gor", false, 100, 0, false)
 	output2 := NewTestOutput(func(*Message) {
 		atomic.AddInt64(&counter, 1)
 		wg.Done()

output_http.go

@@ -215,6 +215,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, msg *Message) {
 	if !isRequestPayload(msg.Meta) {
 		return
 	}
+
 	uuid := payloadID(msg.Meta)
 	start := time.Now()
 	resp, err := client.Send(msg.Data)

output_tcp.go

@@ -95,16 +95,15 @@ func (o *TCPOutput) worker(bufferIndex int) {
 	}
 }
 
-func (o *TCPOutput) getBufferIndex(data []byte) int {
+func (o *TCPOutput) getBufferIndex(msg *Message) int {
 	if !o.config.Sticky {
 		o.workerIndex++
 		return int(o.workerIndex) % o.config.Workers
 	}
 
 	hasher := fnv.New32a()
-	hasher.Write(payloadMeta(data)[1])
+	hasher.Write(payloadID(msg.Meta))
 	return int(hasher.Sum32()) % o.config.Workers
-
 }
 
 // PluginWrite writes message to this plugin
@@ -113,7 +112,7 @@ func (o *TCPOutput) PluginWrite(msg *Message) (n int, err error) {
 		return len(msg.Data), nil
 	}
 
-	bufferIndex := o.getBufferIndex(msg.Data)
+	bufferIndex := o.getBufferIndex(msg)
 	o.buf[bufferIndex] <- msg
 
 	if Settings.OutputTCPStats {

output_tcp_test.go

@@ -125,8 +125,9 @@ func TestBufferDistribution(t *testing.T) {
 	}
 }
 
-func getTestBytes() []byte {
-	reqh := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1)
-	reqb := append(reqh, []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n")...)
-	return reqb
+func getTestBytes() *Message {
+	return &Message{
+		Meta: payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1),
+		Data: []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n"),
+	}
 }

plugins.go

@@ -83,7 +83,6 @@ func (plugins *InOutPlugins) registerPlugin(constructor interface{}, options ...
 		plugins.Outputs = append(plugins.Outputs, w)
 	}
 	plugins.All = append(plugins.All, plugin)
-
 }
 
 // NewPlugins specify and initialize all available plugins
@@ -119,7 +118,7 @@ func NewPlugins() *InOutPlugins {
 	}
 
 	for _, options := range Settings.InputFile {
-		plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop)
+		plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth, Settings.InputFileMaxWait, Settings.InputFileDryRun)
 	}
 
 	for _, path := range Settings.OutputFile {

ring/ring.go

@@ -0,0 +1,205 @@
+package ring
+
+import (
+	"errors"
+	"runtime"
+	"sync/atomic"
+	"time"
+)
+
+var (
+	// ErrDisposed is returned when an operation is performed on a disposed
+	// queue.
+	ErrDisposed = errors.New(`queue: disposed`)
+
+	// ErrTimeout is returned when an applicable queue operation times out.
+	ErrTimeout = errors.New(`queue: poll timed out`)
+
+	// ErrEmptyQueue is returned when an non-applicable queue operation was called
+	// due to the queue's empty item state
+	ErrEmptyQueue = errors.New(`queue: empty queue`)
+)
+
+// roundUp takes a uint64 greater than 0 and rounds it up to the next
+// power of 2.
+func roundUp(v uint64) uint64 {
+	v--
+	v |= v >> 1
+	v |= v >> 2
+	v |= v >> 4
+	v |= v >> 8
+	v |= v >> 16
+	v |= v >> 32
+	v++
+	return v
+}
+
+type node struct {
+	position uint64
+	data     interface{}
+}
+
+type nodes []node
+
+// RingBuffer is a MPMC buffer that achieves threadsafety with CAS operations
+// only.  A put on full or get on empty call will block until an item
+// is put or retrieved.  Calling Dispose on the RingBuffer will unblock
+// any blocked threads with an error.  This buffer is similar to the buffer
+// described here: http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
+// with some minor additions.
+type RingBuffer struct {
+	_padding0      [8]uint64
+	queue          uint64
+	_padding1      [8]uint64
+	dequeue        uint64
+	_padding2      [8]uint64
+	mask, disposed uint64
+	_padding3      [8]uint64
+	nodes          nodes
+}
+
+func (rb *RingBuffer) init(size uint64) {
+	size = roundUp(size)
+	rb.nodes = make(nodes, size)
+	for i := uint64(0); i < size; i++ {
+		rb.nodes[i] = node{position: i}
+	}
+	rb.mask = size - 1 // so we don't have to do this with every put/get operation
+}
+
+// Put adds the provided item to the queue.  If the queue is full, this
+// call will block until an item is added to the queue or Dispose is called
+// on the queue.  An error will be returned if the queue is disposed.
+func (rb *RingBuffer) Put(item interface{}) error {
+	_, err := rb.put(item, false)
+	return err
+}
+
+// Offer adds the provided item to the queue if there is space.  If the queue
+// is full, this call will return false.  An error will be returned if the
+// queue is disposed.
+func (rb *RingBuffer) Offer(item interface{}) (bool, error) {
+	return rb.put(item, true)
+}
+
+func (rb *RingBuffer) put(item interface{}, offer bool) (bool, error) {
+	var n *node
+	pos := atomic.LoadUint64(&rb.queue)
+L:
+	for {
+		if atomic.LoadUint64(&rb.disposed) == 1 {
+			return false, ErrDisposed
+		}
+
+		n = &rb.nodes[pos&rb.mask]
+		seq := atomic.LoadUint64(&n.position)
+		switch dif := seq - pos; {
+		case dif == 0:
+			if atomic.CompareAndSwapUint64(&rb.queue, pos, pos+1) {
+				break L
+			}
+		case dif < 0:
+			panic(`Ring buffer in a compromised state during a put operation.`)
+		default:
+			pos = atomic.LoadUint64(&rb.queue)
+		}
+
+		if offer {
+			return false, nil
+		}
+
+		runtime.Gosched() // free up the cpu before the next iteration
+	}
+
+	n.data = item
+	atomic.StoreUint64(&n.position, pos+1)
+	return true, nil
+}
+
+// Get will return the next item in the queue.  This call will block
+// if the queue is empty.  This call will unblock when an item is added
+// to the queue or Dispose is called on the queue.  An error will be returned
+// if the queue is disposed.
+func (rb *RingBuffer) Get() (interface{}, error) {
+	return rb.Poll(0)
+}
+
+// Poll will return the next item in the queue.  This call will block
+// if the queue is empty.  This call will unblock when an item is added
+// to the queue, Dispose is called on the queue, or the timeout is reached. An
+// error will be returned if the queue is disposed or a timeout occurs. A
+// non-positive timeout will block indefinitely.
+func (rb *RingBuffer) Poll(timeout time.Duration) (interface{}, error) {
+	var (
+		n     *node
+		pos   = atomic.LoadUint64(&rb.dequeue)
+		start time.Time
+	)
+	if timeout > 0 {
+		start = time.Now()
+	}
+L:
+	for {
+		if atomic.LoadUint64(&rb.disposed) == 1 {
+			return nil, ErrDisposed
+		}
+
+		n = &rb.nodes[pos&rb.mask]
+		seq := atomic.LoadUint64(&n.position)
+		switch dif := seq - (pos + 1); {
+		case dif == 0:
+			if atomic.CompareAndSwapUint64(&rb.dequeue, pos, pos+1) {
+				break L
+			}
+		case dif < 0:
+			panic(`Ring buffer in compromised state during a get operation.`)
+		default:
+			pos = atomic.LoadUint64(&rb.dequeue)
+		}
+
+		if timeout > 0 && time.Since(start) >= timeout {
+			return nil, ErrTimeout
+		}
+
+		if timeout < 0 {
+			return nil, ErrTimeout
+		}
+
+		runtime.Gosched() // free up the cpu before the next iteration
+	}
+	data := n.data
+	n.data = nil
+	atomic.StoreUint64(&n.position, pos+rb.mask+1)
+	return data, nil
+}
+
+// Len returns the number of items in the queue.
+func (rb *RingBuffer) Len() uint64 {
+	return atomic.LoadUint64(&rb.queue) - atomic.LoadUint64(&rb.dequeue)
+}
+
+// Cap returns the capacity of this ring buffer.
+func (rb *RingBuffer) Cap() uint64 {
+	return uint64(len(rb.nodes))
+}
+
+// Dispose will dispose of this queue and free any blocked threads
+// in the Put and/or Get methods.  Calling those methods on a disposed
+// queue will return an error.
+func (rb *RingBuffer) Dispose() {
+	atomic.CompareAndSwapUint64(&rb.disposed, 0, 1)
+}
+
+// IsDisposed will return a bool indicating if this queue has been
+// disposed.
+func (rb *RingBuffer) IsDisposed() bool {
+	return atomic.LoadUint64(&rb.disposed) == 1
+}
+
+// NewRingBuffer will allocate, initialize, and return a ring buffer
+// with the specified size.
+func NewRingBuffer(size uint64) *RingBuffer {
+	rb := &RingBuffer{}
+	rb.init(size)
+	return rb
+}

s3_test.go

@@ -127,7 +127,7 @@ func TestInputFileFromS3(t *testing.T) {
 		<-output.closeCh
 	}
 
-	input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false)
+	input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false, 100, 0, false)
 
 	buf := make([]byte, 1000)
 	for i := 0; i <= 19999; i++ {

settings.go

@@ -45,10 +45,13 @@ type AppSettings struct {
 	OutputTCPConfig TCPOutputConfig
 	OutputTCPStats  bool `json:"output-tcp-stats"`
 
-	InputFile        MultiOption `json:"input-file"`
-	InputFileLoop    bool        `json:"input-file-loop"`
-	OutputFile       MultiOption `json:"output-file"`
-	OutputFileConfig FileOutputConfig
+	InputFile          MultiOption   `json:"input-file"`
+	InputFileLoop      bool          `json:"input-file-loop"`
+	InputFileReadDepth int           `json:"input-file-read-depth"`
+	InputFileDryRun    bool          `json:"input-file-dry-run"`
+	InputFileMaxWait   time.Duration `json:"input-file-max-wait"`
+	OutputFile         MultiOption   `json:"output-file"`
+	OutputFileConfig   FileOutputConfig
 
 	InputRAW MultiOption `json:"input_raw"`
 	RAWInputConfig
@@ -113,6 +116,9 @@ func init() {
 
 	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.")
+	flag.IntVar(&Settings.InputFileReadDepth, "input-file-read-depth", 100, "GoReplay tries to read and cache multiple records, in advance. In parallel it also perform sorting of requests, if they came out of order. Since it needs hold this buffer in memory, bigger values can cause worse performance")
+	flag.BoolVar(&Settings.InputFileDryRun, "input-file-dry-run", false, "Simulate reading from the data source without replaying it. You will get information about expected replay time, number of found records etc.")
+	flag.DurationVar(&Settings.InputFileMaxWait, "input-file-max-wait", 0, "Set the maximum time between requests. Can help in situations when you have too long periods between request, and you want to skip them. Example: --input-raw-max-wait 1s")
 
 	flag.Var(&Settings.OutputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor")
 	flag.DurationVar(&Settings.OutputFileConfig.FlushInterval, "output-file-flush-interval", time.Second, "Interval for forcing buffer flush to the file, default: 1s.")
@@ -141,6 +147,7 @@ func init() {
 	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.AllowIncomplete, "input-raw-allow-incomplete", false, "If turned on Gor will record HTTP messages with missing packets")
 
 	flag.StringVar(&Settings.Middleware, "middleware", "", "Used for modifying traffic using external command")
 
@@ -187,7 +194,7 @@ func init() {
 	flag.StringVar(&Settings.KafkaTLSConfig.ClientCert, "kafka-tls-client-cert", "", "Client certificate for Kafka TLS Config (mandatory with to kafka-tls-ca-cert and kafka-tls-client-key)")
 	flag.StringVar(&Settings.KafkaTLSConfig.ClientKey, "kafka-tls-client-key", "", "Client Key for Kafka TLS Config (mandatory with to kafka-tls-client-cert and kafka-tls-client-key)")
 
-	flag.Var(&Settings.ModifierConfig.Headers, "http-set-header", "Inject additional headers to http reqest:\n\tgor --input-raw :8080 --output-http staging.com --http-set-header 'User-Agent: Gor'")
+	flag.Var(&Settings.ModifierConfig.Headers, "http-set-header", "Inject additional headers to http request:\n\tgor --input-raw :8080 --output-http staging.com --http-set-header 'User-Agent: Gor'")
 	flag.Var(&Settings.ModifierConfig.HeaderRewrite, "http-rewrite-header", "Rewrite the request header based on a mapping:\n\tgor --input-raw :8080 --output-http staging.com --http-rewrite-header Host: (.*).example.com,$1.beta.example.com")
 	flag.Var(&Settings.ModifierConfig.Params, "http-set-param", "Set request url param, if param already exists it will be overwritten:\n\tgor --input-raw :8080 --output-http staging.com --http-set-param api_key=1")
 	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")

simpletime/time.go

@@ -0,0 +1,17 @@
+package simpletime
+
+import (
+	"time"
+)
+
+var Now time.Time
+
+func init() {
+	go func() {
+		for {
+			// Accurate enough
+			Now = time.Now()
+			time.Sleep(100 * time.Millisecond)
+		}
+	}()
+}

tcp/tcp_message.go

@@ -3,13 +3,49 @@ package tcp
 import (
 	"encoding/binary"
 	"encoding/hex"
-	_ "fmt"
+	"fmt"
+	"net"
+	"reflect"
 	"sort"
+	"sync"
 	"time"
-
-	"github.com/buger/goreplay/size"
+	"unsafe"
 )
 
+// TCPProtocol is a number to indicate type of protocol
+type TCPProtocol uint8
+
+const (
+	// ProtocolHTTP ...
+	ProtocolHTTP TCPProtocol = iota
+	// ProtocolBinary ...
+	ProtocolBinary
+)
+
+// Set is here so that TCPProtocol can implement flag.Var
+func (protocol *TCPProtocol) Set(v string) error {
+	switch v {
+	case "", "http":
+		*protocol = ProtocolHTTP
+	case "binary":
+		*protocol = ProtocolBinary
+	default:
+		return fmt.Errorf("unsupported protocol %s", v)
+	}
+	return nil
+}
+
+func (protocol *TCPProtocol) String() string {
+	switch *protocol {
+	case ProtocolBinary:
+		return "binary"
+	case ProtocolHTTP:
+		return "http"
+	default:
+		return ""
+	}
+}
+
 // Stats every message carry its own stats object
 type Stats struct {
 	LostData  int
@@ -18,7 +54,7 @@ type Stats struct {
 	End       time.Time // last packet's timestamp
 	SrcAddr   string
 	DstAddr   string
-	IsRequest bool
+	Direction Dir
 	TimedOut  bool // timeout before getting the whole message
 	Truncated bool // last packet truncated due to max message size
 	IPversion byte
@@ -29,6 +65,7 @@ type Message struct {
 	packets  []*Packet
 	parser   *MessageParser
 	feedback interface{}
+	Idx      uint16
 	Stats
 }
 
@@ -38,7 +75,7 @@ func (m *Message) UUID() []byte {
 	pckt := m.packets[0]
 
 	// check if response or request have generated the ID before.
-	if m.IsRequest {
+	if m.Direction == DirIncoming {
 		streamID = uint64(pckt.SrcPort)<<48 | uint64(pckt.DstPort)<<32 |
 			uint64(ip2int(pckt.SrcIP))
 	} else {
@@ -49,7 +86,7 @@ func (m *Message) UUID() []byte {
 	id := make([]byte, 12)
 	binary.BigEndian.PutUint64(id, streamID)
 
-	if m.IsRequest {
+	if m.Direction == DirIncoming {
 		binary.BigEndian.PutUint32(id[8:], pckt.Ack)
 	} else {
 		binary.BigEndian.PutUint32(id[8:], pckt.Seq)
@@ -61,13 +98,11 @@ func (m *Message) UUID() []byte {
 	return uuidHex
 }
 
-func (m *Message) add(packet *Packet) {
-	// fmt.Println("SEQ:", packet.Seq, " - ", len(packet.Payload))
-
+func (m *Message) add(packet *Packet) bool {
 	// Skip duplicates
 	for _, p := range m.packets {
 		if p.Seq == packet.Seq {
-			return
+			return false
 		}
 	}
 
@@ -91,6 +126,8 @@ func (m *Message) add(packet *Packet) {
 	if packet.Timestamp.After(m.End) || m.End.IsZero() {
 		m.End = packet.Timestamp
 	}
+
+	return true
 }
 
 // Packets returns packets of the message
@@ -124,15 +161,11 @@ func (m *Message) PacketData() [][]byte {
 
 // Data returns data in this message
 func (m *Message) Data() []byte {
-	var totalLen int
-	for _, p := range m.packets {
-		totalLen += len(p.Payload)
-	}
-	tmp := make([]byte, totalLen)
+	packetData := m.PacketData()
+	tmp := packetData[0]
 
-	var i int
-	for _, p := range m.packets {
-		i += copy(tmp[i:], p.Payload)
+	if len(packetData) > 0 {
+		tmp, _ = copySlice(tmp, len(packetData[0]), packetData[1:]...)
 	}
 
 	return tmp
@@ -153,20 +186,9 @@ func (m *Message) Sort() {
 	sort.SliceStable(m.packets, func(i, j int) bool { return m.packets[i].Seq < m.packets[j].Seq })
 }
 
-func (m *Message) Finalize() {
-	// Allow re-use memory
-	for _, p := range m.packets {
-		packetPool.Put(p)
-	}
-}
-
 // Emitter message handler
 type Emitter func(*Message)
 
-// Debugger is the debugger function. first params is the indicator of the issue's priority
-// the higher the number, the lower the priority. it can be 4 <= level <= 6.
-type Debugger func(int, ...interface{})
-
 // HintEnd hints the parser to stop the session, see MessageParser.End
 // when set, it will be executed before checking FIN or RST flag
 type HintEnd func(*Message) bool
@@ -178,139 +200,223 @@ 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 {
-	debug         Debugger
-	maxSize       size.Size // maximum message size, default 5mb
-	m             map[uint64]*Message
-	emit          Emitter
-	messageExpire time.Duration // the maximum time to wait for the final packet, minimum is 100ms
-	End           HintEnd
-	Start         HintStart
-	ticker        *time.Ticker
-	packets       chan *Packet
-	msgs          int32         // messages in the parser
-	close         chan struct{} // to signal that we are able to close
+	m  []map[uint64]*Message
+	mL []sync.RWMutex
+
+	messageExpire  time.Duration // the maximum time to wait for the final packet, minimum is 100ms
+	allowIncompete bool
+	End            HintEnd
+	Start          HintStart
+	ticker         *time.Ticker
+	messages       chan *Message
+	packets        chan *PcapPacket
+	close          chan struct{} // to signal that we are able to close
+	ports          []uint16
+	ips            []net.IP
 }
 
 // NewMessageParser returns a new instance of message parser
-func NewMessageParser(maxSize size.Size, messageExpire time.Duration, debugger Debugger, emitHandler Emitter) (parser *MessageParser) {
+func NewMessageParser(messages chan *Message, ports []uint16, ips []net.IP, messageExpire time.Duration, allowIncompete bool) (parser *MessageParser) {
 	parser = new(MessageParser)
-	parser.debug = debugger
-	parser.emit = emitHandler
-	parser.messageExpire = time.Millisecond * 100
-	if parser.messageExpire < messageExpire {
-		parser.messageExpire = messageExpire
+
+	parser.messageExpire = messageExpire
+	if parser.messageExpire == 0 {
+		parser.messageExpire = time.Millisecond * 1000
 	}
-	parser.maxSize = maxSize
-	if parser.maxSize < 1 {
-		parser.maxSize = 5 << 20
+
+	parser.allowIncompete = allowIncompete
+
+	parser.packets = make(chan *PcapPacket, 10000)
+
+	if messages == nil {
+		messages = make(chan *Message, 100)
 	}
-	parser.packets = make(chan *Packet, 1000)
-	parser.m = make(map[uint64]*Message)
-	parser.ticker = time.NewTicker(time.Millisecond * 50)
+	parser.messages = messages
+	parser.ticker = time.NewTicker(time.Millisecond * 100)
 	parser.close = make(chan struct{}, 1)
-	go parser.wait()
+
+	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)
+	}
+
 	return parser
 }
 
+var packetLen int
+
 // Packet returns packet handler
-func (parser *MessageParser) PacketHandler(packet *Packet) {
+func (parser *MessageParser) PacketHandler(packet *PcapPacket) {
+	packetLen++
 	parser.packets <- packet
 }
 
-func (parser *MessageParser) wait() {
+func (parser *MessageParser) wait(index int) {
 	var (
-		pckt *Packet
-		now  time.Time
+		now time.Time
 	)
 	for {
 		select {
-		case pckt = <-parser.packets:
-			parser.processPacket(pckt)
+		case pckt := <-parser.packets:
+			parser.processPacket(parser.parsePacket(pckt))
 		case now = <-parser.ticker.C:
-			parser.timer(now)
+			parser.timer(now, index)
 		case <-parser.close:
 			parser.ticker.Stop()
 			// parser.Close should wait for this function to return
 			parser.close <- struct{}{}
 			return
+			// default:
 		}
 	}
 }
 
+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)
+		return nil
+	}
+
+	for _, p := range parser.ports {
+		if pckt.DstPort == p {
+			for _, ip := range parser.ips {
+				if pckt.DstIP.Equal(ip) {
+					pckt.Direction = DirIncoming
+					break
+				}
+			}
+			break
+		}
+	}
+
+	return pckt
+}
+
 func (parser *MessageParser) processPacket(pckt *Packet) {
-	var in, out bool
+	if pckt == nil {
+		return
+	}
 
 	// 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
-	m, ok := parser.m[pckt.MessageID()]
+	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()
+		}
+	}
+
 	switch {
 	case ok:
 		parser.addPacket(m, pckt)
-		return
-	case parser.Start != nil:
-		if in, out = parser.Start(pckt); !(in || out) {
-			// Packet can be received out of order, so give it another chance
-			if pckt.Retry < 1 && len(pckt.Payload) > 0 {
-				// Requeue not known packets
-				pckt.Retry++
 
-				select {
-				case parser.packets <- pckt:
-				default:
-					packetPool.Put(pckt)
-					// fmt.Println("Skipping packet")
-				}
+		parser.mL[mIDX].Unlock()
+		return
+	case pckt.Direction == DirUnknown && parser.Start != nil:
+		if in, out := parser.Start(pckt); in || out {
+			if in {
+				pckt.Direction = DirIncoming
+			} else {
+				pckt.Direction = DirOutcoming
 			}
-			return
 		}
 	}
 
+	if pckt.Direction == DirIncoming {
+		mIDX = pckt.SrcPort % 10
+	} else {
+		mIDX = pckt.DstPort % 10
+	}
+
+	parser.mL[mIDX].Lock()
+
 	m = new(Message)
-	m.IsRequest = in
-	parser.m[pckt.MessageID()] = m
+	m.Direction = pckt.Direction
+
+	parser.m[mIDX][mID] = m
+
+	m.Idx = mIDX
 	m.Start = pckt.Timestamp
 	m.parser = parser
 	parser.addPacket(m, pckt)
+
+	parser.mL[mIDX].Unlock()
 }
 
-func (parser *MessageParser) addPacket(m *Message, pckt *Packet) {
-	trunc := m.Length + len(pckt.Payload) - int(parser.maxSize)
-	if trunc > 0 {
-		m.Truncated = true
-		pckt.Payload = pckt.Payload[:int(parser.maxSize)-m.Length]
-	}
-	m.add(pckt)
-	switch {
-	// if one of this cases matches, we dispatch the message
-	case trunc >= 0:
-	case parser.End != nil && parser.End(m):
-	default:
-		// continue to receive packets
-		return
+func (parser *MessageParser) addPacket(m *Message, pckt *Packet) bool {
+	if !m.add(pckt) {
+		return false
 	}
 
-	parser.Emit(m)
-}
-
-func (parser *MessageParser) Emit(m *Message) {
-	delete(parser.m, m.packets[0].MessageID())
-	parser.emit(m)
-}
-
-func (parser *MessageParser) timer(now time.Time) {
-	for _, m := range parser.m {
-		if now.Sub(m.End) > parser.messageExpire {
-			m.TimedOut = true
+	// If we are using protocol parsing, like HTTP, depend on its parsing func.
+	// For the binary procols wait for message to expire
+	if parser.End != nil {
+		if parser.End(m) {
 			parser.Emit(m)
 		}
 	}
+
+	return true
 }
 
-// this function should not block other parser operations
-func (parser *MessageParser) Debug(level int, args ...interface{}) {
-	if parser.debug != nil {
-		parser.debug(level, args...)
+func (parser *MessageParser) Read() *Message {
+	m := <-parser.messages
+	return m
+}
+
+func (parser *MessageParser) Emit(m *Message) {
+	stats.Add("message_count", 1)
+
+	delete(parser.m[m.Idx], m.packets[0].MessageID())
+
+	parser.messages <- m
+}
+
+func GetUnexportedField(field reflect.Value) interface{} {
+	return reflect.NewAt(field.Type(), unsafe.Pointer(field.UnsafeAddr())).Elem().Interface()
+}
+
+var failMsg int
+
+func (parser *MessageParser) timer(now time.Time, index int) {
+	packetLen = 0
+	parser.mL[index].Lock()
+
+	packetQueueLen.Set(int64(len(parser.packets)))
+	messageQueueLen.Set(int64(len(parser.m[index])))
+
+	for _, m := range parser.m[index] {
+		if now.Sub(m.End) > parser.messageExpire {
+			m.TimedOut = true
+			stats.Add("message_timeout_count", 1)
+			failMsg++
+			if parser.End == nil || parser.allowIncompete {
+				parser.Emit(m)
+			}
+
+			delete(parser.m[index], m.packets[0].MessageID())
+		}
 	}
+
+	parser.mL[index].Unlock()
 }
 
 func (parser *MessageParser) Close() error {

tcp/tcp_packet.go

@@ -2,28 +2,54 @@ package tcp
 
 import (
 	"encoding/binary"
+	"expvar"
 	"fmt"
 	"net"
-	"sync"
 	"time"
 
 	"github.com/google/gopacket"
 )
 
-func copySlice(b, a []byte) []byte {
-	if cap(b) < len(a) {
-		b = make([]byte, len(a))
+func copySlice(to []byte, skip int, from ...[]byte) ([]byte, int) {
+	var totalLen int
+	for _, s := range from {
+		totalLen += len(s)
 	}
-	copy(b, a)
-	return b[:len(a)]
+	totalLen += skip
+
+	if cap(to) < totalLen {
+		diff := totalLen - cap(to)
+		to = append(to, make([]byte, diff)...)
+	}
+
+	for _, s := range from {
+		skip += copy(to[skip:], s)
+	}
+
+	return to, skip
 }
 
-var packetPool = sync.Pool{
-	New: func() interface{} {
-		return new(Packet)
-	},
+var stats *expvar.Map
+var packetQueueLen, messageQueueLen *expvar.Int
+
+func init() {
+	packetQueueLen = new(expvar.Int)
+	messageQueueLen = new(expvar.Int)
+
+	stats = expvar.NewMap("tcp")
+	stats.Init()
+	stats.Set("packet_queue", packetQueueLen)
+	stats.Set("message_queue", messageQueueLen)
 }
 
+type Dir int
+
+const (
+	DirUnknown = iota
+	DirIncoming
+	DirOutcoming
+)
+
 /*
 Packet represent data and layers of packet.
 parser extracts information from pcap Packet. functions of *Packet doesn't validate if packet is nil,
@@ -31,6 +57,7 @@ calllers must make sure that ParsePacket has'nt returned any error before callin
 function.
 */
 type Packet struct {
+	Direction          Dir
 	messageID          uint64
 	SrcIP, DstIP       net.IP
 	Version            uint8
@@ -39,24 +66,45 @@ type Packet struct {
 	ACK, SYN, FIN, RST bool
 	Lost               uint32
 	Retry              int
+	CaptureLength      int
 	Timestamp          time.Time
 	Payload            []byte
+	buf                []byte
+
+	created time.Time
+	gc      bool
+}
+
+type PcapPacket struct {
+	Data     []byte
+	LType    int
+	LTypeLen int
+	Ci       *gopacket.CaptureInfo
 }
 
 // ParsePacket parse raw packets
-func ParsePacket(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo) (pckt *Packet, err error) {
-	pckt = packetPool.Get().(*Packet)
+func ParsePacket(data []byte, lType, lTypeLen int, ci *gopacket.CaptureInfo, allowEmpty bool) (pckt *Packet, err error) {
+	pckt = new(Packet)
+	if err := pckt.parse(data, lType, lTypeLen, ci, allowEmpty); err != nil {
+		return nil, err
+	}
+
+	return pckt, nil
+}
+
+func (pckt *Packet) parse(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo, allowEmpty bool) error {
 	pckt.Retry = 0
 	pckt.messageID = 0
+	pckt.buf = pckt.buf[:]
 
 	// TODO: check resolution
 	pckt.Timestamp = cp.Timestamp
 
 	if len(data) < lTypeLen {
-		return nil, ErrHdrLength("Link")
+		return ErrHdrLength("Link")
 	}
 	if len(data) <= lTypeLen {
-		return nil, ErrHdrMissing("IPv4 or IPv6")
+		return ErrHdrMissing("IPv4 or IPv6")
 	}
 
 	ldata := data[lTypeLen:]
@@ -66,63 +114,63 @@ func ParsePacket(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo) (pc
 	if ldata[0]>>4 == 4 {
 		// IPv4 header
 		if len(ldata) < 20 {
-			return nil, ErrHdrLength("IPv4")
+			return ErrHdrLength("IPv4")
 		}
 		proto = ldata[9]
 		ihl := int(ldata[0]&0x0F) * 4
 		if ihl < 20 {
-			return nil, ErrHdrInvalid("IPv4's IHL")
+			return ErrHdrInvalid("IPv4's IHL")
 		}
 		if len(ldata) < ihl {
-			return nil, ErrHdrLength("IPv4 opts")
+			return ErrHdrLength("IPv4 opts")
 		}
 		netLayer = ldata[:ihl]
 	} else if ldata[0]>>4 == 6 {
 		if len(ldata) < 40 {
-			return nil, ErrHdrLength("IPv6")
+			return ErrHdrLength("IPv6")
 		}
 		proto = ldata[6]
 		totalLen := 40
 		for ipv6ExtensionHdr(proto) {
 			hdr := len(ldata) - totalLen
 			if hdr < 8 {
-				return nil, ErrHdrExpected("IPv6 opts")
+				return ErrHdrExpected("IPv6 opts")
 			}
 			extLen := 8
 			if proto != 44 {
 				extLen = int(ldata[totalLen+1]+1) * 8
 			}
 			if hdr < extLen {
-				return nil, ErrHdrLength("IPv6 opts")
+				return ErrHdrLength("IPv6 opts")
 			}
 			proto = ldata[totalLen]
 			totalLen += extLen
 		}
 		netLayer = ldata[:totalLen]
 	} else {
-		return nil, ErrHdrExpected("IPv4 or IPv6")
+		return ErrHdrExpected("IPv4 or IPv6")
 	}
 	if proto != 6 {
-		return nil, ErrHdrExpected("TCP")
+		return ErrHdrExpected("TCP")
 	}
 	if len(data) <= len(netLayer) {
-		return nil, ErrHdrMissing("TCP")
+		return ErrHdrMissing("TCP")
 	}
 	ndata := ldata[len(netLayer):]
 	// TCP header
 	if len(ndata) < 20 {
-		return nil, ErrHdrLength("TCP")
+		return ErrHdrLength("TCP")
 	}
 	dOf := int(ndata[12]>>4) * 4
 	if dOf < 20 {
-		return nil, ErrHdrInvalid("TCP's ndata offset")
+		return ErrHdrInvalid("TCP's ndata offset")
 	}
 	if len(ndata) < dOf {
-		return nil, ErrHdrLength("TCP opts")
+		return ErrHdrLength("TCP opts")
 	}
 
-	if len(ndata[dOf:]) == 0 {
-		return nil, fmt.Errorf("Packet without Data")
+	if !allowEmpty && len(ndata[dOf:]) == 0 {
+		return EmptyPacket("")
 	}
 
 	if (netLayer[0] >> 4) == 4 {
@@ -139,6 +187,7 @@ func ParsePacket(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo) (pc
 
 	transLayer = ndata[:dOf]
 
+	pckt.CaptureLength = cp.CaptureLength
 	pckt.SrcPort = binary.BigEndian.Uint16(transLayer[0:2])
 	pckt.DstPort = binary.BigEndian.Uint16(transLayer[2:4])
 	pckt.Seq = binary.BigEndian.Uint32(transLayer[4:8])
@@ -148,8 +197,10 @@ func ParsePacket(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo) (pc
 	pckt.RST = transLayer[13]&0x04 != 0
 	pckt.ACK = transLayer[13]&0x10 != 0
 	pckt.Lost = uint32(cp.Length - cp.CaptureLength)
-	pckt.Payload = copySlice(pckt.Payload, ndata[dOf:])
-	return
+
+	pckt.Payload = ndata[dOf:]
+
+	return nil
 }
 
 func (pckt *Packet) MessageID() uint64 {
@@ -172,6 +223,12 @@ func (pckt *Packet) Dst() string {
 	return fmt.Sprintf("%s:%d", pckt.DstIP, pckt.DstPort)
 }
 
+type EmptyPacket string
+
+func (err EmptyPacket) Error() string {
+	return "Empty packet"
+}
+
 // ErrHdrLength returned on short header length
 type ErrHdrLength string
 

tcp/tcp_test.go

@@ -43,14 +43,19 @@ func generateHeader(request bool, seq uint32, length uint16) []byte {
 
 func GetPackets(request bool, start uint32, _len int, payload []byte) []*Packet {
 	var packets = make([]*Packet, _len)
+	var err error
 	for i := start; i < start+uint32(_len); i++ {
 		d := append(generateHeader(request, i, uint16(len(payload))), payload...)
 		ci := &gopacket.CaptureInfo{Length: len(d), CaptureLength: len(d), Timestamp: time.Now()}
 
-		if len(payload) > 0 {
-			packets[i-start], _ = ParsePacket(d, int(layers.LinkTypeLoop), 4, ci)
+		packets[i-start], err = ParsePacket(d, int(layers.LinkTypeLoop), 4, ci, true)
+		if request {
+			packets[i-start].Direction = DirIncoming
 		} else {
-			packets[i-start] = new(Packet)
+			packets[i-start].Direction = DirOutcoming
+		}
+		if err != nil {
+			panic(err)
 		}
 	}
 	return packets
@@ -58,50 +63,44 @@ func GetPackets(request bool, start uint32, _len int, payload []byte) []*Packet
 
 func TestRequestResponseMapping(t *testing.T) {
 	packets := []*Packet{
-		{SrcPort: 60000, DstPort: 80, Ack: 1, Seq: 1, Timestamp: time.Unix(1, 0), Payload: []byte("GET / HTTP/1.1\r\n")},
-		{SrcPort: 60000, DstPort: 80, Ack: 1, Seq: 17, Timestamp: time.Unix(2, 0), Payload: []byte("Host: localhost\r\n\r\n")},
+		{SrcPort: 60000, DstPort: 80, Ack: 1, Seq: 1, Direction: DirIncoming, Timestamp: time.Unix(1, 0), Payload: []byte("GET / HTTP/1.1\r\n")},
+		{SrcPort: 60000, DstPort: 80, Ack: 1, Seq: 17, Direction: DirIncoming, Timestamp: time.Unix(2, 0), Payload: []byte("Host: localhost\r\n\r\n")},
 
 		// Seq of first response packet match Ack of first request packet
-		{SrcPort: 80, DstPort: 60000, Ack: 36, Seq: 1, Timestamp: time.Unix(3, 0), Payload: []byte("HTTP/1.1 200 OK\r\n")},
-		{SrcPort: 80, DstPort: 60000, Ack: 36, Seq: 18, Timestamp: time.Unix(4, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
+		{SrcPort: 80, DstPort: 60000, Ack: 36, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(3, 0), Payload: []byte("HTTP/1.1 200 OK\r\n")},
+		{SrcPort: 80, DstPort: 60000, Ack: 36, Seq: 18, Direction: DirOutcoming, Timestamp: time.Unix(4, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
 
 		// Same TCP stream
-		{SrcPort: 60000, DstPort: 80, Ack: 39, Seq: 36, Timestamp: time.Unix(5, 0), Payload: []byte("GET / HTTP/1.1\r\n")},
-		{SrcPort: 60000, DstPort: 80, Ack: 39, Seq: 52, Timestamp: time.Unix(6, 0), Payload: []byte("Host: localhost\r\n\r\n")},
+		{SrcPort: 60000, DstPort: 80, Ack: 39, Seq: 36, Direction: DirIncoming, Timestamp: time.Unix(5, 0), Payload: []byte("GET / HTTP/1.1\r\n")},
+		{SrcPort: 60000, DstPort: 80, Ack: 39, Seq: 52, Direction: DirIncoming, Timestamp: time.Unix(6, 0), Payload: []byte("Host: localhost\r\n\r\n")},
 
 		// Seq of first response packet match Ack of first request packet
-		{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 39, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\n")},
-		{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 56, Timestamp: time.Unix(8, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
+		{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 39, Direction: DirOutcoming, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\n")},
+		{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 56, Direction: DirOutcoming, Timestamp: time.Unix(8, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
 	}
 
-	var mssg = make(chan *Message, 4)
-	parser := NewMessageParser(1<<20, time.Second, nil, func(m *Message) { mssg <- m })
+	parser := NewMessageParser(nil, nil, nil, time.Second, false)
 	parser.Start = func(pckt *Packet) (bool, bool) {
 		return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
 	}
 	parser.End = func(m *Message) bool {
-		return proto.HasFullPayload(m, m.Data())
+		return proto.HasFullPayload(m, m.PacketData()...)
 	}
 
 	for _, packet := range packets {
-		parser.PacketHandler(packet)
+		parser.processPacket(packet)
 	}
 
 	messages := []*Message{}
 	for i := 0; i < 4; i++ {
-		select {
-		case <-time.After(time.Second):
-			t.Errorf("can't parse packets fast enough")
-			return
-		case m := <-mssg:
-			messages = append(messages, m)
-		}
+		m := parser.Read()
+		messages = append(messages, m)
 	}
 
-	assert.Equal(t, messages[0].IsRequest, true)
-	assert.Equal(t, messages[1].IsRequest, false)
-	assert.Equal(t, messages[2].IsRequest, true)
-	assert.Equal(t, messages[3].IsRequest, false)
+	assert.Equal(t, int(messages[0].Direction), int(DirIncoming))
+	assert.Equal(t, int(messages[1].Direction), int(DirOutcoming))
+	assert.Equal(t, int(messages[2].Direction), int(DirIncoming))
+	assert.Equal(t, int(messages[3].Direction), int(DirOutcoming))
 
 	assert.Equal(t, messages[0].UUID(), messages[1].UUID())
 	assert.Equal(t, messages[2].UUID(), messages[3].UUID())
@@ -110,36 +109,35 @@ func TestRequestResponseMapping(t *testing.T) {
 }
 
 func TestMessageParserWithHint(t *testing.T) {
-	var mssg = make(chan *Message, 3)
-	parser := NewMessageParser(1<<20, time.Second, nil, func(m *Message) { mssg <- m })
+	parser := NewMessageParser(nil, nil, nil, time.Second, false)
 	parser.Start = func(pckt *Packet) (bool, bool) {
 		return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
 	}
 	parser.End = func(m *Message) bool {
-		return proto.HasFullPayload(m, m.Data())
+		return proto.HasFullPayload(m, m.PacketData()...)
 	}
-	packets := GetPackets(true, 1, 30, nil)
-	packets[4] = GetPackets(false, 4, 1, []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7"))[0]
-	packets[5] = GetPackets(false, 5, 1, []byte("\r\nMozilla\r\n9\r\nDeveloper\r"))[0]
-	packets[6] = GetPackets(false, 6, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0]
-	packets[14] = GetPackets(true, 14, 1, []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n"))[0]
-	packets[15] = GetPackets(true, 15, 1, []byte("MozillaDeveloper"))[0]
-	packets[16] = GetPackets(true, 16, 1, []byte("Network"))[0]
-	packets[24] = GetPackets(true, 24, 1, []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\n\r\n"))[0]
 
-	for i := 0; i < 30; i++ {
-		parser.PacketHandler(packets[i])
+	packets := []*Packet{
+		// Seq of first response packet match Ack of first request packet
+		{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(1, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7\r\n")},
+		{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 18, Direction: DirOutcoming, Timestamp: time.Unix(2, 0), Payload: []byte("\r\nMozilla\r\n9\r\nDeveloper\r")},
+		{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 42, Direction: DirOutcoming, Timestamp: time.Unix(3, 0), Payload: []byte("\n7\r\nNetwork\r\n0\r\n\r\n")},
+
+		{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 1, Direction: DirIncoming, Timestamp: time.Unix(4, 0), Payload: []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n")},
+		{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 66, Direction: DirIncoming, Timestamp: time.Unix(5, 0), Payload: []byte("MozillaDeveloper")},
+		{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 82, Direction: DirIncoming, Timestamp: time.Unix(6, 0), Payload: []byte("Network")},
+
+		{SrcPort: 80, DstPort: 60000, Ack: 89, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\n\r\n")},
+	}
+
+	for _, p := range packets {
+		parser.processPacket(p)
 	}
 
 	messages := []*Message{}
 	for i := 0; i < 3; i++ {
-		select {
-		case <-time.After(time.Second):
-			t.Errorf("can't parse packets fast enough")
-			return
-		case m := <-mssg:
-			messages = append(messages, m)
-		}
+		m := parser.Read()
+		messages = append(messages, m)
 	}
 
 	if !bytes.HasSuffix(messages[0].Data(), []byte("\n7\r\nNetwork\r\n0\r\n\r\n")) {
@@ -156,118 +154,72 @@ func TestMessageParserWithHint(t *testing.T) {
 }
 
 func TestMessageParserWrongOrder(t *testing.T) {
-	var mssg = make(chan *Message, 3)
-	parser := NewMessageParser(1<<20, time.Second, nil, func(m *Message) { mssg <- m })
+	parser := NewMessageParser(nil, nil, nil, time.Second, false)
 	parser.Start = func(pckt *Packet) (bool, bool) {
 		return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
 	}
 	parser.End = func(m *Message) bool {
-		return proto.HasFullPayload(m, m.Data())
+		return proto.HasFullPayload(m, m.PacketData()...)
 	}
-	packets := GetPackets(true, 1, 30, nil)
-	packets[6] = GetPackets(false, 4, 1, []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7"))[0]
-	packets[5] = GetPackets(false, 5, 1, []byte("\r\nMozilla\r\n9\r\nDeveloper\r"))[0]
-	packets[4] = GetPackets(false, 6, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0]
-	// Duplicate with same seq
-	packets[7] = GetPackets(false, 6, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0]
+	packets := []*Packet{
+		// Seq of first response packet match Ack of first request packet
+		{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 66, Direction: DirIncoming, Timestamp: time.Unix(5, 0), Payload: []byte("MozillaDeveloper")},
+		{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(1, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7\r\n")},
+		{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 42, Direction: DirOutcoming, Timestamp: time.Unix(3, 0), Payload: []byte("\n7\r\nNetwork\r\n0\r\n\r\n")},
 
-	packets[16] = GetPackets(true, 14, 1, []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n"))[0]
-	packets[15] = GetPackets(true, 15, 1, []byte("MozillaDeveloper"))[0]
-	packets[14] = GetPackets(true, 16, 1, []byte("Network"))[0]
+		{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 1, Direction: DirIncoming, Timestamp: time.Unix(4, 0), Payload: []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n")},
+		{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 18, Direction: DirOutcoming, Timestamp: time.Unix(2, 0), Payload: []byte("\r\nMozilla\r\n9\r\nDeveloper\r")},
 
-	for i := 0; i < 30; i++ {
-		parser.PacketHandler(packets[i])
+		{SrcPort: 80, DstPort: 60000, Ack: 89, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\n\r\n")},
+		{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 82, Direction: DirIncoming, Timestamp: time.Unix(6, 0), Payload: []byte("Network")},
 	}
-	var m *Message
-	select {
-	case <-time.After(time.Second):
-		t.Errorf("can't parse packets fast enough")
-		return
-	case m = <-mssg:
+
+	for _, p := range packets {
+		parser.processPacket(p)
 	}
+
+	m := parser.Read()
+
 	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"))
 	}
 
-	select {
-	case <-time.After(time.Second):
-		t.Errorf("can't parse packets fast enough")
-		return
-	case m = <-mssg:
+	m = parser.Read()
+
+	if !bytes.HasSuffix(m.Data(), []byte("Content-Length: 0\r\n\r\n")) {
+		t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("Content-Length: 0\r\n\r\n"))
 	}
+
+	m = parser.Read()
+
 	if !bytes.HasSuffix(m.Data(), []byte("Network")) {
 		t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("Network"))
 	}
 }
 
 func TestMessageParserWithoutHint(t *testing.T) {
-	var mssg = make(chan *Message, 1)
 	var data [63 << 10]byte
 	packets := GetPackets(true, 1, 10, data[:])
 
-	p := NewMessageParser(63<<10*10, time.Second, nil, func(m *Message) { mssg <- m })
+	p := NewMessageParser(nil, nil, nil, time.Second, false)
 	for _, v := range packets {
-		p.PacketHandler(v)
-	}
-	var m *Message
-	select {
-	case <-time.After(time.Second):
-		t.Errorf("can't parse packets fast enough")
-		return
-	case m = <-mssg:
+		p.processPacket(v)
 	}
+	m := p.Read()
+
 	if m.Length != 63<<10*10 {
 		t.Errorf("expected %d to equal %d", m.Length, 63<<10*10)
 	}
 }
 
-func TestMessageMaxSizeReached(t *testing.T) {
-	var mssg = make(chan *Message, 2)
-	var data [63 << 10]byte
-	packets := GetPackets(true, 1, 2, data[:])
-	packets = append(packets, GetPackets(true, 1, 1, make([]byte, 63<<10+10))...)
-
-	p := NewMessageParser(63<<10+10, time.Second, nil, func(m *Message) { mssg <- m })
-	for _, v := range packets {
-		p.PacketHandler(v)
-	}
-	var m *Message
-	select {
-	case <-time.After(time.Second):
-		t.Errorf("can't parse packets fast enough")
-		return
-	case m = <-mssg:
-	}
-	if m.Length != 63<<10+10 {
-		t.Errorf("expected %d to equal %d", m.Length, 63<<10+10)
-	}
-	if !m.Truncated {
-		t.Error("expected message to be truncated")
-	}
-
-	select {
-	case <-time.After(time.Second):
-		t.Errorf("can't parse packets fast enough")
-		return
-	case m = <-mssg:
-	}
-	if m.Length != 63<<10+10 {
-		t.Errorf("expected %d to equal %d", m.Length, 63<<10+10)
-	}
-	if m.Truncated {
-		t.Error("expected message to not be truncated")
-	}
-}
-
 func TestMessageTimeoutReached(t *testing.T) {
-	var mssg = make(chan *Message, 2)
 	var data [63 << 10]byte
 	packets := GetPackets(true, 1, 2, data[:])
-	p := NewMessageParser(1<<20, 0, nil, func(m *Message) { mssg <- m })
-	p.PacketHandler(packets[0])
-	time.Sleep(time.Millisecond * 400)
-	p.PacketHandler(packets[1])
-	m := <-mssg
+	p := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true)
+	p.processPacket(packets[0])
+	time.Sleep(time.Millisecond * 100)
+	p.processPacket(packets[1])
+	m := p.Read()
 	if m.Length != 63<<10 {
 		t.Errorf("expected %d to equal %d", m.Length, 63<<10)
 	}
@@ -276,38 +228,17 @@ func TestMessageTimeoutReached(t *testing.T) {
 	}
 }
 
-func TestMessageUUID(t *testing.T) {
-	packets := GetPackets(true, 1, 10, nil)
-
-	var uuid, uuid1 []byte
-	parser := NewMessageParser(0, 0, nil, func(msg *Message) {
-		if len(uuid) == 0 {
-			uuid = msg.UUID()
-			return
-		}
-		uuid1 = msg.UUID()
-	})
-
-	for _, p := range packets {
-		parser.PacketHandler(p)
-	}
-
-	if string(uuid) != string(uuid1) {
-		t.Errorf("expected %s, to equal %s", uuid, uuid1)
-	}
-}
-
 func BenchmarkMessageUUID(b *testing.B) {
 	packets := GetPackets(true, 1, 5, nil)
 
 	var uuid []byte
-	var msg *Message
-	parser := NewMessageParser(0, 0, nil, func(m *Message) {
-		msg = m
-	})
+	parser := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true)
 	for _, p := range packets {
-		parser.PacketHandler(p)
+		parser.processPacket(p)
 	}
+
+	msg := parser.Read()
+
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		uuid = msg.UUID()
@@ -328,20 +259,16 @@ func BenchmarkPacketParseAndSort(b *testing.B) {
 }
 
 func BenchmarkMessageParserWithoutHint(b *testing.B) {
-	// runtime.GOMAXPROCS(8)
-	var mssg = make(chan *Message, 1)
 	var chunk = []byte("111111111111111111111111111111")
 	packets := GetPackets(true, 1, 1000, chunk)
-	p := NewMessageParser(1<<20, time.Second*2, nil, func(m *Message) {
-		mssg <- m
-	})
+	p := NewMessageParser(nil, nil, nil, 2*time.Second, false)
 	b.ResetTimer()
 	b.ReportMetric(float64(1000), "packets/op")
 	for i := 0; i < b.N; i++ {
 		for _, v := range packets {
-			p.PacketHandler(v)
+			p.processPacket(v)
 		}
-		<-mssg
+		p.Read()
 	}
 }
 
@@ -357,22 +284,22 @@ func BenchmarkMessageParserWithHint(b *testing.B) {
 	for i := 0; i < len(buf); i++ {
 		packets[i] = GetPackets(false, 1, 1, buf[i])[0]
 	}
-	var mssg = make(chan *Message, 1)
-	parser := NewMessageParser(1<<30, time.Second*10, nil, func(m *Message) { mssg <- m })
+
+	parser := NewMessageParser(nil, nil, nil, 2*time.Second, false)
 	parser.Start = func(pckt *Packet) (bool, bool) {
 		return false, proto.HasResponseTitle(pckt.Payload)
 	}
 	parser.End = func(m *Message) bool {
-		return proto.HasFullPayload(m, m.Data())
+		return proto.HasFullPayload(m, m.PacketData()...)
 	}
 	b.ResetTimer()
 	b.ReportMetric(float64(len(packets)), "packets/op")
 	b.ReportMetric(float64(1000), "chunks/op")
 	for i := 0; i < b.N; i++ {
 		for j := range packets {
-			parser.PacketHandler(packets[j])
+			parser.processPacket(packets[j])
 		}
-		<-mssg
+		parser.Read()
 	}
 }
 
@@ -380,6 +307,6 @@ func BenchmarkNewAndParsePacket(b *testing.B) {
 	data := append(generateHeader(true, 1024, 10), make([]byte, 10)...)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		ParsePacket(data, int(layers.LinkTypeLoop), 4, &gopacket.CaptureInfo{})
+		ParsePacket(data, int(layers.LinkTypeLoop), 4, &gopacket.CaptureInfo{}, true)
 	}
 }