Repository: valyala/fasthttp
Pull request #1414 (2022-11-02 12:00:06)

support response body stream
why:
When developing HTTP proxy tools, we encountered the need to forward large packets, and the second is to support streaming forwarding, so we need to modify the sdk.

For security reasons,`Response.BodyStream()` function return `io.ReadCloser` and lets the user manually release the resource . 

Finally, I want request to support stream and open it up to developers.

byene0923: i'm also look deep into it, i think the code can be more elegant, do you think use `Requeststream`   in response is a good way? 
Anthony-Dong: > 我也在深入研究它，我认为代码可以更优雅，你认为使用`Requeststream` 响应是一个好方法吗？



> i'm also look deep into it, i think the code can be more elegant, do you think use `Requeststream` in response is a good way?


Anthony-Dong: > i'm also look deep into it, i think the code can be more elegant, do you think use `Requeststream` in response is a good way?



> 

I think it's just a naming issue, since request stream was already supported.
byene0923: 



> > i'm also look deep into it, i think the code can be more elegant, do you think use `Requeststream` in response is a good way?
> 
> > 
> 
> I think it's just a naming issue, since request stream was already supported.

you could change the name  such as `bodyStream`
Anthony-Dong: @erikdubbelboer do you have any questions ?
Anthony-Dong: > 



> > > i'm also look deep into it, i think the code can be more elegant, do you think use `Requeststream` in response is a good way?
> > 
> > 
> > > 
> > 
> > 
> > I think it's just a naming issue, since request stream was already supported.
> 
> you could change the name such as `bodyStream`

`type responseStream requestStream` .  might be better

Patch:
diff --git a/args.go b/args.go
index 7b0c0f3..b4292e4 100644
--- a/args.go
+++ b/args.go
@@ -63,7 +63,6 @@ func (a *Args) Reset() {
 
 // CopyTo copies all args to dst.
 func (a *Args) CopyTo(dst *Args) {
-	dst.Reset()
 	dst.args = copyArgs(dst.args, a.args)
 }
 
diff --git a/brotli.go b/brotli.go
index 815e4b3..c829c39 100644
--- a/brotli.go
+++ b/brotli.go
@@ -138,10 +138,7 @@ func nonblockingWriteBrotli(ctxv interface{}) {
 	ctx := ctxv.(*compressCtx)
 	zw := acquireRealBrotliWriter(ctx.w, ctx.level)
 
-	_, err := zw.Write(ctx.p)
-	if err != nil {
-		panic(fmt.Sprintf("BUG: brotli.Writer.Write for len(p)=%d returned unexpected error: %v", len(ctx.p), err))
-	}
+	zw.Write(ctx.p) //nolint:errcheck // no way to handle this error anyway
 
 	releaseRealBrotliWriter(zw, ctx.level)
 }
diff --git a/bytesconv.go b/bytesconv.go
index 9b2ffeb..809b490 100644
--- a/bytesconv.go
+++ b/bytesconv.go
@@ -79,6 +79,7 @@ func ParseIPv4(dst net.IP, ipStr []byte) (net.IP, error) {
 	copy(dst, net.IPv4zero)
 	dst = dst.To4()
 	if dst == nil {
+		// developer sanity-check
 		panic("BUG: dst must not be nil")
 	}
 
@@ -126,6 +127,7 @@ func ParseHTTPDate(date []byte) (time.Time, error) {
 // AppendUint appends n to dst and returns the extended dst.
 func AppendUint(dst []byte, n int) []byte {
 	if n < 0 {
+		// developer sanity-check
 		panic("BUG: int must be positive")
 	}
 
@@ -281,6 +283,7 @@ var hexIntBufPool sync.Pool
 
 func writeHexInt(w *bufio.Writer, n int) error {
 	if n < 0 {
+		// developer sanity-check
 		panic("BUG: int must be positive")
 	}
 
diff --git a/client.go b/client.go
index 8546a12..b3e26cc 100644
--- a/client.go
+++ b/client.go
@@ -297,6 +297,9 @@ type Client struct {
 	// Connection pool strategy. Can be either LIFO or FIFO (default).
 	ConnPoolStrategy ConnPoolStrategyType
 
+	// StreamResponseBody enables response body streaming
+	StreamResponseBody bool
+
 	// ConfigureClient configures the fasthttp.HostClient.
 	ConfigureClient func(hc *HostClient) error
 
@@ -521,6 +524,7 @@ func (c *Client) Do(req *Request, resp *Response) error {
 			MaxConnWaitTimeout:            c.MaxConnWaitTimeout,
 			RetryIf:                       c.RetryIf,
 			ConnPoolStrategy:              c.ConnPoolStrategy,
+			StreamResponseBody:            c.StreamResponseBody,
 			clientReaderPool:              &c.readerPool,
 			clientWriterPool:              &c.writerPool,
 		}
@@ -795,6 +799,9 @@ type HostClient struct {
 	// Connection pool strategy. Can be either LIFO or FIFO (default).
 	ConnPoolStrategy ConnPoolStrategyType
 
+	// StreamResponseBody enables response body streaming
+	StreamResponseBody bool
+
 	lastUseTime uint32
 
 	connsLock  sync.Mutex
@@ -1294,26 +1301,23 @@ func isIdempotent(req *Request) bool {
 }
 
 func (c *HostClient) do(req *Request, resp *Response) (bool, error) {
-	nilResp := false
 	if resp == nil {
-		nilResp = true
 		resp = AcquireResponse()
+		defer ReleaseResponse(resp)
 	}
 
 	ok, err := c.doNonNilReqResp(req, resp)
 
-	if nilResp {
-		ReleaseResponse(resp)
-	}
-
 	return ok, err
 }
 
 func (c *HostClient) doNonNilReqResp(req *Request, resp *Response) (bool, error) {
 	if req == nil {
+		// for debugging purposes
 		panic("BUG: req cannot be nil")
 	}
 	if resp == nil {
+		// for debugging purposes
 		panic("BUG: resp cannot be nil")
 	}
 
@@ -1334,8 +1338,10 @@ func (c *HostClient) doNonNilReqResp(req *Request, resp *Response) (bool, error)
 
 	// backing up SkipBody in case it was set explicitly
 	customSkipBody := resp.SkipBody
+	customStreamBody := resp.StreamBody || c.StreamResponseBody
 	resp.Reset()
 	resp.SkipBody = customSkipBody
+	resp.StreamBody = customStreamBody
 
 	req.URI().DisablePathNormalizing = c.DisablePathNormalizing
 
@@ -1445,12 +1451,28 @@ func (c *HostClient) doNonNilReqResp(req *Request, resp *Response) (bool, error)
 		return retry, err
 	}
 
-	if resetConnection || req.ConnectionClose() || resp.ConnectionClose() || isConnRST {
+	closeConn := resetConnection || req.ConnectionClose() || resp.ConnectionClose() || isConnRST
+	if customStreamBody && resp.bodyStream != nil {
+		rbs := resp.bodyStream
+		resp.bodyStream = newCloseReader(rbs, func() error {
+			if r, ok := rbs.(*requestStream); ok {
+				releaseRequestStream(r)
+			}
+			if closeConn {
+				c.closeConn(cc)
+			} else {
+				c.releaseConn(cc)
+			}
+			return nil
+		})
+		return false, nil
+	}
+
+	if closeConn {
 		c.closeConn(cc)
 	} else {
 		c.releaseConn(cc)
 	}
-
 	return false, nil
 }
 
@@ -1620,8 +1642,7 @@ func (c *HostClient) dialConnFor(w *wantConn) {
 	}
 
 	cc := acquireClientConn(conn)
-	delivered := w.tryDeliver(cc, nil)
-	if !delivered {
+	if !w.tryDeliver(cc, nil) {
 		// not delivered, return idle connection
 		c.releaseConn(cc)
 	}
@@ -1994,7 +2015,7 @@ func dialAddr(addr string, dial DialFunc, dialDualStack, isTLS bool, tlsConfig *
 		return nil, err
 	}
 	if conn == nil {
-		panic("BUG: DialFunc returned (nil, nil)")
+		return nil, errors.New("dialling unsuccessful. Please report this bug!")
 	}
 
 	// We assume that any conn that has the Handshake() method is a TLS conn already.
diff --git a/compress.go b/compress.go
index 5781c77..50d381b 100644
--- a/compress.go
+++ b/compress.go
@@ -66,6 +66,7 @@ func releaseFlateReader(zr io.ReadCloser) {
 func resetFlateReader(zr io.ReadCloser, r io.Reader) error {
 	zrr, ok := zr.(zlib.Resetter)
 	if !ok {
+		// sanity check. should only be called with a zlib.Reader
 		panic("BUG: zlib.Reader doesn't implement zlib.Resetter???")
 	}
 	return zrr.Reset(r, nil)
@@ -101,7 +102,14 @@ func acquireRealGzipWriter(w io.Writer, level int) *gzip.Writer {
 	if v == nil {
 		zw, err := gzip.NewWriterLevel(w, level)
 		if err != nil {
-			panic(fmt.Sprintf("BUG: unexpected error from gzip.NewWriterLevel(%d): %v", level, err))
+			// gzip.NewWriterLevel only errors for invalid
+			// compression levels. Clamp it to be min or max.
+			if level < gzip.HuffmanOnly {
+				level = gzip.HuffmanOnly
+			} else {
+				level = gzip.BestCompression
+			}
+			zw, _ = gzip.NewWriterLevel(w, level)
 		}
 		return zw
 	}
@@ -175,10 +183,7 @@ func nonblockingWriteGzip(ctxv interface{}) {
 	ctx := ctxv.(*compressCtx)
 	zw := acquireRealGzipWriter(ctx.w, ctx.level)
 
-	_, err := zw.Write(ctx.p)
-	if err != nil {
-		panic(fmt.Sprintf("BUG: gzip.Writer.Write for len(p)=%d returned unexpected error: %v", len(ctx.p), err))
-	}
+	zw.Write(ctx.p) //nolint:errcheck // no way to handle this error anyway
 
 	releaseRealGzipWriter(zw, ctx.level)
 }
@@ -271,10 +276,7 @@ func nonblockingWriteDeflate(ctxv interface{}) {
 	ctx := ctxv.(*compressCtx)
 	zw := acquireRealDeflateWriter(ctx.w, ctx.level)
 
-	_, err := zw.Write(ctx.p)
-	if err != nil {
-		panic(fmt.Sprintf("BUG: zlib.Writer.Write for len(p)=%d returned unexpected error: %v", len(ctx.p), err))
-	}
+	zw.Write(ctx.p) //nolint:errcheck // no way to handle this error anyway
 
 	releaseRealDeflateWriter(zw, ctx.level)
 }
@@ -379,7 +381,14 @@ func acquireRealDeflateWriter(w io.Writer, level int) *zlib.Writer {
 	if v == nil {
 		zw, err := zlib.NewWriterLevel(w, level)
 		if err != nil {
-			panic(fmt.Sprintf("BUG: unexpected error from zlib.NewWriterLevel(%d): %v", level, err))
+			// zlib.NewWriterLevel only errors for invalid
+			// compression levels. Clamp it to be min or max.
+			if level < zlib.HuffmanOnly {
+				level = zlib.HuffmanOnly
+			} else {
+				level = zlib.BestCompression
+			}
+			zw, _ = zlib.NewWriterLevel(w, level)
 		}
 		return zw
 	}
diff --git a/fasthttpproxy/http.go b/fasthttpproxy/http.go
index 9dd5a25..94d2b09 100644
--- a/fasthttpproxy/http.go
+++ b/fasthttpproxy/http.go
@@ -51,7 +51,7 @@ func FasthttpHTTPDialerTimeout(proxy string, timeout time.Duration) fasthttp.Dia
 			return nil, err
 		}
 
-		req := fmt.Sprintf("CONNECT %s HTTP/1.1\r\nHost: %s\r\n", addr, addr)
+		req := "CONNECT " + addr + " HTTP/1.1\r\nHost: " + addr + "\r\n"
 		if auth != "" {
 			req += "Proxy-Authorization: Basic " + auth + "\r\n"
 		}
diff --git a/fs.go b/fs.go
index d4ea8bd..1278298 100644
--- a/fs.go
+++ b/fs.go
@@ -602,7 +602,7 @@ func (ff *fsFile) decReadersCount() {
 	ff.h.cacheLock.Lock()
 	ff.readersCount--
 	if ff.readersCount < 0 {
-		panic("BUG: negative fsFile.readersCount!")
+		ff.readersCount = 0
 	}
 	ff.h.cacheLock.Unlock()
 }
@@ -1395,6 +1395,7 @@ func readFileHeader(f *os.File, compressed bool, fileEncoding string) ([]byte, e
 func stripLeadingSlashes(path []byte, stripSlashes int) []byte {
 	for stripSlashes > 0 && len(path) > 0 {
 		if path[0] != '/' {
+			// developer sanity-check
 			panic("BUG: path must start with slash")
 		}
 		n := bytes.IndexByte(path[1:], '/')
diff --git a/http.go b/http.go
index ce38c1f..0508857 100644
--- a/http.go
+++ b/http.go
@@ -92,6 +92,10 @@ type Response struct {
 	// Relevant for bodyStream only.
 	ImmediateHeaderFlush bool
 
+	// StreamBody enables response body streaming.
+	// Use SetBodyStream to set the body stream.
+	StreamBody bool
+
 	bodyStream io.Reader
 	w          responseBodyWriter
 	body       *bytebufferpool.ByteBuffer
@@ -293,6 +297,47 @@ func (resp *Response) BodyWriter() io.Writer {
 	return &resp.w
 }
 
+// BodyStream returns io.Reader
+//
+// You must CloseBodyStream or ReleaseRequest after you use it.
+func (req *Request) BodyStream() io.Reader {
+	return req.bodyStream
+}
+
+func (req *Request) CloseBodyStream() error {
+	return req.closeBodyStream()
+}
+
+// BodyStream returns io.Reader
+//
+// You must CloseBodyStream or ReleaseResponse after you use it.
+func (resp *Response) BodyStream() io.Reader {
+	return resp.bodyStream
+}
+
+func (resp *Response) CloseBodyStream() error {
+	return resp.closeBodyStream()
+}
+
+type closeReader struct {
+	io.Reader
+	closeFunc func() error
+}
+
+func newCloseReader(r io.Reader, closeFunc func() error) io.ReadCloser {
+	if r == nil {
+		panic(`BUG: reader is nil`)
+	}
+	return &closeReader{Reader: r, closeFunc: closeFunc}
+}
+
+func (c *closeReader) Close() error {
+	if c.closeFunc == nil {
+		return nil
+	}
+	return c.closeFunc()
+}
+
 // BodyWriter returns writer for populating request body.
 func (req *Request) BodyWriter() io.Writer {
 	req.w.r = req
@@ -963,7 +1008,7 @@ func WriteMultipartForm(w io.Writer, f *multipart.Form, boundary string) error {
 	// Do not care about memory allocations here, since multipart
 	// form processing is slow.
 	if len(boundary) == 0 {
-		panic("BUG: form boundary cannot be empty")
+		return errors.New("form boundary cannot be empty")
 	}
 
 	mw := multipart.NewWriter(w)
@@ -1068,6 +1113,7 @@ func (resp *Response) Reset() {
 	resp.raddr = nil
 	resp.laddr = nil
 	resp.ImmediateHeaderFlush = false
+	resp.StreamBody = false
 }
 
 func (resp *Response) resetSkipHeader() {
@@ -1360,7 +1406,7 @@ func (resp *Response) ReadLimitBody(r *bufio.Reader, maxBodySize int) error {
 		}
 	}
 
-	if resp.Header.ContentLength() == -1 {
+	if resp.Header.ContentLength() == -1 && !resp.StreamBody {
 		err = resp.Header.ReadTrailer(r)
 		if err != nil && err != io.EOF {
 			if isConnectionReset(err) {
@@ -1383,14 +1429,23 @@ func (resp *Response) ReadBody(r *bufio.Reader, maxBodySize int) (err error) {
 	contentLength := resp.Header.ContentLength()
 	if contentLength >= 0 {
 		bodyBuf.B, err = readBody(r, contentLength, maxBodySize, bodyBuf.B)
-
+		if err == ErrBodyTooLarge && resp.StreamBody {
+			resp.bodyStream = acquireRequestStream(bodyBuf, r, &resp.Header)
+			err = nil
+		}
 	} else if contentLength == -1 {
-		bodyBuf.B, err = readBodyChunked(r, maxBodySize, bodyBuf.B)
-
+		if resp.StreamBody {
+			resp.bodyStream = acquireRequestStream(bodyBuf, r, &resp.Header)
+		} else {
+			bodyBuf.B, err = readBodyChunked(r, maxBodySize, bodyBuf.B)
+		}
 	} else {
 		bodyBuf.B, err = readBodyIdentity(r, maxBodySize, bodyBuf.B)
 		resp.Header.SetContentLength(len(bodyBuf.B))
 	}
+	if err == nil && resp.StreamBody && resp.bodyStream == nil {
+		resp.bodyStream = bytes.NewReader(bodyBuf.B)
+	}
 	return err
 }
 
@@ -1952,6 +2007,9 @@ func (resp *Response) closeBodyStream() error {
 	if bsc, ok := resp.bodyStream.(io.Closer); ok {
 		err = bsc.Close()
 	}
+	if bsr, ok := resp.bodyStream.(*requestStream); ok {
+		releaseRequestStream(bsr)
+	}
 	resp.bodyStream = nil
 	return err
 }
@@ -2134,13 +2192,14 @@ func readBodyIdentity(r *bufio.Reader, maxBodySize int, dst []byte) ([]byte, err
 	for {
 		nn, err := r.Read(dst[offset:])
 		if nn <= 0 {
-			if err != nil {
-				if err == io.EOF {
-					return dst[:offset], nil
-				}
+			switch {
+			case errors.Is(err, io.EOF):
+				return dst[:offset], nil
+			case err != nil:
 				return dst[:offset], err
+			default:
+				return dst[:offset], fmt.Errorf("bufio.Read() returned (%d, nil)", nn)
 			}
-			panic(fmt.Sprintf("BUG: bufio.Read() returned (%d, nil)", nn))
 		}
 		offset += nn
 		if maxBodySize > 0 && offset > maxBodySize {
@@ -2175,13 +2234,14 @@ func appendBodyFixedSize(r *bufio.Reader, dst []byte, n int) ([]byte, error) {
 	for {
 		nn, err := r.Read(dst[offset:])
 		if nn <= 0 {
-			if err != nil {
-				if err == io.EOF {
-					err = io.ErrUnexpectedEOF
-				}
+			switch {
+			case errors.Is(err, io.EOF):
+				return dst[:offset], io.ErrUnexpectedEOF
+			case err != nil:
 				return dst[:offset], err
+			default:
+				return dst[:offset], fmt.Errorf("bufio.Read() returned (%d, nil)", nn)
 			}
-			panic(fmt.Sprintf("BUG: bufio.Read() returned (%d, nil)", nn))
 		}
 		offset += nn
 		if offset == dstLen {
@@ -2197,6 +2257,8 @@ type ErrBrokenChunk struct {
 
 func readBodyChunked(r *bufio.Reader, maxBodySize int, dst []byte) ([]byte, error) {
 	if len(dst) > 0 {
+		// data integrity might be in danger. No idea what we received,
+		// but nothing we should write to.
 		panic("BUG: expected zero-length buffer")
 	}
 
diff --git a/lbclient.go b/lbclient.go
index bc0b63c..6be2dc9 100644
--- a/lbclient.go
+++ b/lbclient.go
@@ -84,6 +84,7 @@ func (cc *LBClient) init() {
 	cc.mu.Lock()
 	defer cc.mu.Unlock()
 	if len(cc.Clients) == 0 {
+		// developer sanity-check
 		panic("BUG: LBClient.Clients cannot be empty")
 	}
 	for _, c := range cc.Clients {
@@ -112,16 +113,14 @@ func (cc *LBClient) AddClient(c BalancingClient) int {
 func (cc *LBClient) RemoveClients(rc func(BalancingClient) bool) int {
 	cc.mu.Lock()
 	n := 0
-	for _, cs := range cc.cs {
+	for idx, cs := range cc.cs {
+		cc.cs[idx] = nil
 		if rc(cs.c) {
 			continue
 		}
 		cc.cs[n] = cs
 		n++
 	}
-	for i := n; i < len(cc.cs); i++ {
-		cc.cs[i] = nil
-	}
 	cc.cs = cc.cs[:n]
 
 	cc.mu.Unlock()
diff --git a/peripconn.go b/peripconn.go
index c2d1827..123c55e 100644
--- a/peripconn.go
+++ b/peripconn.go
@@ -1,7 +1,6 @@
 package fasthttp
 
 import (
-	"fmt"
 	"net"
 	"sync"
 )
@@ -25,17 +24,16 @@ func (cc *perIPConnCounter) Register(ip uint32) int {
 
 func (cc *perIPConnCounter) Unregister(ip uint32) {
 	cc.lock.Lock()
+	defer cc.lock.Unlock()
 	if cc.m == nil {
-		cc.lock.Unlock()
+		// developer safeguard
 		panic("BUG: perIPConnCounter.Register() wasn't called")
 	}
 	n := cc.m[ip] - 1
 	if n < 0 {
-		cc.lock.Unlock()
-		panic(fmt.Sprintf("BUG: negative per-ip counter=%d for ip=%d", n, ip))
+		n = 0
 	}
 	cc.m[ip] = n
-	cc.lock.Unlock()
 }
 
 type perIPConn struct {
diff --git a/server.go b/server.go
index 1c1c39d..4ed82b1 100644
--- a/server.go
+++ b/server.go
@@ -1929,9 +1929,6 @@ func acceptConn(s *Server, ln net.Listener, lastPerIPErrorTime *time.Time) (net.
 	for {
 		c, err := ln.Accept()
 		if err != nil {
-			if c != nil {
-				panic("BUG: net.Listener returned non-nil conn and non-nil error")
-			}
 			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
 				s.logger().Printf("Timeout error when accepting new connections: %v", netErr)
 				time.Sleep(time.Second)
@@ -1943,9 +1940,6 @@ func acceptConn(s *Server, ln net.Listener, lastPerIPErrorTime *time.Time) (net.
 			}
 			return nil, io.EOF
 		}
-		if c == nil {
-			panic("BUG: net.Listener returned (nil, nil)")
-		}
 
 		if tc, ok := c.(*net.TCPConn); ok && s.TCPKeepalive {
 			if err := tc.SetKeepAlive(s.TCPKeepalive); err != nil {
@@ -2578,7 +2572,7 @@ func (ctx *RequestCtx) LastTimeoutErrorResponse() *Response {
 
 func writeResponse(ctx *RequestCtx, w *bufio.Writer) error {
 	if ctx.timeoutResponse != nil {
-		panic("BUG: cannot write timed out response")
+		return errors.New("cannot write timed out response")
 	}
 	err := ctx.Response.Write(w)
 
@@ -2596,8 +2590,8 @@ func acquireByteReader(ctxP **RequestCtx) (*bufio.Reader, error) {
 	c := ctx.c
 	s.releaseCtx(ctx)
 
-	// Make GC happy, so it could garbage collect ctx
-	// while we waiting for the next request.
+	// Make GC happy, so it could garbage collect ctx while we wait for the
+	// next request.
 	ctx = nil
 	*ctxP = nil
 
@@ -2612,6 +2606,7 @@ func acquireByteReader(ctxP **RequestCtx) (*bufio.Reader, error) {
 		return nil, io.EOF
 	}
 	if n != 1 {
+		// developer sanity-check
 		panic("BUG: Reader must return at least one byte")
 	}
 
@@ -2787,19 +2782,23 @@ func (fa *fakeAddrer) LocalAddr() net.Addr {
 }
 
 func (fa *fakeAddrer) Read(p []byte) (int, error) {
+	// developer sanity-check
 	panic("BUG: unexpected Read call")
 }
 
 func (fa *fakeAddrer) Write(p []byte) (int, error) {
+	// developer sanity-check
 	panic("BUG: unexpected Write call")
 }
 
 func (fa *fakeAddrer) Close() error {
+	// developer sanity-check
 	panic("BUG: unexpected Close call")
 }
 
 func (s *Server) releaseCtx(ctx *RequestCtx) {
 	if ctx.timeoutResponse != nil {
+		// developer sanity-check
 		panic("BUG: cannot release timed out RequestCtx")
 	}
 
diff --git a/stackless/func.go b/stackless/func.go
index a50b3eb..70521e1 100644
--- a/stackless/func.go
+++ b/stackless/func.go
@@ -20,6 +20,7 @@ import (
 // at the moment due to high load.
 func NewFunc(f func(ctx interface{})) func(ctx interface{}) bool {
 	if f == nil {
+		// developer sanity-check
 		panic("BUG: f cannot be nil")
 	}
 
diff --git a/streaming.go b/streaming.go
index fc04916..a0a374b 100644
--- a/streaming.go
+++ b/streaming.go
@@ -9,8 +9,13 @@ import (
 	"github.com/valyala/bytebufferpool"
 )
 
+type headerInterface interface {
+	ContentLength() int
+	ReadTrailer(r *bufio.Reader) error
+}
+
 type requestStream struct {
-	header          *RequestHeader
+	header          headerInterface
 	prefetchedBytes *bytes.Reader
 	reader          *bufio.Reader
 	totalBytesRead  int
@@ -22,7 +27,7 @@ func (rs *requestStream) Read(p []byte) (int, error) {
 		n   int
 		err error
 	)
-	if rs.header.contentLength == -1 {
+	if rs.header.ContentLength() == -1 {
 		if rs.chunkLeft == 0 {
 			chunkSize, err := parseChunkSize(rs.reader)
 			if err != nil {
@@ -52,7 +57,7 @@ func (rs *requestStream) Read(p []byte) (int, error) {
 		}
 		return n, err
 	}
-	if rs.totalBytesRead == rs.header.contentLength {
+	if rs.totalBytesRead == rs.header.ContentLength() {
 		return 0, io.EOF
 	}
 	prefetchedSize := int(rs.prefetchedBytes.Size())
@@ -63,12 +68,12 @@ func (rs *requestStream) Read(p []byte) (int, error) {
 		}
 		n, err := rs.prefetchedBytes.Read(p)
 		rs.totalBytesRead += n
-		if n == rs.header.contentLength {
+		if n == rs.header.ContentLength() {
 			return n, io.EOF
 		}
 		return n, err
 	} else {
-		left := rs.header.contentLength - rs.totalBytesRead
+		left := rs.header.ContentLength() - rs.totalBytesRead
 		if len(p) > left {
 			p = p[:left]
 		}
@@ -79,13 +84,13 @@ func (rs *requestStream) Read(p []byte) (int, error) {
 		}
 	}
 
-	if rs.totalBytesRead == rs.header.contentLength {
+	if rs.totalBytesRead == rs.header.ContentLength() {
 		err = io.EOF
 	}
 	return n, err
 }
 
-func acquireRequestStream(b *bytebufferpool.ByteBuffer, r *bufio.Reader, h *RequestHeader) *requestStream {
+func acquireRequestStream(b *bytebufferpool.ByteBuffer, r *bufio.Reader, h headerInterface) *requestStream {
 	rs := requestStreamPool.Get().(*requestStream)
 	rs.prefetchedBytes = bytes.NewReader(b.B)
 	rs.reader = r
diff --git a/timer.go b/timer.go
index 885aa69..6c06ba0 100644
--- a/timer.go
+++ b/timer.go
@@ -10,6 +10,7 @@ func initTimer(t *time.Timer, timeout time.Duration) *time.Timer {
 		return time.NewTimer(timeout)
 	}
 	if t.Reset(timeout) {
+		// developer sanity-check
 		panic("BUG: active timer trapped into initTimer()")
 	}
 	return t
diff --git a/workerpool.go b/workerpool.go
index 50a5c75..71da1f2 100644
--- a/workerpool.go
+++ b/workerpool.go
@@ -47,7 +47,7 @@ type workerChan struct {
 
 func (wp *workerPool) Start() {
 	if wp.stopCh != nil {
-		panic("BUG: workerPool already started")
+		return
 	}
 	wp.stopCh = make(chan struct{})
 	stopCh := wp.stopCh
@@ -72,7 +72,7 @@ func (wp *workerPool) Start() {
 
 func (wp *workerPool) Stop() {
 	if wp.stopCh == nil {
-		panic("BUG: workerPool wasn't started")
+		return
 	}
 	close(wp.stopCh)
 	wp.stopCh = nil


Test patch:
diff --git a/http_test.go b/http_test.go
index ebbfa82..32194a8 100644
--- a/http_test.go
+++ b/http_test.go
@@ -2946,6 +2946,125 @@ func TestResponseBodyStreamErrorOnPanicDuringClose(t *testing.T) {
 	}
 }
 
+func TestResponseBodyStream(t *testing.T) {
+	t.Parallel()
+	chunkedResp := "HTTP/1.1 200 OK\r\n" + "Transfer-Encoding: chunked\r\n" + "\r\n" + "6\r\n123456\r\n" + "7\r\n1234567\r\n" + "0\r\n\r\n"
+	simpleResp := "HTTP/1.1 200 OK\r\n" + "Content-Length: 9\r\n" + "\r\n" + "123456789"
+	t.Run("read chunked response", func(t *testing.T) {
+		response := AcquireResponse()
+		response.StreamBody = true
+		if err := response.Read(bufio.NewReader(bytes.NewBufferString(chunkedResp))); err != nil {
+			t.Fatalf("parse response find err: %v", err)
+		}
+		defer func() {
+			if err := response.closeBodyStream(); err != nil {
+				t.Fatalf("close body stream err: %v", err)
+			}
+		}()
+		body, err := io.ReadAll(response.bodyStream)
+		if err != nil {
+			t.Fatalf("read body stream err: %v", err)
+		}
+		if string(body) != "1234561234567" {
+			t.Fatalf("unexpected body content, got: %#v, want: %#v", string(body), "1234561234567")
+		}
+	})
+	t.Run("read simple response", func(t *testing.T) {
+		resp := AcquireResponse()
+		resp.StreamBody = true
+		err := resp.ReadLimitBody(bufio.NewReader(bytes.NewBufferString(simpleResp)), 8)
+		if err != nil {
+			t.Fatalf("read limit body err: %v", err)
+		}
+		body := resp.BodyStream()
+		defer func() {
+			if err := resp.CloseBodyStream(); err != nil {
+				t.Fatalf("close body stream err: %v", err)
+			}
+		}()
+		content, err := io.ReadAll(body)
+		if err != nil {
+			t.Fatalf("read limit body err: %v", err)
+		}
+		if string(content) != "123456789" {
+			t.Fatalf("unexpected body content, got: %#v, want: %#v", string(content), "123456789")
+		}
+	})
+	t.Run("http client", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+			if request.URL.Query().Get("chunked") == "true" {
+				for x := 0; x < 10; x++ {
+					time.Sleep(time.Millisecond)
+					writer.Write([]byte(strconv.Itoa(x))) //nolint:errcheck
+					writer.(http.Flusher).Flush()
+				}
+				return
+			}
+			writer.Write([]byte(`hello world`)) //nolint:errcheck
+		}))
+
+		defer server.Close()
+
+		t.Run("normal request", func(t *testing.T) {
+			client := Client{StreamResponseBody: true}
+			resp := AcquireResponse()
+			request := AcquireRequest()
+			request.SetRequestURI(server.URL)
+			if err := client.Do(request, resp); err != nil {
+				t.Fatal(err)
+			}
+			stream := resp.BodyStream()
+			defer func() {
+				ReleaseResponse(resp)
+			}()
+			content, _ := io.ReadAll(stream)
+			if string(content) != "hello world" {
+				t.Fatalf("unexpected body content, got: %#v, want: %#v", string(content), "hello world")
+			}
+		})
+
+		t.Run("limit response body size size", func(t *testing.T) {
+			client := Client{StreamResponseBody: true, MaxResponseBodySize: 20}
+			resp := AcquireResponse()
+			request := AcquireRequest()
+			request.SetRequestURI(server.URL)
+			if err := client.Do(request, resp); err != nil {
+				t.Fatal(err)
+			}
+			stream := resp.BodyStream()
+			defer func() {
+				if err := resp.CloseBodyStream(); err != nil {
+					t.Fatalf("close stream err: %v", err)
+				}
+			}()
+			content, _ := io.ReadAll(stream)
+			if string(content) != "hello world" {
+				t.Fatalf("unexpected body content, got: %#v, want: %#v", string(content), "hello world")
+			}
+		})
+
+		t.Run("chunked", func(t *testing.T) {
+			client := Client{StreamResponseBody: true}
+			resp := AcquireResponse()
+			request := AcquireRequest()
+			request.SetRequestURI(server.URL + "?chunked=true")
+			if err := client.Do(request, resp); err != nil {
+				t.Fatal(err)
+			}
+			stream := resp.BodyStream()
+			defer func() {
+				if err := resp.CloseBodyStream(); err != nil {
+					t.Fatalf("close stream err: %v", err)
+				}
+			}()
+			content, _ := io.ReadAll(stream)
+			if string(content) != "0123456789" {
+				t.Fatalf("unexpected body content, got: %#v, want: %#v", string(content), "0123456789")
+			}
+		})
+	})
+}
+
 func TestRequestMultipartFormPipeEmptyFormField(t *testing.T) {
 	t.Parallel()
 
diff --git a/peripconn_test.go b/peripconn_test.go
index e2137c0..5571654 100644
--- a/peripconn_test.go
+++ b/peripconn_test.go
@@ -25,8 +25,6 @@ func TestPerIPConnCounter(t *testing.T) {
 
 	var cc perIPConnCounter
 
-	expectPanic(t, func() { cc.Unregister(123) })
-
 	for i := 1; i < 100; i++ {
 		if n := cc.Register(123); n != i {
 			t.Fatalf("Unexpected counter value=%d. Expected %d", n, i)
@@ -43,21 +41,9 @@ func TestPerIPConnCounter(t *testing.T) {
 	}
 	cc.Unregister(456)
 
-	expectPanic(t, func() { cc.Unregister(123) })
-	expectPanic(t, func() { cc.Unregister(456) })
-
 	n = cc.Register(123)
 	if n != 1 {
 		t.Fatalf("Unexpected counter value=%d. Expected 1", n)
 	}
 	cc.Unregister(123)
 }
-
-func expectPanic(t *testing.T, f func()) {
-	defer func() {
-		if r := recover(); r == nil {
-			t.Fatalf("Expecting panic")
-		}
-	}()
-	f()
-}