streaming: fix Read interface

It wasn't actually compatible with io.Reader as io.Reader _never_ returns n >
len(p) while this function easily did that for chunked payloads confusing its
users:

panic: runtime error: slice bounds out of range [:528] with capacity 512

goroutine 562 [running]:
io.ReadAll(0x9f4380, 0xc0003be1a0, 0xc0004fcd80, 0x0, 0x0, 0xc00086bc30, 0x46f99b)
        /usr/lib64/go/1.16/src/io/io.go:634 +0x205
io/ioutil.ReadAll(...)
        /usr/lib64/go/1.16/src/io/ioutil/ioutil.go:27
github.com/valyala/fasthttp.getChunkedTestEnv.func1(0xc001fdc680)
        /home/rik/dev/fasthttp/streaming_test.go:108 +0x6c
github.com/valyala/fasthttp.(*Server).serveConn(0xc000416d80, 0xa034e8, 0xc0004da880, 0x0, 0x0)
        /home/rik/dev/fasthttp/server.go:2219 +0x12ee
github.com/valyala/fasthttp.(*workerPool).workerFunc(0xc000148960, 0xc0003be160)
        /home/rik/dev/fasthttp/workerpool.go:223 +0xba
github.com/valyala/fasthttp.(*workerPool).getCh.func1(0xc000148960, 0xc0003be160, 0x8b4ec0, 0xc0003be160)
        /home/rik/dev/fasthttp/workerpool.go:195 +0x35
created by github.com/valyala/fasthttp.(*workerPool).getCh
        /home/rik/dev/fasthttp/workerpool.go:194 +0x11f

It also returned len(p) in some cases where it read less than that.

streaming.go

@@ -3,7 +3,6 @@ package fasthttp
 import (
 	"bufio"
 	"bytes"
-	"fmt"
 	"io"
 	"sync"
 
@@ -15,36 +14,47 @@ type requestStream struct {
 	reader          *bufio.Reader
 	totalBytesRead  int
 	contentLength   int
+	chunkLeft       int
 }
 
 func (rs *requestStream) Read(p []byte) (int, error) {
+	var (
+		n   int
+		err error
+	)
 	if rs.contentLength == -1 {
-		p = p[:0]
-		strCRLFLen := len(strCRLF)
-		chunkSize, err := parseChunkSize(rs.reader)
-		if err != nil {
-			return len(p), err
+		if rs.chunkLeft == 0 {
+			chunkSize, err := parseChunkSize(rs.reader)
+			if err != nil {
+				return 0, err
+			}
+			if chunkSize == 0 {
+				err = readCrLf(rs.reader)
+				if err == nil {
+					err = io.EOF
+				}
+				return 0, err
+			}
+			rs.chunkLeft = chunkSize
 		}
-		p, err = appendBodyFixedSize(rs.reader, p, chunkSize+strCRLFLen)
-		if err != nil {
-			return len(p), err
+		bytesToRead := len(p)
+		if rs.chunkLeft < len(p) {
+			bytesToRead = rs.chunkLeft
 		}
-		if !bytes.Equal(p[len(p)-strCRLFLen:], strCRLF) {
-			return len(p), ErrBrokenChunk{
-				error: fmt.Errorf("cannot find crlf at the end of chunk"),
-			}
+		n, err = rs.reader.Read(p[:bytesToRead])
+		rs.totalBytesRead += n
+		rs.chunkLeft -= n
+		if err == io.EOF {
+			err = io.ErrUnexpectedEOF
 		}
-		p = p[:len(p)-strCRLFLen]
-		if chunkSize == 0 {
-			return len(p), io.EOF
+		if err == nil && rs.chunkLeft == 0 {
+			err = readCrLf(rs.reader)
 		}
-		return len(p), nil
+		return n, err
 	}
 	if rs.totalBytesRead == rs.contentLength {
 		return 0, io.EOF
 	}
-	var n int
-	var err error
 	prefetchedSize := int(rs.prefetchedBytes.Size())
 	if prefetchedSize > rs.totalBytesRead {
 		left := prefetchedSize - rs.totalBytesRead
@@ -87,6 +97,7 @@ func acquireRequestStream(b *bytebufferpool.ByteBuffer, r *bufio.Reader, content
 func releaseRequestStream(rs *requestStream) {
 	rs.prefetchedBytes = nil
 	rs.totalBytesRead = 0
+	rs.chunkLeft = 0
 	rs.reader = nil
 	requestStreamPool.Put(rs)
 }

streaming_test.go

@@ -101,7 +101,7 @@ aaaaaaaaaa`
 }
 
 func getChunkedTestEnv(t testing.TB) (*fasthttputil.InmemoryListener, []byte) {
-	body := createFixedBody(3)
+	body := createFixedBody(128 * 1024)
 	chunkedBody := createChunkedBody(body)
 
 	testHandler := func(ctx *RequestCtx) {