Repository: cloudwego/hertz
Pull request #1329 (2025-04-11 10:23:38)

fix(client): stream body for sse instead of timeout
#### What type of PR is this?
fix

<!--
Add one of the following kinds:

build: Changes that affect the build system or external dependencies (example scopes: gulp, broccoli, npm)
ci: Changes to our CI configuration files and scripts (example scopes: Travis, Circle, BrowserStack, SauceLabs)
docs: Documentation only changes
feat: A new feature
optimize: A new optimization
fix: A bug fix
perf: A code change that improves performance
refactor: A code change that neither fixes a bug nor adds a feature
style: Changes that do not affect the meaning of the code (white space, formatting, missing semi-colons, etc)
test: Adding missing tests or correcting existing tests
chore: Changes to the build process or auxiliary tools and libraries such as documentation generation
-->

#### Check the PR title.
<!--
The description of the title will be attached in Release Notes, 
so please describe it from user-oriented, what this PR does / why we need it.
Please check your PR title with the below requirements:
-->
- [ ] This PR title match the format: \<type\>(optional scope): \<description\>
- [ ] The description of this PR title is user-oriented and clear enough for others to understand.
- [ ] Attach the PR updating the user documentation if the current PR requires user awareness at the usage level. [User docs repo](https://github.com/cloudwego/cloudwego.github.io)


#### (Optional) Translate the PR title into Chinese.


#### (Optional) More detailed description for this PR(en: English/zh: Chinese).
<!--
Provide more detailed info for review(e.g., it's recommended to provide perf data if this is a perf type PR).
-->
en:
zh(optional):

#### (Optional) Which issue(s) this PR fixes:
<!--
Automatically closes linked issue when PR is merged.
Eg: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->

#### (Optional) The PR that updates user documentation:
<!--
If the current PR requires user awareness at the usage level, please submit a PR to update user docs. [User docs repo](https://github.com/cloudwego/cloudwego.github.io)
-->

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) Report
Attention: Patch coverage is `81.63265%` with `9 lines` in your changes missing coverage. Please review.
> Project coverage is 72.78%. Comparing base [(`67fe3c7`)](https://app.codecov.io/gh/cloudwego/hertz/commit/67fe3c7837092580627a2b3f45f8adf26304a694?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) to head [(`a5b4b3d`)](https://app.codecov.io/gh/cloudwego/hertz/commit/a5b4b3d52f8f071c45e1030cae017a5864b99ac4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).

| [Files with missing lines](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) | Patch % | Lines |
|---|---|---|
| [pkg/protocol/http1/client.go](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?src=pr&el=tree&filepath=pkg%2Fprotocol%2Fhttp1%2Fclient.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL2NsaWVudC5nbw==) | 68.75% | [4 Missing and 1 partial :warning: ](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) |
| [pkg/protocol/http1/resp/response.go](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?src=pr&el=tree&filepath=pkg%2Fprotocol%2Fhttp1%2Fresp%2Fresponse.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2h0dHAxL3Jlc3AvcmVzcG9uc2UuZ28=) | 92.00% | [0 Missing and 2 partials :warning: ](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) |
| [pkg/protocol/client/client.go](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?src=pr&el=tree&filepath=pkg%2Fprotocol%2Fclient%2Fclient.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL2NsaWVudC9jbGllbnQuZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) |
| [pkg/protocol/request.go](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?src=pr&el=tree&filepath=pkg%2Fprotocol%2Frequest.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego#diff-cGtnL3Byb3RvY29sL3JlcXVlc3QuZ28=) | 66.66% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           develop    #1329      +/-   ##
===========================================
- Coverage    72.80%   72.78%   -0.02%     
===========================================
  Files          139      139              
  Lines        16244    16259      +15     
===========================================
+ Hits         11826    11834       +8     
- Misses        3823     3827       +4     
- Partials       595      598       +3     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/cloudwego/hertz/pull/1329?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=cloudwego).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>

Patch:
diff --git a/pkg/protocol/client/client.go b/pkg/protocol/client/client.go
index a5b2c88..e0ed457 100644
--- a/pkg/protocol/client/client.go
+++ b/pkg/protocol/client/client.go
@@ -184,7 +184,7 @@ func GetURLDeadline(ctx context.Context, dst []byte, url string, deadline time.T
 func PostURL(ctx context.Context, dst []byte, url string, postArgs *protocol.Args, c Doer, requestOptions ...config.RequestOption) (statusCode int, body []byte, err error) {
 	req := protocol.AcquireRequest()
 	req.Header.SetMethodBytes(bytestr.StrPost)
-	req.Header.SetContentTypeBytes(bytestr.StrPostArgsContentType)
+	req.Header.SetContentTypeBytes(bytestr.MIMEPostForm)
 	req.SetOptions(requestOptions...)
 
 	if postArgs != nil {
diff --git a/pkg/protocol/header.go b/pkg/protocol/header.go
index e68ec5c..6993494 100644
--- a/pkg/protocol/header.go
+++ b/pkg/protocol/header.go
@@ -404,7 +404,7 @@ func (h *RequestHeader) AppendBytes(dst []byte) []byte {
 
 	contentType := h.ContentType()
 	if len(contentType) == 0 && !h.IgnoreBody() && !h.noDefaultContentType {
-		contentType = bytestr.StrPostArgsContentType
+		contentType = bytestr.MIMEPostForm
 	}
 	if len(contentType) > 0 {
 		dst = appendHeaderLine(dst, bytestr.StrContentType, contentType)
@@ -1144,7 +1144,7 @@ func (h *RequestHeader) Peek(key string) []byte {
 // where ... is substituted by the given boundary.
 func (h *RequestHeader) SetMultipartFormBoundary(boundary string) {
 	b := h.bufKV.value[:0]
-	b = append(b, bytestr.StrMultipartFormData...)
+	b = append(b, bytestr.MIMEFormData...)
 	b = append(b, ';', ' ')
 	b = append(b, bytestr.StrBoundary...)
 	b = append(b, '=')
@@ -1202,10 +1202,10 @@ func (h *RequestHeader) InitContentLengthWithValue(contentLength int) {
 // from 'multipart/form-data; boundary=...' Content-Type.
 func (h *RequestHeader) MultipartFormBoundary() []byte {
 	b := h.ContentType()
-	if !bytes.HasPrefix(b, bytestr.StrMultipartFormData) {
+	if !bytes.HasPrefix(b, bytestr.MIMEFormData) {
 		return nil
 	}
-	b = b[len(bytestr.StrMultipartFormData):]
+	b = b[len(bytestr.MIMEFormData):]
 	if len(b) == 0 || b[0] != ';' {
 		return nil
 	}
diff --git a/pkg/protocol/http1/client.go b/pkg/protocol/http1/client.go
index eb57d0d..c49da74 100644
--- a/pkg/protocol/http1/client.go
+++ b/pkg/protocol/http1/client.go
@@ -683,10 +683,24 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 	// This is to solve the circular dependency problem of Response and BodyStream
 	shouldCloseConn := false
 
-	if !c.ResponseBodyStream {
-		err = respI.ReadHeaderAndLimitBody(resp, zr, c.MaxResponseBodySize)
+	if err = respI.ReadHeaders(resp, zr); err != nil {
+		_ = zr.Release()
+		c.closeConn(cc)
+		return true, err
+	}
+
+	stream := c.ResponseBodyStream
+
+	// if it's server-sent event response,
+	// we should set stream=true or it may block till timeout
+	if !stream && resp.Header.ContentLength() < 0 &&
+		bytes.HasPrefix(resp.Header.ContentType(), bytestr.MIMETextEventStream) {
+		stream = true
+	}
+	if !stream {
+		err = respI.ReadRespBody(resp, zr, c.MaxResponseBodySize)
 	} else {
-		err = respI.ReadBodyStream(resp, zr, c.MaxResponseBodySize, func(shouldClose bool) error {
+		err = respI.ReadRespBodyStream(resp, zr, c.MaxResponseBodySize, func(shouldClose bool) error {
 			if shouldCloseConn || shouldClose {
 				c.closeConn(cc)
 			} else {
@@ -714,7 +728,7 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 	}
 
 	// In stream mode, we still can close/release the connection immediately if there is no content on the wire.
-	if c.ResponseBodyStream && resp.BodyStream() != protocol.NoResponseBody {
+	if stream && resp.BodyStream() != protocol.NoResponseBody {
 		return false, nil
 	}
 
diff --git a/pkg/protocol/http1/resp/response.go b/pkg/protocol/http1/resp/response.go
index 0366336..c55743f 100644
--- a/pkg/protocol/http1/resp/response.go
+++ b/pkg/protocol/http1/resp/response.go
@@ -89,13 +89,8 @@ func GetHTTP1Response(resp *protocol.Response) fmt.Stringer {
 	return &h1Response{resp}
 }
 
-// ReadHeaderAndLimitBody reads response from the given r, limiting the body size.
-//
-// If maxBodySize > 0 and the body size exceeds maxBodySize,
-// then ErrBodyTooLarge is returned.
-//
-// io.EOF is returned if r is closed before reading the first header byte.
-func ReadHeaderAndLimitBody(resp *protocol.Response, r network.Reader, maxBodySize int) error {
+// ReadHeaders reads http header into *protocol.Response
+func ReadHeaders(resp *protocol.Response, r network.Reader) error {
 	resp.ResetBody()
 	err := ReadHeader(&resp.Header, r)
 	if err != nil {
@@ -107,23 +102,40 @@ func ReadHeaderAndLimitBody(resp *protocol.Response, r network.Reader, maxBodySi
 			return err
 		}
 	}
+	return nil
+}
 
-	if !resp.MustSkipBody() {
-		bodyBuf := resp.BodyBuffer()
-		bodyBuf.Reset()
-		bodyBuf.B, err = ext.ReadBody(r, resp.Header.ContentLength(), maxBodySize, bodyBuf.B)
-		if err != nil {
+// ReadHeaderAndLimitBody ...
+func ReadHeaderAndLimitBody(resp *protocol.Response, r network.Reader, maxBodySize int) error {
+	if err := ReadHeaders(resp, r); err != nil {
+		return err
+	}
+	return ReadRespBody(resp, r, maxBodySize)
+}
+
+// ReadRespBody reads response body from the given r, limiting the body size.
+//
+// If maxBodySize > 0 and the body size exceeds maxBodySize,
+// then ErrBodyTooLarge is returned.
+//
+// io.EOF is returned if r is closed before reading the first header byte.
+func ReadRespBody(resp *protocol.Response, r network.Reader, maxBodySize int) (err error) {
+	if resp.MustSkipBody() {
+		return nil
+	}
+	bodyBuf := resp.BodyBuffer()
+	bodyBuf.Reset()
+	bodyBuf.B, err = ext.ReadBody(r, resp.Header.ContentLength(), maxBodySize, bodyBuf.B)
+	if err != nil {
+		return err
+	}
+	if resp.Header.ContentLength() == -1 {
+		err = ext.ReadTrailer(resp.Header.Trailer(), r)
+		if err != nil && err != io.EOF {
 			return err
 		}
-		if resp.Header.ContentLength() == -1 {
-			err = ext.ReadTrailer(resp.Header.Trailer(), r)
-			if err != nil && err != io.EOF {
-				return err
-			}
-		}
-		resp.Header.SetContentLength(len(bodyBuf.B))
 	}
-
+	resp.Header.SetContentLength(len(bodyBuf.B))
 	return nil
 }
 
@@ -173,25 +185,24 @@ func convertClientRespStream(bs io.Reader, fn func(shouldClose bool) error) *cli
 	return clientStream
 }
 
-// ReadBodyStream reads response body in stream
-func ReadBodyStream(resp *protocol.Response, r network.Reader, maxBodySize int, closeCallBack func(shouldClose bool) error) error {
-	resp.ResetBody()
-	err := ReadHeader(&resp.Header, r)
-	if err != nil {
+// ReadHeaderBodyStream ...
+func ReadHeaderBodyStream(resp *protocol.Response, r network.Reader,
+	maxBodySize int, closeCallBack func(shouldClose bool) error) error {
+	if err := ReadHeaders(resp, r); err != nil {
 		return err
 	}
+	return ReadRespBodyStream(resp, r, maxBodySize, closeCallBack)
+}
 
-	if resp.Header.StatusCode() == consts.StatusContinue {
-		// Read the next response according to http://www.w3.org/Protocols/rfc2616/rfc2616-sec8.html .
-		if err = ReadHeader(&resp.Header, r); err != nil {
-			return err
-		}
-	}
+// Deprecated: use ReadHeaderBodyStream
+var ReadBodyStream = ReadHeaderBodyStream
 
+// ReadRespBodyStream reads response body in stream
+func ReadRespBodyStream(resp *protocol.Response, r network.Reader,
+	maxBodySize int, closeCallBack func(shouldClose bool) error) (err error) {
 	if resp.MustSkipBody() {
 		return nil
 	}
-
 	bodyBuf := resp.BodyBuffer()
 	bodyBuf.Reset()
 	bodyBuf.B, err = ext.ReadBodyWithStreaming(r, resp.Header.ContentLength(), maxBodySize, bodyBuf.B)
@@ -211,7 +222,6 @@ func ReadBodyStream(resp *protocol.Response, r network.Reader, maxBodySize int,
 		resp.Reset()
 		return err
 	}
-
 	bodyStream := ext.AcquireBodyStream(bodyBuf, r, resp.Header.Trailer(), resp.Header.ContentLength())
 	resp.ConstructBodyStream(bodyBuf, convertClientRespStream(bodyStream, closeCallBack))
 	return nil
diff --git a/pkg/protocol/request.go b/pkg/protocol/request.go
index 1bec09f..9c20233 100644
--- a/pkg/protocol/request.go
+++ b/pkg/protocol/request.go
@@ -464,7 +464,7 @@ func (req *Request) SetFormData(data map[string]string) {
 		req.postArgs.Add(k, v)
 	}
 	req.parsedPostArgs = true
-	req.Header.SetContentTypeBytes(bytestr.StrPostArgsContentType)
+	req.Header.SetContentTypeBytes(bytestr.MIMEPostForm)
 }
 
 // SetFormDataFromValues sets x-www-form-urlencoded params from url values.
@@ -475,7 +475,7 @@ func (req *Request) SetFormDataFromValues(data url.Values) {
 		}
 	}
 	req.parsedPostArgs = true
-	req.Header.SetContentTypeBytes(bytestr.StrPostArgsContentType)
+	req.Header.SetContentTypeBytes(bytestr.MIMEPostForm)
 }
 
 // SetFile sets single file field name and its path for multipart upload.
@@ -684,8 +684,7 @@ func (req *Request) parsePostArgs() {
 		return
 	}
 	req.parsedPostArgs = true
-
-	if !bytes.HasPrefix(req.Header.ContentType(), bytestr.StrPostArgsContentType) {
+	if !bytes.HasPrefix(req.Header.ContentType(), bytestr.MIMEPostForm) {
 		return
 	}
 	req.postArgs.ParseBytes(req.Body())


Test patch:
diff --git a/internal/bytestr/bytes.go b/internal/bytestr/bytes.go
index bbf3dcb..bdf530f 100644
--- a/internal/bytestr/bytes.go
+++ b/internal/bytestr/bytes.go
@@ -99,22 +99,26 @@ var (
 	StrCookieSameSiteNone   = []byte("None")
 	StrCookiePartitioned    = []byte("Partitioned")
 
-	StrClose               = []byte("close")
-	StrGzip                = []byte("gzip")
-	StrDeflate             = []byte("deflate")
-	StrKeepAlive           = []byte("keep-alive")
-	StrUpgrade             = []byte("Upgrade")
-	StrChunked             = []byte("chunked")
-	StrIdentity            = []byte("identity")
-	Str100Continue         = []byte("100-continue")
-	StrPostArgsContentType = []byte("application/x-www-form-urlencoded")
-	StrMultipartFormData   = []byte("multipart/form-data")
-	StrBoundary            = []byte("boundary")
-	StrBytes               = []byte("bytes")
-	StrTextSlash           = []byte("text/")
-	StrApplicationSlash    = []byte("application/")
-	StrBasicSpace          = []byte("Basic ")
+	StrClose            = []byte("close")
+	StrGzip             = []byte("gzip")
+	StrDeflate          = []byte("deflate")
+	StrKeepAlive        = []byte("keep-alive")
+	StrUpgrade          = []byte("Upgrade")
+	StrChunked          = []byte("chunked")
+	StrIdentity         = []byte("identity")
+	Str100Continue      = []byte("100-continue")
+	StrBoundary         = []byte("boundary")
+	StrBytes            = []byte("bytes")
+	StrTextSlash        = []byte("text/")
+	StrApplicationSlash = []byte("application/")
+	StrBasicSpace       = []byte("Basic ")
 
 	// http2
 	StrClientPreface = []byte(consts.ClientPreface)
 )
+
+var ( // content types
+	MIMEPostForm        = []byte("application/x-www-form-urlencoded")
+	MIMEFormData        = []byte("multipart/form-data")
+	MIMETextEventStream = []byte("text/event-stream") // for server-sent events
+)
diff --git a/pkg/protocol/http1/client_test.go b/pkg/protocol/http1/client_test.go
index 353cf61..715a2ce 100644
--- a/pkg/protocol/http1/client_test.go
+++ b/pkg/protocol/http1/client_test.go
@@ -64,6 +64,7 @@ import (
 	"github.com/cloudwego/hertz/pkg/common/test/mock"
 	"github.com/cloudwego/hertz/pkg/common/utils"
 	"github.com/cloudwego/hertz/pkg/network"
+	"github.com/cloudwego/hertz/pkg/network/dialer"
 	"github.com/cloudwego/hertz/pkg/protocol"
 	"github.com/cloudwego/hertz/pkg/protocol/client"
 	"github.com/cloudwego/hertz/pkg/protocol/consts"
@@ -145,6 +146,36 @@ func TestHostClientMaxConnWaitTimeoutWithEarlierDeadline(t *testing.T) {
 	}
 }
 
+func TestReadHeaderErr(t *testing.T) {
+	ln, _ := net.Listen("tcp", "localhost:0")
+	defer ln.Close()
+	svr := http.Server{}
+	svr.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hj := w.(http.Hijacker)
+		conn, rw, err := hj.Hijack()
+		assert.Nil(t, err)
+		defer conn.Close()
+		rw.Write([]byte("HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Ty"))
+		rw.Flush()
+	})
+	go svr.Serve(ln)
+
+	req := protocol.AcquireRequest()
+	defer protocol.ReleaseRequest(req)
+	req.SetRequestURI("http://" + ln.Addr().String())
+
+	resp := protocol.AcquireResponse()
+	defer protocol.ReleaseResponse(resp)
+	c := &HostClient{
+		Addr: ln.Addr().String(),
+		ClientOptions: &ClientOptions{
+			Dialer: dialer.DefaultDialer(),
+		},
+	}
+	err := c.Do(context.Background(), req, resp)
+	assert.NotNil(t, err)
+}
+
 func TestResponseReadBodyStream(t *testing.T) {
 	// small body
 	genBody := "abcdef4343"
@@ -182,7 +213,7 @@ func TestResponseReadBodyStream(t *testing.T) {
 func testContinueReadResponseBodyStream(t *testing.T, header, body string, maxBodySize, firstRead, leftBytes, bytesLeftInReader int) {
 	mr := netpoll.NewReader(bytes.NewBufferString(header + body))
 	var r protocol.Response
-	if err := resp.ReadBodyStream(&r, mr, maxBodySize, nil); err != nil {
+	if err := resp.ReadHeaderBodyStream(&r, mr, maxBodySize, nil); err != nil {
 		t.Fatalf("error when reading request body stream: %s", err)
 	}
 	fRead := firstRead
@@ -257,7 +288,7 @@ func TestReadTimeoutPriority(t *testing.T) {
 
 	req := protocol.AcquireRequest()
 	req.SetRequestURI("http://foobar/baz")
-	req.SetOptions(config.WithReadTimeout(time.Second * 1))
+	req.SetOptions(config.WithReadTimeout(200 * time.Millisecond))
 	resp := protocol.AcquireResponse()
 
 	ch := make(chan error, 1)
@@ -312,7 +343,52 @@ func TestDoNonNilReqResp1(t *testing.T) {
 	assert.NotNil(t, err)
 }
 
-func TestConnUpgrade(t *testing.T) {
+func doGET(t *testing.T, addr string, path string) *protocol.Response {
+	req := protocol.AcquireRequest()
+	defer protocol.ReleaseRequest(req)
+	req.SetRequestURI("http://" + addr + path)
+
+	resp := protocol.AcquireResponse()
+	c := &HostClient{
+		Addr: addr,
+		ClientOptions: &ClientOptions{
+			Dialer: dialer.DefaultDialer(),
+		},
+	}
+	err := c.Do(context.Background(), req, resp)
+	assert.Nil(t, err)
+	return resp
+}
+
+func TestStreamResponse_EventStream(t *testing.T) {
+	ln, _ := net.Listen("tcp", "localhost:0")
+	defer ln.Close()
+	svr := http.Server{}
+	svr.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		f := w.(http.Flusher)
+		for i := 0; i < 5; i++ {
+			_, err := w.Write([]byte(fmt.Sprintf("data:%d", i)))
+			assert.Nil(t, err)
+			f.Flush() // Transfer-Encoding chunked
+			time.Sleep(20 * time.Millisecond)
+		}
+	})
+	go svr.Serve(ln)
+
+	resp := doGET(t, ln.Addr().String(), "/")
+	defer protocol.ReleaseResponse(resp)
+	assert.Assert(t, resp.IsBodyStream())
+	r := resp.BodyStream()
+	b := make([]byte, 10)
+	for i := 0; i < 5; i++ {
+		n, err := r.Read(b)
+		assert.Nil(t, err)
+		assert.Assert(t, string(b[:n]) == fmt.Sprintf("data:%d", i))
+	}
+}
+
+func TestStreamResponse_ConnUpgrade(t *testing.T) {
 	ln, _ := net.Listen("tcp", "localhost:0")
 	defer ln.Close()
 	svr := http.Server{}
@@ -346,16 +422,8 @@ func TestConnUpgrade(t *testing.T) {
 	})
 	go svr.Serve(ln)
 
-	c := &HostClient{
-		Addr:          ln.Addr().String(),
-		ClientOptions: &ClientOptions{},
-	}
-	req := protocol.AcquireRequest()
-	req.SetRequestURI("http://" + ln.Addr().String() + "/")
-	resp := protocol.AcquireResponse()
-	retry, err := c.doNonNilReqResp(req, resp)
-	assert.False(t, retry)
-	assert.Nil(t, err)
+	resp := doGET(t, ln.Addr().String(), "/")
+	defer protocol.ReleaseResponse(resp)
 	assert.DeepEqual(t, resp.StatusCode(), 101)
 
 	s := resp.BodyStream()
@@ -385,7 +453,7 @@ func TestWriteTimeoutPriority(t *testing.T) {
 
 	req := protocol.AcquireRequest()
 	req.SetRequestURI("http://foobar/baz")
-	req.SetOptions(config.WithWriteTimeout(time.Second * 1))
+	req.SetOptions(config.WithWriteTimeout(200 * time.Millisecond))
 	resp := protocol.AcquireResponse()
 
 	ch := make(chan error, 1)
@@ -413,7 +481,7 @@ func TestDialTimeoutPriority(t *testing.T) {
 
 	req := protocol.AcquireRequest()
 	req.SetRequestURI("http://foobar/baz")
-	req.SetOptions(config.WithDialTimeout(time.Second * 1))
+	req.SetOptions(config.WithDialTimeout(200 * time.Millisecond))
 	resp := protocol.AcquireResponse()
 
 	ch := make(chan error, 1)
diff --git a/pkg/protocol/http1/resp/response_test.go b/pkg/protocol/http1/resp/response_test.go
index 3c206e4..ea21ea6 100644
--- a/pkg/protocol/http1/resp/response_test.go
+++ b/pkg/protocol/http1/resp/response_test.go
@@ -732,7 +732,7 @@ func testResponseReadBodyStreamSuccess(t *testing.T, resp *protocol.Response, re
 	expectedContentType, expectedBody string, expectedTrailer map[string]string, expectedProtocol string,
 ) {
 	zr := mock.NewZeroCopyReader(response)
-	err := ReadBodyStream(resp, zr, 0, nil)
+	err := ReadHeaderBodyStream(resp, zr, 0, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error: %s", err)
 	}
@@ -751,7 +751,7 @@ func testResponseReadBodyStreamSuccess(t *testing.T, resp *protocol.Response, re
 
 func testResponseReadBodyStreamBadTrailer(t *testing.T, resp *protocol.Response, response string) {
 	zr := mock.NewZeroCopyReader(response)
-	err := ReadBodyStream(resp, zr, 0, nil)
+	err := ReadHeaderBodyStream(resp, zr, 0, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error: %s", err)
 	}