Fix detection of packet detection for complex bpf fiters

Right now it depends on the defined ports, but you can have more compex bpf, so wee may stil need ti check it with Start function

capture/capture.go

@@ -357,7 +357,7 @@ func (l *Listener) read(handler PacketHandler) {
 								if pckt.DstPort == p {
 									for _, ip := range hndl.ips {
 										if pckt.DstIP.Equal(ip) {
-											pckt.Incoming = true
+											pckt.Direction = tcp.DirIncoming
 											break
 										}
 									}

input_raw.go

@@ -122,7 +122,7 @@ func (i *RAWInput) PluginRead() (*Message, error) {
 	}
 
 	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))

tcp/tcp_message.go

@@ -20,7 +20,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
@@ -40,7 +40,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 {
@@ -51,7 +51,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)
@@ -152,7 +152,6 @@ func (m *Message) Sort() {
 }
 
 func (m *Message) Finalize() {
-
 }
 
 // Emitter message handler
@@ -242,8 +241,6 @@ func (parser *MessageParser) wait() {
 }
 
 func (parser *MessageParser) processPacket(pckt *Packet) {
-	var in bool
-
 	// 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()]
@@ -251,12 +248,32 @@ func (parser *MessageParser) processPacket(pckt *Packet) {
 	case ok:
 		parser.addPacket(m, pckt)
 		return
-	default:
-		in = pckt.Incoming
+	case pckt.Direction == DirUnknown && 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 < 2 && len(pckt.Payload) > 0 {
+				// Requeue not known packets
+				pckt.Retry++
+
+				select {
+				case parser.packets <- pckt:
+					return
+				default:
+				}
+			}
+
+			return
+		} else {
+			if in {
+				pckt.Direction = DirIncoming
+			} else {
+				pckt.Direction = DirOutcoming
+			}
+		}
 	}
 
 	m = new(Message)
-	m.IsRequest = in
+	m.Direction = pckt.Direction
 	parser.m[pckt.MessageID()] = m
 	m.Start = pckt.Timestamp
 	m.parser = parser

tcp/tcp_packet.go

@@ -43,6 +43,14 @@ func init() {
 	stats.Set("buffer_released", releasedCount)
 }
 
+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,
@@ -50,7 +58,7 @@ calllers must make sure that ParsePacket has'nt returned any error before callin
 function.
 */
 type Packet struct {
-	Incoming           bool
+	Direction          Dir
 	messageID          uint64
 	SrcIP, DstIP       net.IP
 	Version            uint8

tcp/tcp_test.go

@@ -49,7 +49,7 @@ func GetPackets(request bool, start uint32, _len int, payload []byte) []*Packet
 		ci := &gopacket.CaptureInfo{Length: len(d), CaptureLength: len(d), Timestamp: time.Now()}
 
 		packets[i-start], err = ParsePacket(d, int(layers.LinkTypeLoop), 4, ci, true)
-		packets[i-start].Incoming = request
+		packets[i-start].Direction = DirIncoming
 		if err != nil {
 			panic(err)
 		}
@@ -93,10 +93,10 @@ func TestRequestResponseMapping(t *testing.T) {
 		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, messages[0].Direction, DirIncoming)
+	assert.Equal(t, messages[1].Direction, DirOutcoming)
+	assert.Equal(t, messages[2].Direction, DirIncoming)
+	assert.Equal(t, messages[3].Direction, DirOutcoming)
 
 	assert.Equal(t, messages[0].UUID(), messages[1].UUID())
 	assert.Equal(t, messages[2].UUID(), messages[3].UUID())