Fix: message size check in timeout test (#1014)

Fix message size check in `TestMessageTimeoutReached`. Since the message parser has parsed two packets of size 63 << 10, then the message size should be 63 << 11

Makefile

@@ -23,19 +23,24 @@ FPMCOMMON= \
     -s dir \
     -C /tmp/gor-build \
 
-release: vendor release-x64 release-mac release-windows
+release: release-x64 release-x86 release-mac release-windows
 
 vendor:
 	go mod vendor
 
-release-bin:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+release-bin: vendor
+	docker run --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
 
-release-bin-mac:
-	GOOS=darwin go build -o $(BIN_NAME) $(MAC_LDFLAGS)
+release-bin-x64: vendor
+	docker run --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
 
-release-x64:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+release-bin-mac: vendor
+	GOOS=darwin go build -mod=vendor -o $(BIN_NAME) $(MAC_LDFLAGS)
+
+release-bin-windows: vendor
+	docker run -it --rm -v `pwd`:$(SOURCE_PATH) -w $(SOURCE_PATH) -e CGO_ENABLED=1 docker.elastic.co/beats-dev/golang-crossbuild:1.16.4-main --build-cmd "make VERSION=$(VERSION) build" -p "windows/amd64"
+
+release-x64: release-bin
 	tar -czf gor_$(VERSION)$(PREFIX)_x64.tar.gz $(BIN_NAME)
 	mkdir -p /tmp/gor-build
 	mv ./$(BIN_NAME) /tmp/gor-build/$(BIN_NAME)
@@ -46,13 +51,11 @@ release-x64:
 	fpm $(FPMCOMMON) -a amd64 -t rpm ./=/usr/local/bin
 	rm -rf /tmp/gor-build
 
-release-x86:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
+release-x86: release-bin-x64
 	tar -czf gor_$(VERSION)$(PREFIX)_x86.tar.gz $(BIN_NAME)
 	rm $(BIN_NAME)
 
-release-mac:
-	go build -mod=vendor -o $(BIN_NAME) $(MAC_LDFLAGS)
+release-mac: release-bin-mac
 	tar -czf gor_$(VERSION)$(PREFIX)_mac.tar.gz $(BIN_NAME)
 	mkdir -p /tmp/gor-build
 	mv ./$(BIN_NAME) /tmp/gor-build/$(BIN_NAME)
@@ -61,14 +64,7 @@ release-mac:
 	fpm $(FPMCOMMON) -a amd64 -t osxpkg ./=/usr/local/bin
 	rm -rf /tmp/gor-build
 
-release-windows:
-	docker run -it --rm \
-	  -v `pwd`:/go/src/github.com/buger/goreplay \
-	  -w /go/src/github.com/buger/goreplay \
-	  -e CGO_ENABLED=1 \
-	  docker.elastic.co/beats-dev/golang-crossbuild:1.16.4-main \
-	  --build-cmd "make VERSION=$(VERSION) build" \
-	  -p "windows/amd64"
+release-windows: release-bin-windows
 	mv ./gor ./gor.exe
 	zip gor-$(VERSION)$(PREFIX)_windows.zip ./gor.exe
 	rm -rf ./gor.exe

capture/capture.go

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

input_file.go

@@ -63,6 +63,8 @@ type fileInputReader struct {
 	s3        bool
 	queue     payloadQueue
 	readDepth int
+	dryRun    bool
+	path      string
 }
 
 func (f *fileInputReader) parse(init chan struct{}) error {
@@ -70,8 +72,11 @@ func (f *fileInputReader) parse(init chan struct{}) error {
 	var buffer bytes.Buffer
 	var initialized bool
 
+	lineNum := 0
+
 	for {
 		line, err := f.reader.ReadBytes('\n')
+		lineNum++
 
 		if err != nil {
 			if err != io.EOF {
@@ -92,6 +97,12 @@ func (f *fileInputReader) parse(init chan struct{}) error {
 			asBytes := buffer.Bytes()
 			meta := payloadMeta(asBytes)
 
+			if len(meta) < 3 {
+				Debug(1, fmt.Sprintf("Found malformed record, file: %s, line %d", f.path, lineNum))
+				buffer = bytes.Buffer{}
+				continue
+			}
+
 			timestamp, _ := strconv.ParseInt(string(meta[2]), 10, 64)
 			data := asBytes[:len(asBytes)-1]
 
@@ -112,7 +123,9 @@ func (f *fileInputReader) parse(init chan struct{}) error {
 					initialized = true
 				}
 
-				time.Sleep(100 * time.Millisecond)
+				if !f.dryRun {
+					time.Sleep(100 * time.Millisecond)
+				}
 			}
 
 			buffer = bytes.Buffer{}
@@ -133,7 +146,9 @@ func (f *fileInputReader) wait() {
 			return
 		}
 
-		time.Sleep(100 * time.Millisecond)
+		if !f.dryRun {
+			time.Sleep(100 * time.Millisecond)
+		}
 	}
 
 	return
@@ -149,7 +164,7 @@ func (f *fileInputReader) Close() error {
 	return nil
 }
 
-func newFileInputReader(path string, readDepth int) *fileInputReader {
+func newFileInputReader(path string, readDepth int, dryRun bool) *fileInputReader {
 	var file io.ReadCloser
 	var err error
 
@@ -164,7 +179,7 @@ func newFileInputReader(path string, readDepth int) *fileInputReader {
 		return nil
 	}
 
-	r := &fileInputReader{file: file, closed: 0, readDepth: readDepth}
+	r := &fileInputReader{path: path, file: file, closed: 0, readDepth: readDepth, dryRun: dryRun}
 	if strings.HasSuffix(path, ".gz") {
 		gzReader, err := gzip.NewReader(file)
 		if err != nil {
@@ -262,7 +277,7 @@ func (i *FileInput) init() (err error) {
 	i.readers = make([]*fileInputReader, len(matches))
 
 	for idx, p := range matches {
-		i.readers[idx] = newFileInputReader(p, i.readDepth)
+		i.readers[idx] = newFileInputReader(p, i.readDepth, i.dryRun)
 	}
 
 	i.stats.Add("reader_count", int64(len(matches)))

output_file.go

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

proto/proto.go

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

proto/proto_test.go

@@ -56,6 +56,12 @@ func TestHeader(t *testing.T) {
 	if val = Header(payload, []byte("host")); !bytes.Equal(val, []byte("www.w3.org")) {
 		t.Error("Should find lower case 1 word header")
 	}
+
+	payload = []byte("GT\r\nContent-Length: 10\r\n\r\n")
+
+	if val = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("10")) {
+		t.Error("Should find in partial payload")
+	}
 }
 
 func TestMIMEHeadersEndPos(t *testing.T) {

tcp/tcp_message.go

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

tcp/tcp_packet.go

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

tcp/tcp_test.go

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