diff --git a/tcp/flow.go b/tcp/flow.go
index af9d8c2..1732e0b 100644
--- a/tcp/flow.go
+++ b/tcp/flow.go
@@ -1,6 +1,7 @@
 package tcp
 
 import (
+	"bufio"
 	"encoding/binary"
 	"fmt"
 	"io"
@@ -208,10 +209,12 @@ func (f *Flow) consumeMarshalled() {
 }
 
 func (f *Flow) produceMarshalled() {
+	buf := bufio.NewReader(f.conn)
+
 	for {
 		t1 := time.Now()
 		lengthBytes := make([]byte, 4)
-		if n, err := io.LimitReader(f.conn, 4).Read(lengthBytes); err != nil {
+		if n, err := io.LimitReader(buf, 4).Read(lengthBytes); err != nil {
 			f.produceErrors <- err
 			return
 		} else if n != 4 {
@@ -224,7 +227,7 @@ func (f *Flow) produceMarshalled() {
 
 		var read uint32
 		for read < length {
-			if n, err := io.LimitReader(f.conn, int64(length-read)).Read(dataBytes[read:]); err != nil {
+			if n, err := io.LimitReader(buf, int64(length-read)).Read(dataBytes[read:]); err != nil {
 				f.produceErrors <- err
 				return
 			} else {