From 1e0bda09f4c53b7d2aa2860e8aba08532086ab15 Mon Sep 17 00:00:00 2001
From: Jake Hillion <jake@hillion.co.uk>
Date: Sat, 23 Jan 2021 18:39:01 +0000
Subject: [PATCH] buffered reader

---
 tcp/flow.go | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

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 {