diff --git a/capture/capture.go b/capture/capture.go index 53ff59d..2fc0be2 100644 --- a/capture/capture.go +++ b/capture/capture.go @@ -15,6 +15,7 @@ import ( "syscall" "time" + "github.com/buger/goreplay/proto" "github.com/buger/goreplay/size" "github.com/buger/goreplay/tcp" @@ -59,9 +60,13 @@ type Listener struct { 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) @@ -121,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 @@ -139,6 +144,11 @@ func NewListener(host string, ports []uint16, transport string, engine EngineTyp 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 @@ -171,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: @@ -185,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 } }() @@ -332,11 +342,34 @@ 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 packetHandle) { + runtime.LockOSThread() + defer l.closeHandles(key) linkSize := 14 linkType := int(layers.LinkTypeEthernet) @@ -351,6 +384,13 @@ 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 { @@ -371,23 +411,12 @@ func (l *Listener) read(handler PacketHandler) { ci.Timestamp = time.Now() } - pckt, err := tcp.ParsePacket(data, linkType, linkSize, &ci, false) - - if err == nil { - for _, p := range l.ports { - if pckt.DstPort == p { - for _, ip := range hndl.ips { - if pckt.DstIP.Equal(ip) { - pckt.Direction = tcp.DirIncoming - break - } - } - break - } - } - - handler(pckt) - } + messageParser.PacketHandler(&tcp.PcapPacket{ + Data: data, + LType: linkType, + LTypeLen: linkSize, + Ci: &ci, + }) continue } if enext, ok := err.(pcap.NextError); ok && enext == pcap.NextErrorTimeoutExpired { @@ -413,6 +442,10 @@ 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() diff --git a/input_raw.go b/input_raw.go index 7c9a43a..1c7fc04 100644 --- a/input_raw.go +++ b/input_raw.go @@ -16,40 +16,6 @@ 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 @@ -57,7 +23,7 @@ type RAWInputConfig struct { 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"` + 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"` @@ -117,7 +83,7 @@ func (i *RAWInput) PluginRead() (*Message, error) { select { case <-i.quit: return nil, ErrorStopped - case msgTCP = <-i.messageParser.Messages(): + case msgTCP = <-i.listener.Messages(): msg.Data = msgTCP.Data() } @@ -142,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) } @@ -158,15 +123,10 @@ func (i *RAWInput) listen(address string) { if err != nil { log.Fatal(err) } - i.messageParser = tcp.NewMessageParser(i.CopyBufferSize, i.Expire, i.AllowIncomplete, Debug) - if i.Protocol == ProtocolHTTP { - i.messageParser.Start = http1StartHint - i.messageParser.End = http1EndHint - } var ctx context.Context ctx, i.cancelListener = context.WithCancel(context.Background()) - errCh := i.listener.ListenBackground(ctx, i.messageParser.PacketHandler) + errCh := i.listener.ListenBackground(ctx) <-i.listener.Reading Debug(1, i) go func() { @@ -210,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()...) -} diff --git a/input_raw_test.go b/input_raw_test.go index 969e5a7..646fb2a 100644 --- a/input_raw_test.go +++ b/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", } @@ -113,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) @@ -178,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) @@ -235,7 +236,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) { conf := RAWInputConfig{ Engine: capture.EnginePcap, Expire: time.Second, - Protocol: ProtocolHTTP, + Protocol: tcp.ProtocolHTTP, TrackResponse: true, AllowIncomplete: true, } @@ -315,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) diff --git a/tcp/tcp_message.go b/tcp/tcp_message.go index 6c532e3..09d1503 100644 --- a/tcp/tcp_message.go +++ b/tcp/tcp_message.go @@ -3,15 +3,48 @@ package tcp import ( "encoding/binary" "encoding/hex" - _ "fmt" + "fmt" + "net" "reflect" "sort" "time" "unsafe" - - "github.com/buger/goreplay/size" ) +// 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 @@ -151,16 +184,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() { -} - // 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 @@ -172,9 +198,7 @@ type HintStart func(*Packet) (IsRequest, IsOutgoing bool) // MessageParser holds data of all tcp messages in progress(still receiving/sending packets). // message is identified by its source port and dst port, and last 4bytes of src IP. type MessageParser struct { - debug Debugger - maxSize size.Size // maximum message size, default 5mb - m map[uint64]*Message + m map[uint64]*Message messageExpire time.Duration // the maximum time to wait for the final packet, minimum is 100ms allowIncompete bool @@ -182,14 +206,15 @@ type MessageParser struct { Start HintStart ticker *time.Ticker messages chan *Message - packets chan *Packet + 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, allowIncompete bool, debugger Debugger) (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.messageExpire = messageExpire if parser.messageExpire == 0 { @@ -197,17 +222,21 @@ func NewMessageParser(maxSize size.Size, messageExpire time.Duration, allowIncom } parser.allowIncompete = allowIncompete - parser.maxSize = maxSize - if parser.maxSize < 1 { - parser.maxSize = 5 << 20 - } - parser.packets = make(chan *Packet, 10000) - parser.messages = make(chan *Message, 10000) + parser.packets = make(chan *PcapPacket, 10000) + + if messages == nil { + messages = make(chan *Message, 1000) + } + parser.messages = messages parser.m = make(map[uint64]*Message) parser.ticker = time.NewTicker(time.Millisecond * 100) parser.close = make(chan struct{}, 1) + + parser.ports = ports + parser.ips = ips + go parser.wait() return parser } @@ -215,7 +244,7 @@ func NewMessageParser(maxSize size.Size, messageExpire time.Duration, allowIncom var packetLen int // Packet returns packet handler -func (parser *MessageParser) PacketHandler(packet *Packet) { +func (parser *MessageParser) PacketHandler(packet *PcapPacket) { packetLen++ parser.packets <- packet } @@ -227,7 +256,7 @@ func (parser *MessageParser) wait() { for { select { case pckt := <-parser.packets: - parser.processPacket(pckt) + parser.processPacket(parser.parsePacket(pckt)) case now = <-parser.ticker.C: parser.timer(now) case <-parser.close: @@ -240,30 +269,51 @@ func (parser *MessageParser) wait() { } } +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) { + 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()] switch { case ok: + if m.Direction == DirUnknown { + if in, out := parser.Start(pckt); in || out { + if in { + m.Direction = DirIncoming + } else { + m.Direction = DirOutcoming + } + } + } parser.addPacket(m, pckt) return 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, out := parser.Start(pckt); in || out { if in { pckt.Direction = DirIncoming } else { @@ -281,20 +331,10 @@ func (parser *MessageParser) processPacket(pckt *Packet) { } func (parser *MessageParser) addPacket(m *Message, pckt *Packet) bool { - trunc := m.Length + len(pckt.Payload) - int(parser.maxSize) - if trunc > 0 { - m.Truncated = true - stats.Add("message_timeout_count", 1) - pckt.Payload = pckt.Payload[:int(parser.maxSize)-m.Length] - } if !m.add(pckt) { return false } - if trunc > 0 { - return false - } - // 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 { @@ -311,10 +351,6 @@ func (parser *MessageParser) Read() *Message { return m } -func (parser *MessageParser) Messages() chan *Message { - return parser.messages -} - func (parser *MessageParser) Emit(m *Message) { stats.Add("message_count", 1) @@ -332,6 +368,9 @@ var failMsg int func (parser *MessageParser) timer(now time.Time) { packetLen = 0 + packetQueueLen.Set(int64(len(parser.packets))) + messageQueueLen.Set(int64(len(parser.m))) + for _, m := range parser.m { if now.Sub(m.End) > parser.messageExpire { m.TimedOut = true @@ -339,19 +378,10 @@ func (parser *MessageParser) timer(now time.Time) { failMsg++ if parser.End == nil || parser.allowIncompete { parser.Emit(m) - } else { - // Just remove - delete(parser.m, m.packets[0].MessageID()) - m.Finalize() } - } - } -} -// this function should not block other parser operations -func (parser *MessageParser) Debug(level int, args ...interface{}) { - if parser.debug != nil { - parser.debug(level, args...) + delete(parser.m, m.packets[0].MessageID()) + } } } diff --git a/tcp/tcp_packet.go b/tcp/tcp_packet.go index 2a25163..c7f7641 100644 --- a/tcp/tcp_packet.go +++ b/tcp/tcp_packet.go @@ -30,17 +30,16 @@ func copySlice(to []byte, skip int, from ...[]byte) ([]byte, int) { } var stats *expvar.Map -var bufPoolCount *expvar.Int -var releasedCount *expvar.Int +var packetQueueLen, messageQueueLen *expvar.Int func init() { - bufPoolCount = new(expvar.Int) - releasedCount = new(expvar.Int) + packetQueueLen = new(expvar.Int) + messageQueueLen = new(expvar.Int) stats = expvar.NewMap("tcp") stats.Init() - stats.Set("buffer_pool_count", bufPoolCount) - stats.Set("buffer_released", releasedCount) + stats.Set("packet_queue", packetQueueLen) + stats.Set("message_queue", messageQueueLen) } type Dir int @@ -76,10 +75,17 @@ type Packet struct { 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, allowEmpty bool) (pckt *Packet, err error) { +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, cp, allowEmpty); err != nil { + if err := pckt.parse(data, lType, lTypeLen, ci, allowEmpty); err != nil { return nil, err } diff --git a/tcp/tcp_test.go b/tcp/tcp_test.go index 055d8f1..e82b415 100644 --- a/tcp/tcp_test.go +++ b/tcp/tcp_test.go @@ -79,7 +79,7 @@ func TestRequestResponseMapping(t *testing.T) { {SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 56, Direction: DirOutcoming, Timestamp: time.Unix(8, 0), Payload: []byte("Content-Length: 0\r\n\r\n")}, } - parser := NewMessageParser(1<<20, time.Second, false, nil) + parser := NewMessageParser(nil, nil, nil, time.Second, false) parser.Start = func(pckt *Packet) (bool, bool) { return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload) } @@ -88,7 +88,7 @@ func TestRequestResponseMapping(t *testing.T) { } for _, packet := range packets { - parser.PacketHandler(packet) + parser.processPacket(packet) } messages := []*Message{} @@ -109,7 +109,7 @@ func TestRequestResponseMapping(t *testing.T) { } func TestMessageParserWithHint(t *testing.T) { - parser := NewMessageParser(1<<20, time.Second, false, nil) + parser := NewMessageParser(nil, nil, nil, time.Second, false) parser.Start = func(pckt *Packet) (bool, bool) { return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload) } @@ -131,7 +131,7 @@ func TestMessageParserWithHint(t *testing.T) { } for _, p := range packets { - parser.PacketHandler(p) + parser.processPacket(p) } messages := []*Message{} @@ -154,26 +154,28 @@ func TestMessageParserWithHint(t *testing.T) { } func TestMessageParserWrongOrder(t *testing.T) { - parser := NewMessageParser(1<<20, time.Second, true, nil) + 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.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")}, + } + + for _, p := range packets { + parser.processPacket(p) } m := parser.Read() @@ -184,6 +186,12 @@ func TestMessageParserWrongOrder(t *testing.T) { 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")) } @@ -193,9 +201,9 @@ func TestMessageParserWithoutHint(t *testing.T) { var data [63 << 10]byte packets := GetPackets(true, 1, 10, data[:]) - p := NewMessageParser(63<<10*10, time.Second, false, nil) + p := NewMessageParser(nil, nil, nil, time.Second, false) for _, v := range packets { - p.PacketHandler(v) + p.processPacket(v) } m := p.Read() @@ -204,42 +212,13 @@ func TestMessageParserWithoutHint(t *testing.T) { } } -func TestMessageMaxSizeReached(t *testing.T) { - var data [63 << 10]byte - packets := GetPackets(true, 1, 2, data[:]) - packets = append(packets, GetPackets(false, 1, 1, make([]byte, 63<<10+10))...) - - p := NewMessageParser(63<<10+10, time.Millisecond, false, nil) - for _, v := range packets { - p.PacketHandler(v) - } - time.Sleep(10 * time.Millisecond) - - m := p.Read() - 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") - } - - m = p.Read() - - 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 data [63 << 10]byte packets := GetPackets(true, 1, 2, data[:]) - p := NewMessageParser(1<<20, 10*time.Millisecond, true, nil) - p.PacketHandler(packets[0]) + p := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true) + p.processPacket(packets[0]) time.Sleep(time.Millisecond * 100) - p.PacketHandler(packets[1]) + p.processPacket(packets[1]) m := p.Read() if m.Length != 63<<10 { t.Errorf("expected %d to equal %d", m.Length, 63<<10) @@ -253,9 +232,9 @@ func BenchmarkMessageUUID(b *testing.B) { packets := GetPackets(true, 1, 5, nil) var uuid []byte - parser := NewMessageParser(0, 0, false, nil) + parser := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true) for _, p := range packets { - parser.PacketHandler(p) + parser.processPacket(p) } msg := parser.Read() @@ -282,12 +261,12 @@ func BenchmarkPacketParseAndSort(b *testing.B) { func BenchmarkMessageParserWithoutHint(b *testing.B) { var chunk = []byte("111111111111111111111111111111") packets := GetPackets(true, 1, 1000, chunk) - p := NewMessageParser(1<<20, time.Second*2, false, nil) + 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) } p.Read() } @@ -306,7 +285,7 @@ func BenchmarkMessageParserWithHint(b *testing.B) { packets[i] = GetPackets(false, 1, 1, buf[i])[0] } - parser := NewMessageParser(1<<30, time.Second*10, false, nil) + parser := NewMessageParser(nil, nil, nil, 2*time.Second, false) parser.Start = func(pckt *Packet) (bool, bool) { return false, proto.HasResponseTitle(pckt.Payload) } @@ -318,7 +297,7 @@ func BenchmarkMessageParserWithHint(b *testing.B) { 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]) } parser.Read() }