Repository: cloudwego/hertz
Pull request #799 (2023-05-30 11:19:15)

feat(hz): add validator anno
#### What type of PR is this?
<!--
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
-->
feat
#### 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:
-->
- [x] This PR title match the format: `<type>(optional scope): <description>`.
- [x] The description of this PR title is user-oriented and clear enough for others to understand.
- [x] 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.
添加 Validator 注解

#### (Optional) More detail description for this PR(en: English/zh: Chinese).
<!--
Provide more detailed info for review. If it is a perf type PR, perf data is suggested to give.
-->
en:  Add Validator annotation to avoid possible conflicts 
zh(optional): 添加 Validator注解，避免可能出现的冲突

#### (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)
-->



Patch:
diff --git a/_typos.toml b/_typos.toml
index 113fc63..3d3103e 100644
--- a/_typos.toml
+++ b/_typos.toml
@@ -16,4 +16,6 @@ contant = "contant"
 referer = "referer"
 HeaderReferer = "HeaderReferer"
 expectedReferer = "expectedReferer"
-Referer = "Referer"
\ No newline at end of file
+Referer = "Referer"
+O_WRONLY = "O_WRONLY"
+WRONLY = "WRONLY"
\ No newline at end of file
diff --git a/cmd/hz/app/app.go b/cmd/hz/app/app.go
index 4196709..269e5ba 100644
--- a/cmd/hz/app/app.go
+++ b/cmd/hz/app/app.go
@@ -70,7 +70,7 @@ func New(c *cli.Context) error {
 		return cli.Exit(fmt.Errorf("persist manifest failed: %v", err), meta.PersistError)
 	}
 	if !args.NeedGoMod && args.IsNew() {
-		fmt.Println(meta.AddThriftReplace)
+		logs.Warn(meta.AddThriftReplace)
 	}
 
 	return nil
@@ -184,6 +184,7 @@ func Init() *cli.App {
 	unsetOmitemptyFlag := cli.BoolFlag{Name: "unset_omitempty", Usage: "Remove 'omitempty' tag for generated struct.", Destination: &globalArgs.UnsetOmitempty}
 	protoCamelJSONTag := cli.BoolFlag{Name: "pb_camel_json_tag", Usage: "Convert Name style for json tag to camel(Only works protobuf).", Destination: &globalArgs.ProtobufCamelJSONTag}
 	snakeNameFlag := cli.BoolFlag{Name: "snake_tag", Usage: "Use snake_case style naming for tags. (Only works for 'form', 'query', 'json')", Destination: &globalArgs.SnakeName}
+	rmTagFlag := cli.StringSliceFlag{Name: "rm_tag", Usage: "Remove the specified tag"}
 	customLayout := cli.StringFlag{Name: "customize_layout", Usage: "Specify the path for layout template.", Destination: &globalArgs.CustomizeLayout}
 	customLayoutData := cli.StringFlag{Name: "customize_layout_data_path", Usage: "Specify the path for layout template render data.", Destination: &globalArgs.CustomizeLayoutData}
 	customPackage := cli.StringFlag{Name: "customize_package", Usage: "Specify the path for package template.", Destination: &globalArgs.CustomizePackage}
@@ -230,6 +231,7 @@ func Init() *cli.App {
 				&unsetOmitemptyFlag,
 				&protoCamelJSONTag,
 				&snakeNameFlag,
+				&rmTagFlag,
 				&excludeFilesFlag,
 				&customLayout,
 				&customLayoutData,
@@ -263,6 +265,7 @@ func Init() *cli.App {
 				&unsetOmitemptyFlag,
 				&protoCamelJSONTag,
 				&snakeNameFlag,
+				&rmTagFlag,
 				&excludeFilesFlag,
 				&customPackage,
 				&handlerByMethod,
@@ -289,6 +292,7 @@ func Init() *cli.App {
 				&unsetOmitemptyFlag,
 				&protoCamelJSONTag,
 				&snakeNameFlag,
+				&rmTagFlag,
 				&excludeFilesFlag,
 			},
 			Action: Model,
@@ -315,6 +319,7 @@ func Init() *cli.App {
 				&unsetOmitemptyFlag,
 				&protoCamelJSONTag,
 				&snakeNameFlag,
+				&rmTagFlag,
 				&excludeFilesFlag,
 				&customLayout,
 				&customLayoutData,
diff --git a/cmd/hz/config/argument.go b/cmd/hz/config/argument.go
index 8eef041..b8bf41c 100644
--- a/cmd/hz/config/argument.go
+++ b/cmd/hz/config/argument.go
@@ -64,6 +64,7 @@ type Argument struct {
 	ProtobufPlugins      []string
 	ThriftPlugins        []string
 	SnakeName            bool
+	RmTags               []string
 	Excludes             []string
 	NoRecurse            bool
 	HandlerByMethod      bool
@@ -121,6 +122,7 @@ func (arg *Argument) parseStringSlice(c *cli.Context) {
 	arg.ProtocOptions = c.StringSlice("protoc")
 	arg.ThriftPlugins = c.StringSlice("thrift-plugins")
 	arg.ProtobufPlugins = c.StringSlice("protoc-plugins")
+	arg.RmTags = c.StringSlice("rm_tag")
 }
 
 func (arg *Argument) UpdateByManifest(m *meta.Manifest) {
diff --git a/cmd/hz/generator/router.go b/cmd/hz/generator/router.go
index 54c6d21..10f4316 100644
--- a/cmd/hz/generator/router.go
+++ b/cmd/hz/generator/router.go
@@ -415,11 +415,11 @@ func (pkgGen *HttpPackageGenerator) updateMiddlewareReg(router interface{}, midd
 	}
 
 	for _, mw := range middlewareList {
-		mwName := mw + "Mw"
+		mwNamePattern := fmt.Sprintf(" %sMw", mw)
 		if pkgGen.SnakeStyleMiddleware {
-			mwName = mw + "_mw"
+			mwNamePattern = fmt.Sprintf(" %s_mw", mw)
 		}
-		if bytes.Contains(file, []byte(mwName)) {
+		if bytes.Contains(file, []byte(mwNamePattern)) {
 			continue
 		}
 		middlewareSingleTpl := pkgGen.tpls[middlewareSingleTplName]
diff --git a/cmd/hz/generator/template_funcs.go b/cmd/hz/generator/template_funcs.go
index 359723d..556ffb1 100644
--- a/cmd/hz/generator/template_funcs.go
+++ b/cmd/hz/generator/template_funcs.go
@@ -17,6 +17,7 @@
 package generator
 
 import (
+	"strings"
 	"text/template"
 
 	"github.com/cloudwego/hertz/cmd/hz/util"
@@ -25,6 +26,8 @@ import (
 var funcMap = template.FuncMap{
 	"GetUniqueHandlerOutDir": getUniqueHandlerOutDir,
 	"ToSnakeCase":            util.ToSnakeCase,
+	"Split":                  strings.Split,
+	"Trim":                   strings.Trim,
 }
 
 // getUniqueHandlerOutDir uses to get unique "api.handler_path"
diff --git a/cmd/hz/meta/const.go b/cmd/hz/meta/const.go
index 75cacca..6ec98d6 100644
--- a/cmd/hz/meta/const.go
+++ b/cmd/hz/meta/const.go
@@ -19,7 +19,7 @@ package meta
 import "runtime"
 
 // Version hz version
-const Version = "v0.6.4"
+const Version = "v0.6.5"
 
 // Mode hz run modes
 type Mode int
diff --git a/cmd/hz/protobuf/api/api.proto b/cmd/hz/protobuf/api/api.proto
index a41e2f8..72edf41 100644
--- a/cmd/hz/protobuf/api/api.proto
+++ b/cmd/hz/protobuf/api/api.proto
@@ -24,6 +24,8 @@ extend google.protobuf.FieldOptions {
   optional string js_conv_compatible = 50132;
   optional string file_name_compatible = 50133;
   optional string none_compatible = 50134;
+  // 50135 is reserved to vt_compatible
+  // optional FieldRules vt_compatible = 50135;
 
   optional string go_tag = 51001;
 }
@@ -62,4 +64,12 @@ extend google.protobuf.ServiceOptions {
 
   // 50731~50760 used to extend service option by hz
   optional string base_domain_compatible = 50731;
+}
+
+extend google.protobuf.MessageOptions {
+  // optional FieldRules msg_vt = 50111;
+
+  optional string reserve = 50830;
+  // 550831 is reserved to msg_vt_compatible
+  // optional FieldRules msg_vt_compatible = 50831;
 }
\ No newline at end of file
diff --git a/cmd/hz/protobuf/plugin.go b/cmd/hz/protobuf/plugin.go
index 82b01cd..e258038 100644
--- a/cmd/hz/protobuf/plugin.go
+++ b/cmd/hz/protobuf/plugin.go
@@ -76,10 +76,22 @@ type Plugin struct {
 	ModelDir     string
 	UseDir       string
 	IdlClientDir string
+	RmTags       RemoveTags
 	PkgMap       map[string]string
 	logger       *logs.StdLogger
 }
 
+type RemoveTags []string
+
+func (rm *RemoveTags) Exist(tag string) bool {
+	for _, rmTag := range *rm {
+		if rmTag == tag {
+			return true
+		}
+	}
+	return false
+}
+
 func (plugin *Plugin) Run() int {
 	plugin.setLogger()
 	args := &config.Argument{}
@@ -192,6 +204,7 @@ func (plugin *Plugin) Handle(req *pluginpb.CodeGeneratorRequest, args *config.Ar
 	opts := protogen.Options{}
 	gen, err := opts.New(req)
 	plugin.Plugin = gen
+	plugin.RmTags = args.RmTags
 	if err != nil {
 		return fmt.Errorf("new protoc plugin failed: %s", err.Error())
 	}
@@ -322,7 +335,6 @@ func (plugin *Plugin) fixModelPathAndPackage(pkg string) (impt, path string) {
 func (plugin *Plugin) GenerateFiles(pluginPb *protogen.Plugin) error {
 	idl := pluginPb.Request.FileToGenerate[len(pluginPb.Request.FileToGenerate)-1]
 	pluginPb.SupportedFeatures = gengo.SupportedFeatures
-
 	for _, f := range pluginPb.Files {
 		if f.Proto.GetName() == idl {
 			err := plugin.GenerateFile(pluginPb, f)
@@ -358,7 +370,7 @@ func (plugin *Plugin) GenerateFile(gen *protogen.Plugin, f *protogen.File) error
 	if len(plugin.UseDir) != 0 {
 		return nil
 	}
-	file, err := generateFile(gen, f)
+	file, err := generateFile(gen, f, plugin.RmTags)
 	if err != nil || file == nil {
 		return fmt.Errorf("generate file %s failed: %s", f.Proto.GetName(), err.Error())
 	}
@@ -366,7 +378,7 @@ func (plugin *Plugin) GenerateFile(gen *protogen.Plugin, f *protogen.File) error
 }
 
 // generateFile generates the contents of a .pb.go file.
-func generateFile(gen *protogen.Plugin, file *protogen.File) (*protogen.GeneratedFile, error) {
+func generateFile(gen *protogen.Plugin, file *protogen.File, rmTags RemoveTags) (*protogen.GeneratedFile, error) {
 	filename := file.GeneratedFilenamePrefix + ".pb.go"
 	g := gen.NewGeneratedFile(filename, file.GoImportPath)
 	f := newFileInfo(file)
@@ -398,7 +410,7 @@ func generateFile(gen *protogen.Plugin, file *protogen.File) (*protogen.Generate
 	}
 	var err error
 	for _, message := range f.allMessages {
-		err = genMessage(g, f, message)
+		err = genMessage(g, f, message, rmTags)
 		if err != nil {
 			return nil, err
 		}
@@ -410,7 +422,7 @@ func generateFile(gen *protogen.Plugin, file *protogen.File) (*protogen.Generate
 	return g, nil
 }
 
-func genMessage(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo) error {
+func genMessage(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo, rmTags RemoveTags) error {
 	if m.Desc.IsMapEntry() {
 		return nil
 	}
@@ -421,7 +433,7 @@ func genMessage(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo) error {
 		m.Desc.Options().(*descriptorpb.MessageOptions).GetDeprecated())
 	g.P(leadingComments,
 		"type ", m.GoIdent, " struct {")
-	err := genMessageFields(g, f, m)
+	err := genMessageFields(g, f, m, rmTags)
 	if err != nil {
 		return err
 	}
@@ -435,12 +447,12 @@ func genMessage(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo) error {
 	return nil
 }
 
-func genMessageFields(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo) error {
+func genMessageFields(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo, rmTags RemoveTags) error {
 	sf := f.allMessageFieldsByPtr[m]
 	genMessageInternalFields(g, f, m, sf)
 	var err error
 	for _, field := range m.Fields {
-		err = genMessageField(g, f, m, field, sf)
+		err = genMessageField(g, f, m, field, sf, rmTags)
 		if err != nil {
 			return err
 		}
@@ -448,7 +460,7 @@ func genMessageFields(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo) er
 	return nil
 }
 
-func genMessageField(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo, field *protogen.Field, sf *structFields) error {
+func genMessageField(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo, field *protogen.Field, sf *structFields, rmTags RemoveTags) error {
 	if oneof := field.Oneof; oneof != nil && !oneof.Desc.IsSynthetic() {
 		// It would be a bit simpler to iterate over the oneofs below,
 		// but generating the field here keeps the contents of the Go
@@ -506,6 +518,16 @@ func genMessageField(g *protogen.GeneratedFile, f *fileInfo, m *messageInfo, fie
 		tags = append(tags, gotrackTags...)
 	}
 
+	if len(rmTags) > 0 {
+		tmp := structTags{}
+		for _, tag := range tags {
+			if !rmTags.Exist(tag[0]) {
+				tmp = append(tmp, tag)
+			}
+		}
+		tags = tmp
+	}
+
 	name := field.GoName
 	if field.Desc.IsWeak() {
 		name = WeakFieldPrefix_goname + name
diff --git a/cmd/hz/thrift/plugin.go b/cmd/hz/thrift/plugin.go
index 23c5c75..ff4fdf6 100644
--- a/cmd/hz/thrift/plugin.go
+++ b/cmd/hz/thrift/plugin.go
@@ -41,6 +41,7 @@ type Plugin struct {
 	req    *thriftgo_plugin.Request
 	args   *config.Argument
 	logger *logs.StdLogger
+	rmTags []string
 }
 
 func (plugin *Plugin) Run() int {
@@ -74,6 +75,7 @@ func (plugin *Plugin) Run() int {
 		logs.Errorf("parse args failed: %s", err.Error())
 		return meta.PluginError
 	}
+	plugin.rmTags = args.RmTags
 	if args.CmdType == meta.CmdModel {
 		res, err := plugin.GetResponse(nil, args.OutDir)
 		if err != nil {
@@ -334,7 +336,7 @@ func (plugin *Plugin) InsertTag() ([]*thriftgo_plugin.Generated, error) {
 			stName := st.GetName()
 			for _, f := range st.Fields {
 				fieldName := f.GetName()
-				tagString, err := getTagString(f)
+				tagString, err := getTagString(f, plugin.rmTags)
 				if err != nil {
 					return nil, err
 				}
@@ -360,7 +362,7 @@ func (plugin *Plugin) InsertTag() ([]*thriftgo_plugin.Generated, error) {
 			stName := st.GetName()
 			for _, f := range st.Fields {
 				fieldName := f.GetName()
-				tagString, err := getTagString(f)
+				tagString, err := getTagString(f, plugin.rmTags)
 				if err != nil {
 					return nil, err
 				}
@@ -400,7 +402,7 @@ func (plugin *Plugin) GetResponse(files []generator.File, outputDir string) (*th
 	}, nil
 }
 
-func getTagString(f *parser.Field) (string, error) {
+func getTagString(f *parser.Field, rmTags []string) (string, error) {
 	field := model.Field{}
 	err := injectTags(f, &field, true, false)
 	if err != nil {
@@ -412,6 +414,11 @@ func getTagString(f *parser.Field) (string, error) {
 			disableTag = true
 		}
 	}
+
+	for _, rmTag := range rmTags {
+		field.Tags.Remove(rmTag)
+	}
+
 	var tagString string
 	tags := field.Tags
 	for idx, tag := range tags {
diff --git a/pkg/app/context.go b/pkg/app/context.go
index 3ed3f19..e4fcd0a 100644
--- a/pkg/app/context.go
+++ b/pkg/app/context.go
@@ -1294,3 +1294,36 @@ func (ctx *RequestContext) Bind(obj interface{}) error {
 func (ctx *RequestContext) Validate(obj interface{}) error {
 	return binding.Validate(obj)
 }
+
+// VisitAllQueryArgs calls f for each existing query arg.
+//
+// f must not retain references to key and value after returning.
+// Make key and/or value copies if you need storing them after returning.
+func (ctx *RequestContext) VisitAllQueryArgs(f func(key, value []byte)) {
+	ctx.QueryArgs().VisitAll(f)
+}
+
+// VisitAllPostArgs calls f for each existing post arg.
+//
+// f must not retain references to key and value after returning.
+// Make key and/or value copies if you need storing them after returning.
+func (ctx *RequestContext) VisitAllPostArgs(f func(key, value []byte)) {
+	ctx.Request.PostArgs().VisitAll(f)
+}
+
+// VisitAllHeaders calls f for each request header.
+//
+// f must not retain references to key and/or value after returning.
+// Copy key and/or value contents before returning if you need retaining them.
+//
+// To get the headers in order they were received use VisitAllInOrder.
+func (ctx *RequestContext) VisitAllHeaders(f func(key, value []byte)) {
+	ctx.Request.Header.VisitAll(f)
+}
+
+// VisitAllCookie calls f for each request cookie.
+//
+// f must not retain references to key and/or value after returning.
+func (ctx *RequestContext) VisitAllCookie(f func(key, value []byte)) {
+	ctx.Request.Header.VisitAllCookie(f)
+}
diff --git a/pkg/network/standard/connection.go b/pkg/network/standard/connection.go
index 65fd08f..e38ee9e 100644
--- a/pkg/network/standard/connection.go
+++ b/pkg/network/standard/connection.go
@@ -46,6 +46,8 @@ type Conn struct {
 	outputBuffer *linkBuffer
 	caches       [][]byte // buf allocated by Next when cross-package, which should be freed when release
 	maxSize      int      // history max malloc size
+
+	err error
 }
 
 func (c *Conn) ToHertzError(err error) error {
@@ -91,6 +93,8 @@ func (c *Conn) Read(b []byte) (l int, err error) {
 	// If left buffer size is less than block4k, first Peek(1) to fill the buffer.
 	// Then copy min(c.Len, len(b)) to b.
 	if len(b) <= block4k {
+		// If c.fill(1) return err, conn.Read must return 0, err. So there is no need
+		// to check c.Len
 		err = c.fill(1)
 		if err != nil {
 			return 0, err
@@ -293,6 +297,9 @@ func (c *Conn) handleTail() {
 	c.inputBuffer.write.Reset()
 }
 
+// Peek returns the next n bytes without advancing the reader. The bytes stop
+// being valid at the next read call. If Peek returns fewer than n bytes, it
+// also returns an error explaining why the read is short.
 func (c *Conn) Peek(i int) (p []byte, err error) {
 	node := c.inputBuffer.read
 	// fill the inputBuffer so that there is enough data
@@ -301,10 +308,15 @@ func (c *Conn) Peek(i int) (p []byte, err error) {
 		return
 	}
 
+	if c.Len() < i {
+		i = c.Len()
+		err = c.readErr()
+	}
+
 	l := node.Len()
 	// Enough data in a single node, so that just return the slice of the node.
 	if l >= i {
-		return node.buf[node.off : node.off+i], nil
+		return node.buf[node.off : node.off+i], err
 	}
 
 	// not enough data in a signal node
@@ -314,12 +326,12 @@ func (c *Conn) Peek(i int) (p []byte, err error) {
 	} else {
 		p = make([]byte, i)
 	}
-	err = c.peekBuffer(i, p)
+	c.peekBuffer(i, p)
 	return p, err
 }
 
 // peekBuffer loads the buf with data of size i without moving read pointer.
-func (c *Conn) peekBuffer(i int, buf []byte) error {
+func (c *Conn) peekBuffer(i int, buf []byte) {
 	l, pIdx, node := 0, 0, c.inputBuffer.read
 	for ack := i; ack > 0; ack = ack - l {
 		l = node.Len()
@@ -331,7 +343,6 @@ func (c *Conn) peekBuffer(i int, buf []byte) error {
 		}
 		node = node.next
 	}
-	return nil
 }
 
 // next loads the buf with data of size i with moving read pointer.
@@ -345,12 +356,22 @@ func (c *Conn) next(length int, b []byte) error {
 }
 
 // fill loads more data than size i, otherwise it will block read.
+// NOTE: fill may fill data less than i and store err in Conn.err
+// when last read returns n > 0 and err != nil. So after calling
+// fill, it is necessary to check whether c.Len() > i
 func (c *Conn) fill(i int) (err error) {
 	// Check if there is enough data in inputBuffer.
 	if c.Len() >= i {
 		return nil
 	}
-
+	// check whether conn has returned err before.
+	if err = c.readErr(); err != nil {
+		if c.Len() > 0 {
+			c.err = err
+			return nil
+		}
+		return
+	}
 	node := c.inputBuffer.write
 	node.buf = node.buf[:cap(node.buf)]
 	left := cap(node.buf) - node.malloc
@@ -374,15 +395,22 @@ func (c *Conn) fill(i int) (err error) {
 	i -= c.Len()
 	node = c.inputBuffer.write
 	node.buf = node.buf[:cap(node.buf)]
+
 	// Circulate reading data so that the node holds enough data
 	for i > 0 {
 		n, err := c.c.Read(c.inputBuffer.write.buf[node.malloc:])
+		if n > 0 {
+			node.malloc += n
+			c.inputBuffer.len += n
+			i -= n
+			if err != nil {
+				c.err = err
+				return nil
+			}
+		}
 		if err != nil {
 			return err
 		}
-		node.malloc += n
-		c.inputBuffer.len += n
-		i -= n
 	}
 	return nil
 }
@@ -533,6 +561,12 @@ func (c *Conn) HandleSpecificError(err error, rip string) (needIgnore bool) {
 	return false
 }
 
+func (c *Conn) readErr() error {
+	err := c.err
+	c.err = nil
+	return err
+}
+
 func (c *TLSConn) Handshake() error {
 	return c.c.(network.ConnTLSer).Handshake()
 }
diff --git a/pkg/protocol/http1/client.go b/pkg/protocol/http1/client.go
index 353ecc8..13f07a1 100644
--- a/pkg/protocol/http1/client.go
+++ b/pkg/protocol/http1/client.go
@@ -628,8 +628,8 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 	if !c.ResponseBodyStream {
 		err = respI.ReadHeaderAndLimitBody(resp, zr, c.MaxResponseBodySize)
 	} else {
-		err = respI.ReadBodyStream(resp, zr, c.MaxResponseBodySize, func() error {
-			if shouldCloseConn {
+		err = respI.ReadBodyStream(resp, zr, c.MaxResponseBodySize, func(shouldClose bool) error {
+			if shouldCloseConn || shouldClose {
 				c.closeConn(cc)
 			} else {
 				c.releaseConn(cc)
diff --git a/pkg/protocol/http1/ext/stream.go b/pkg/protocol/http1/ext/stream.go
index a09e1e3..ae81b56 100644
--- a/pkg/protocol/http1/ext/stream.go
+++ b/pkg/protocol/http1/ext/stream.go
@@ -309,16 +309,24 @@ func (rs *bodyStream) skipRest() error {
 }
 
 // ReleaseBodyStream releases the body stream.
+// Error of skipRest may be returned if there is one.
 //
 // NOTE: Be careful to use this method unless you know what it's for.
 func ReleaseBodyStream(requestReader io.Reader) (err error) {
 	if rs, ok := requestReader.(*bodyStream); ok {
 		err = rs.skipRest()
-		rs.prefetchedBytes = nil
-		rs.offset = 0
-		rs.reader = nil
-		rs.trailer = nil
+		rs.reset()
 		bodyStreamPool.Put(rs)
 	}
 	return
 }
+
+func (rs *bodyStream) reset() {
+	rs.prefetchedBytes = nil
+	rs.offset = 0
+	rs.reader = nil
+	rs.trailer = nil
+	rs.chunkEOF = false
+	rs.chunkLeft = 0
+	rs.contentLength = 0
+}
diff --git a/pkg/protocol/http1/resp/response.go b/pkg/protocol/http1/resp/response.go
index 819d39d..0366336 100644
--- a/pkg/protocol/http1/resp/response.go
+++ b/pkg/protocol/http1/resp/response.go
@@ -50,6 +50,7 @@ import (
 
 	"github.com/cloudwego/hertz/pkg/common/bytebufferpool"
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
+	"github.com/cloudwego/hertz/pkg/common/hlog"
 	"github.com/cloudwego/hertz/pkg/network"
 	"github.com/cloudwego/hertz/pkg/protocol"
 	"github.com/cloudwego/hertz/pkg/protocol/consts"
@@ -107,8 +108,6 @@ func ReadHeaderAndLimitBody(resp *protocol.Response, r network.Reader, maxBodySi
 		}
 	}
 
-	var cLen int
-
 	if !resp.MustSkipBody() {
 		bodyBuf := resp.BodyBuffer()
 		bodyBuf.Reset()
@@ -116,18 +115,13 @@ func ReadHeaderAndLimitBody(resp *protocol.Response, r network.Reader, maxBodySi
 		if err != nil {
 			return err
 		}
-		cLen = len(bodyBuf.B)
-	}
-
-	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
+			}
 		}
-	}
-
-	if !resp.MustSkipBody() {
-		resp.Header.SetContentLength(cLen)
+		resp.Header.SetContentLength(len(bodyBuf.B))
 	}
 
 	return nil
@@ -135,14 +129,21 @@ func ReadHeaderAndLimitBody(resp *protocol.Response, r network.Reader, maxBodySi
 
 type clientRespStream struct {
 	r             io.Reader
-	closeCallback func() error
+	closeCallback func(shouldClose bool) error
 }
 
 func (c *clientRespStream) Close() (err error) {
 	runtime.SetFinalizer(c, nil)
-	ext.ReleaseBodyStream(c.r)
+	// If error happened in release, the connection may be in abnormal state.
+	// Close it in the callback in order to avoid other unexpected problems.
+	err = ext.ReleaseBodyStream(c.r)
+	shouldClose := false
+	if err != nil {
+		shouldClose = true
+		hlog.Warnf("connection will be closed instead of recycled because an error occurred during the stream body release: %s", err.Error())
+	}
 	if c.closeCallback != nil {
-		err = c.closeCallback()
+		err = c.closeCallback(shouldClose)
 	}
 	c.reset()
 	return
@@ -164,7 +165,7 @@ var clientRespStreamPool = sync.Pool{
 	},
 }
 
-func convertClientRespStream(bs io.Reader, fn func() error) *clientRespStream {
+func convertClientRespStream(bs io.Reader, fn func(shouldClose bool) error) *clientRespStream {
 	clientStream := clientRespStreamPool.Get().(*clientRespStream)
 	clientStream.r = bs
 	clientStream.closeCallback = fn
@@ -173,7 +174,7 @@ func convertClientRespStream(bs io.Reader, fn func() error) *clientRespStream {
 }
 
 // ReadBodyStream reads response body in stream
-func ReadBodyStream(resp *protocol.Response, r network.Reader, maxBodySize int, closeCallBack func() error) error {
+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 {
diff --git a/pkg/protocol/http1/server.go b/pkg/protocol/http1/server.go
index ba8915a..8aaed6d 100644
--- a/pkg/protocol/http1/server.go
+++ b/pkg/protocol/http1/server.go
@@ -350,11 +350,6 @@ func (s Server) Serve(c context.Context, conn network.Conn) (err error) {
 		}
 
 		if hijackHandler != nil {
-			if zr != nil {
-				zr.Release() //nolint:errcheck
-				zr = nil
-			}
-
 			// Hijacked conn process the timeout by itself
 			err = ctx.GetConn().SetReadTimeout(0)
 			if err != nil {
diff --git a/version.go b/version.go
index bf3cf79..f56cbc3 100644
--- a/version.go
+++ b/version.go
@@ -19,5 +19,5 @@ package hertz
 // Name and Version info of this framework, used for statistics and debug
 const (
 	Name    = "Hertz"
-	Version = "v0.6.4"
+	Version = "v0.6.6"
 )


Test patch:
diff --git a/pkg/app/context_test.go b/pkg/app/context_test.go
index 7853d2a..b9e9d2f 100644
--- a/pkg/app/context_test.go
+++ b/pkg/app/context_test.go
@@ -1248,3 +1248,57 @@ func TestRequestContext_SetCookiePathEmpty(t *testing.T) {
 	c.SetCookie("user", "hertz", 1, "", "localhost", protocol.CookieSameSiteDisabled, true, true)
 	assert.DeepEqual(t, "user=hertz; max-age=1; domain=localhost; path=/; HttpOnly; secure", c.Response.Header.Get("Set-Cookie"))
 }
+
+func TestRequestContext_VisitAll(t *testing.T) {
+	t.Run("VisitAllQueryArgs", func(t *testing.T) {
+		c := NewContext(0)
+		var s []string
+		c.QueryArgs().Add("cloudwego", "hertz")
+		c.QueryArgs().Add("hello", "world")
+		c.VisitAllQueryArgs(func(key, value []byte) {
+			s = append(s, string(key), string(value))
+		})
+		assert.DeepEqual(t, []string{"cloudwego", "hertz", "hello", "world"}, s)
+	})
+
+	t.Run("VisitAllPostArgs", func(t *testing.T) {
+		c := NewContext(0)
+		var s []string
+		c.PostArgs().Add("cloudwego", "hertz")
+		c.PostArgs().Add("hello", "world")
+		c.VisitAllPostArgs(func(key, value []byte) {
+			s = append(s, string(key), string(value))
+		})
+		assert.DeepEqual(t, []string{"cloudwego", "hertz", "hello", "world"}, s)
+	})
+
+	t.Run("VisitAllCookie", func(t *testing.T) {
+		c := NewContext(0)
+		var s []string
+		c.Request.Header.Set("Cookie", "aaa=bbb;ccc=ddd")
+		c.VisitAllCookie(func(key, value []byte) {
+			s = append(s, string(key), string(value))
+		})
+		assert.DeepEqual(t, []string{"aaa", "bbb", "ccc", "ddd"}, s)
+	})
+
+	t.Run("VisitAllHeaders", func(t *testing.T) {
+		c := NewContext(0)
+		c.Request.Header.Set("xxx", "yyy")
+		c.Request.Header.Set("xxx2", "yyy2")
+		c.VisitAllHeaders(
+			func(k, v []byte) {
+				key := string(k)
+				value := string(v)
+				if key != "Xxx" && key != "Xxx2" {
+					t.Fatalf("Unexpected %v. Expected %v", key, "xxx or yyy")
+				}
+				if key == "Xxx" && value != "yyy" {
+					t.Fatalf("Unexpected %v. Expected %v", value, "yyy")
+				}
+				if key == "Xxx2" && value != "yyy2" {
+					t.Fatalf("Unexpected %v. Expected %v", value, "yyy2")
+				}
+			})
+	})
+}
diff --git a/pkg/common/test/assert/assert.go b/pkg/common/test/assert/assert.go
index 9ed544f..ed15783 100644
--- a/pkg/common/test/assert/assert.go
+++ b/pkg/common/test/assert/assert.go
@@ -86,10 +86,12 @@ func NotEqual(t testing.TB, expected, actual interface{}) {
 }
 
 func True(t testing.TB, obj interface{}) {
+	t.Helper()
 	DeepEqual(t, true, obj)
 }
 
 func False(t testing.TB, obj interface{}) {
+	t.Helper()
 	DeepEqual(t, false, obj)
 }
 
diff --git a/pkg/common/test/mock/body_data.go b/pkg/common/test/mock/body_data.go
index 808e25e..807c585 100644
--- a/pkg/common/test/mock/body_data.go
+++ b/pkg/common/test/mock/body_data.go
@@ -16,6 +16,8 @@
 
 package mock
 
+import "fmt"
+
 func CreateFixedBody(bodySize int) []byte {
 	var b []byte
 	for i := 0; i < bodySize; i++ {
@@ -23,3 +25,29 @@ func CreateFixedBody(bodySize int) []byte {
 	}
 	return b
 }
+
+func CreateChunkedBody(body []byte, trailer map[string]string, hasTrailer bool) []byte {
+	var b []byte
+	chunkSize := 1
+	for len(body) > 0 {
+		if chunkSize > len(body) {
+			chunkSize = len(body)
+		}
+		b = append(b, []byte(fmt.Sprintf("%x\r\n", chunkSize))...)
+		b = append(b, body[:chunkSize]...)
+		b = append(b, []byte("\r\n")...)
+		body = body[chunkSize:]
+		chunkSize++
+	}
+	if hasTrailer {
+		b = append(b, "0\r\n"...)
+		for k, v := range trailer {
+			b = append(b, k...)
+			b = append(b, ": "...)
+			b = append(b, v...)
+			b = append(b, "\r\n"...)
+		}
+		b = append(b, "\r\n"...)
+	}
+	return b
+}
diff --git a/pkg/common/test/mock/body_data_test.go b/pkg/common/test/mock/body_data_test.go
index a783a47..62ebf1b 100644
--- a/pkg/common/test/mock/body_data_test.go
+++ b/pkg/common/test/mock/body_data_test.go
@@ -18,6 +18,8 @@ package mock
 
 import (
 	"testing"
+
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
 )
 
 func TestGenerateCreateFixedBody(t *testing.T) {
@@ -33,3 +35,13 @@ func TestGenerateCreateFixedBody(t *testing.T) {
 		t.Fatalf("Unexpected %s. Expecting a nil", nilFixedBody)
 	}
 }
+
+func TestGenerateCreateChunkedBody(t *testing.T) {
+	bodySize := 10
+	b := CreateFixedBody(bodySize)
+	trailer := map[string]string{"Foo": "chunked shit"}
+	expectCb := "1\r\n0\r\n2\r\n12\r\n3\r\n345\r\n4\r\n6789\r\n0\r\nFoo: chunked shit\r\n\r\n"
+
+	cb := CreateChunkedBody(b, trailer, true)
+	assert.DeepEqual(t, expectCb, string(cb))
+}
diff --git a/pkg/common/test/mock/network.go b/pkg/common/test/mock/network.go
index 1038eb7..7ad0836 100644
--- a/pkg/common/test/mock/network.go
+++ b/pkg/common/test/mock/network.go
@@ -125,6 +125,10 @@ func (m *Conn) WriterRecorder() Recorder {
 	return &recorder{c: m, Reader: m.zw}
 }
 
+func (m *Conn) GetReadTimeout() time.Duration {
+	return m.readTimeout
+}
+
 type recorder struct {
 	c *Conn
 	network.Reader
@@ -161,6 +165,22 @@ func NewSlowReadConn(source string) *SlowReadConn {
 	return &SlowReadConn{Conn: NewConn(source)}
 }
 
+type ErrorReadConn struct {
+	*Conn
+	errorToReturn error
+}
+
+func NewErrorReadConn(err error) *ErrorReadConn {
+	return &ErrorReadConn{
+		Conn:          NewConn(""),
+		errorToReturn: err,
+	}
+}
+
+func (er *ErrorReadConn) Peek(n int) ([]byte, error) {
+	return nil, er.errorToReturn
+}
+
 type SlowWriteConn struct {
 	*Conn
 	writeTimeout time.Duration
diff --git a/pkg/common/test/mock/network_test.go b/pkg/common/test/mock/network_test.go
index 84df1cd..2d4ff97 100644
--- a/pkg/common/test/mock/network_test.go
+++ b/pkg/common/test/mock/network_test.go
@@ -34,6 +34,7 @@ func TestConn(t *testing.T) {
 		assert.DeepEqual(t, nil, err)
 		err = conn1.SetReadTimeout(time.Millisecond * 100)
 		assert.DeepEqual(t, nil, err)
+		assert.DeepEqual(t, time.Millisecond*100, conn1.GetReadTimeout())
 
 		// Peek Skip Read
 		b, _ := conn1.Peek(1)
diff --git a/pkg/network/standard/connection_test.go b/pkg/network/standard/connection_test.go
index 6fcc456..fa950f3 100644
--- a/pkg/network/standard/connection_test.go
+++ b/pkg/network/standard/connection_test.go
@@ -274,9 +274,10 @@ func TestHandleSpecificError(t *testing.T) {
 }
 
 type mockConn struct {
-	buffer     bytes.Buffer
-	localAddr  net.Addr
-	remoteAddr net.Addr
+	buffer        bytes.Buffer
+	localAddr     net.Addr
+	remoteAddr    net.Addr
+	readReturnErr bool
 }
 
 func (m *mockConn) Handshake() error {
@@ -292,17 +293,21 @@ func (m mockConn) Read(b []byte) (n int, err error) {
 	for i := 0; i < length; i++ {
 		b[i] = 0
 	}
+
+	if m.readReturnErr {
+		err = io.EOF
+	}
 	if length > 8192 {
-		return 8192, nil
+		return 8192, err
 	}
 	if len(b) < 1024 {
-		return 100, nil
+		return 100, err
 	}
 	if len(b) < 5000 {
-		return 4096, nil
+		return 4096, err
 	}
 
-	return 4099, nil
+	return 4099, err
 }
 
 func (m *mockConn) Write(b []byte) (n int, err error) {
@@ -378,3 +383,17 @@ func TestConnSetFinalizer(t *testing.T) {
 
 	assert.DeepEqual(t, uint32(2), atomic.LoadUint32(&release_count))
 }
+
+func TestFillReturnErrAndN(t *testing.T) {
+	c := &mockConn{
+		readReturnErr: true,
+	}
+	conn := newConn(c, 4099)
+	b, err := conn.Peek(4099)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, len(b), 4099)
+	conn.Skip(10)
+	b, err = conn.Peek(4099)
+	assert.DeepEqual(t, err, io.EOF)
+	assert.DeepEqual(t, len(b), 4089)
+}
diff --git a/pkg/protocol/http1/client_test.go b/pkg/protocol/http1/client_test.go
index 1e4e013..1915445 100644
--- a/pkg/protocol/http1/client_test.go
+++ b/pkg/protocol/http1/client_test.go
@@ -55,12 +55,15 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cloudwego/hertz/pkg/app/client/retry"
 	"github.com/cloudwego/hertz/pkg/common/config"
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"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/protocol"
+	"github.com/cloudwego/hertz/pkg/protocol/client"
 	"github.com/cloudwego/hertz/pkg/protocol/consts"
 	"github.com/cloudwego/hertz/pkg/protocol/http1/resp"
 	"github.com/cloudwego/netpoll"
@@ -363,6 +366,103 @@ func TestDialTimeoutPriority(t *testing.T) {
 	}
 }
 
+func TestStateObserve(t *testing.T) {
+	syncState := struct {
+		mu    sync.Mutex
+		state config.ConnPoolState
+	}{}
+	c := &HostClient{
+		ClientOptions: &ClientOptions{
+			Dialer: newSlowConnDialer(func(network, addr string) (network.Conn, error) {
+				return mock.SlowReadDialer(addr)
+			}),
+			StateObserve: func(hcs config.HostClientState) {
+				syncState.mu.Lock()
+				defer syncState.mu.Unlock()
+				syncState.state = hcs.ConnPoolState()
+			},
+			ObservationInterval: 50 * time.Millisecond,
+		},
+		Addr:   "foobar",
+		closed: make(chan struct{}),
+	}
+
+	c.SetDynamicConfig(&client.DynamicConfig{
+		Addr: utils.AddMissingPort(c.Addr, true),
+	})
+
+	time.Sleep(500 * time.Millisecond)
+	assert.Nil(t, c.Close())
+	syncState.mu.Lock()
+	assert.DeepEqual(t, "foobar:443", syncState.state.Addr)
+	syncState.mu.Unlock()
+}
+
+func TestCachedTLSConfig(t *testing.T) {
+	c := &HostClient{
+		ClientOptions: &ClientOptions{
+			Dialer: newSlowConnDialer(func(network, addr string) (network.Conn, error) {
+				return mock.SlowReadDialer(addr)
+			}),
+			TLSConfig: &tls.Config{
+				InsecureSkipVerify: true,
+			},
+		},
+		Addr:  "foobar",
+		IsTLS: true,
+	}
+
+	cfg1 := c.cachedTLSConfig("foobar")
+	cfg2 := c.cachedTLSConfig("baz")
+	assert.NotEqual(t, cfg1, cfg2)
+	cfg3 := c.cachedTLSConfig("foobar")
+	assert.DeepEqual(t, cfg1, cfg3)
+}
+
+func TestRetry(t *testing.T) {
+	var times int32
+	c := &HostClient{
+		ClientOptions: &ClientOptions{
+			Dialer: newSlowConnDialer(func(network, addr string) (network.Conn, error) {
+				times++
+				if times < 3 {
+					return &retryConn{
+						Conn: mock.NewConn(""),
+					}, nil
+				}
+				return mock.NewConn("HTTP/1.1 200 OK\r\nContent-Length: 10\r\nContent-Type: foo/bar\r\n\r\n0123456789"), nil
+			}),
+			RetryConfig: &retry.Config{
+				MaxAttemptTimes: 5,
+				Delay:           time.Millisecond * 10,
+			},
+			RetryIfFunc: func(req *protocol.Request, resp *protocol.Response, err error) bool {
+				return true
+			},
+		},
+		Addr: "foobar",
+	}
+
+	req := protocol.AcquireRequest()
+	req.SetRequestURI("http://foobar/baz")
+	req.SetOptions(config.WithWriteTimeout(time.Millisecond * 100))
+	resp := protocol.AcquireResponse()
+
+	ch := make(chan error, 1)
+	go func() {
+		ch <- c.Do(context.Background(), req, resp)
+	}()
+	select {
+	case <-time.After(time.Second * 2):
+		t.Fatalf("should use writeTimeout in request options")
+	case err := <-ch:
+		assert.Nil(t, err)
+		assert.True(t, times == 3)
+		assert.DeepEqual(t, resp.StatusCode(), 200)
+		assert.DeepEqual(t, resp.Body(), []byte("0123456789"))
+	}
+}
+
 // mockConn for getting error when write binary data.
 type writeErrConn struct {
 	network.Conn
@@ -371,3 +471,11 @@ type writeErrConn struct {
 func (w writeErrConn) WriteBinary(b []byte) (n int, err error) {
 	return 0, errs.ErrConnectionClosed
 }
+
+type retryConn struct {
+	network.Conn
+}
+
+func (w retryConn) SetWriteTimeout(t time.Duration) error {
+	return errors.New("should retry")
+}
diff --git a/pkg/protocol/http1/client_unix_test.go b/pkg/protocol/http1/client_unix_test.go
index ba616ad..6e417ba 100644
--- a/pkg/protocol/http1/client_unix_test.go
+++ b/pkg/protocol/http1/client_unix_test.go
@@ -19,11 +19,15 @@ package http1
 
 import (
 	"context"
+	"errors"
 	"net/http"
 	"runtime"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	errs "github.com/cloudwego/hertz/pkg/common/errors"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"github.com/cloudwego/hertz/pkg/network/netpoll"
 	"github.com/cloudwego/hertz/pkg/protocol"
@@ -63,3 +67,55 @@ func TestGcBodyStream(t *testing.T) {
 	c.CloseIdleConnections()
 	assert.DeepEqual(t, 0, c.ConnPoolState().TotalConnNum)
 }
+
+func TestMaxConn(t *testing.T) {
+	srv := &http.Server{Addr: ":11002", Handler: http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		w.Write([]byte("hello world\n"))
+	})}
+	go srv.ListenAndServe()
+	time.Sleep(100 * time.Millisecond)
+
+	c := &HostClient{
+		ClientOptions: &ClientOptions{
+			Dialer:             netpoll.NewDialer(),
+			ResponseBodyStream: true,
+			MaxConnWaitTimeout: time.Millisecond * 100,
+			MaxConns:           5,
+		},
+		Addr: "127.0.0.1:11002",
+	}
+
+	var successCount int32
+	var noFreeCount int32
+	wg := sync.WaitGroup{}
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			req, resp := protocol.AcquireRequest(), protocol.AcquireResponse()
+			req.SetRequestURI("http://127.0.0.1:11002")
+			req.SetMethod(consts.MethodPost)
+			err := c.Do(context.Background(), req, resp)
+			if err != nil {
+				if errors.Is(err, errs.ErrNoFreeConns) {
+					atomic.AddInt32(&noFreeCount, 1)
+					return
+				}
+				t.Errorf("client Do error=%v", err.Error())
+			}
+			atomic.AddInt32(&successCount, 1)
+		}()
+	}
+	wg.Wait()
+
+	assert.True(t, atomic.LoadInt32(&successCount) == 5)
+	assert.True(t, atomic.LoadInt32(&noFreeCount) == 5)
+	assert.DeepEqual(t, 0, c.ConnectionCount())
+	assert.DeepEqual(t, 5, c.WantConnectionCount())
+
+	runtime.GC()
+	// wait for gc
+	time.Sleep(100 * time.Millisecond)
+	c.CloseIdleConnections()
+	assert.DeepEqual(t, 0, c.WantConnectionCount())
+}
diff --git a/pkg/protocol/http1/ext/common_test.go b/pkg/protocol/http1/ext/common_test.go
index 5ad9c00..824ccf4 100644
--- a/pkg/protocol/http1/ext/common_test.go
+++ b/pkg/protocol/http1/ext/common_test.go
@@ -18,12 +18,16 @@ package ext
 
 import (
 	"bytes"
+	"errors"
+	"io"
 	"strings"
 	"testing"
 
+	errs "github.com/cloudwego/hertz/pkg/common/errors"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"github.com/cloudwego/hertz/pkg/common/test/mock"
 	"github.com/cloudwego/hertz/pkg/protocol"
+	"github.com/cloudwego/netpoll"
 )
 
 func Test_stripSpace(t *testing.T) {
@@ -77,6 +81,12 @@ func TestReadTrailerError(t *testing.T) {
 	if err == nil {
 		t.Fatalf("expecting error.")
 	}
+
+	// eof
+	er := mock.EOFReader{}
+	trailer = protocol.Trailer{}
+	err = ReadTrailer(&trailer, &er)
+	assert.DeepEqual(t, io.EOF, err)
 }
 
 func TestReadTrailer1(t *testing.T) {
@@ -95,3 +105,87 @@ func TestReadTrailer1(t *testing.T) {
 		}
 	}
 }
+
+func TestReadRawHeaders(t *testing.T) {
+	s := "HTTP/1.1 200 OK\r\n" +
+		"EmptyValue1:\r\n" +
+		"Content-Type: foo/bar;\r\n\tnewline;\r\n another/newline\r\n" +
+		"Foo: Bar\r\n" +
+		"Multi-Line: one;\r\n two\r\n" +
+		"Values: v1;\r\n v2; v3;\r\n v4;\tv5\r\n" +
+		"Content-Length: 5\r\n\r\n" +
+		"HELLOaaa"
+
+	var dst []byte
+	rawHeaders, index, err := ReadRawHeaders(dst, []byte(s))
+	assert.Nil(t, err)
+	assert.DeepEqual(t, s[:index], string(rawHeaders))
+}
+
+func TestBodyChunked(t *testing.T) {
+	body := "foobar baz aaa bbb ccc"
+	chunk := "16\r\nfoobar baz aaa bbb ccc\r\n0\r\n"
+	b := bytes.NewBufferString(body)
+
+	var w bytes.Buffer
+	zw := netpoll.NewWriter(&w)
+	WriteBodyChunked(zw, b)
+
+	assert.DeepEqual(t, chunk, w.String())
+
+	zr := mock.NewZeroCopyReader(chunk)
+	rb, err := ReadBody(zr, -1, 0, nil)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, body, string(rb))
+}
+
+func TestBodyFixedSize(t *testing.T) {
+	body := mock.CreateFixedBody(10)
+	b := bytes.NewBuffer(body)
+
+	var w bytes.Buffer
+	zw := netpoll.NewWriter(&w)
+	WriteBodyFixedSize(zw, b, int64(len(body)))
+
+	assert.DeepEqual(t, body, w.Bytes())
+
+	zr := mock.NewZeroCopyReader(string(body))
+	rb, err := ReadBody(zr, len(body), 0, nil)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, body, rb)
+}
+
+func TestBodyIdentity(t *testing.T) {
+	body := mock.CreateFixedBody(1024)
+	zr := mock.NewZeroCopyReader(string(body))
+	rb, err := ReadBody(zr, -2, 0, nil)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, string(body), string(rb))
+}
+
+func TestBodySkipTrailer(t *testing.T) {
+	t.Run("TestBodySkipTrailer", func(t *testing.T) {
+		body := mock.CreateFixedBody(10)
+		trailer := map[string]string{"Foo": "chunked shit"}
+		chunkedBody := mock.CreateChunkedBody(body, trailer, true)
+		r := mock.NewSlowReadConn(string(chunkedBody))
+		err := SkipTrailer(r)
+		assert.Nil(t, err)
+		_, err = r.ReadByte()
+		assert.NotNil(t, err)
+		assert.True(t, errors.Is(err, netpoll.ErrEOF))
+	})
+
+	t.Run("TestBodySkipTrailerError", func(t *testing.T) {
+		//  timeout error
+		sr := mock.NewSlowReadConn("")
+		err := SkipTrailer(sr)
+		assert.NotNil(t, err)
+		assert.True(t, errors.Is(err, errs.ErrTimeout))
+		//  EOF error
+		er := &mock.EOFReader{}
+		err = SkipTrailer(er)
+		assert.NotNil(t, err)
+		assert.True(t, errors.Is(err, io.EOF))
+	})
+}
diff --git a/pkg/protocol/http1/ext/headerscanner_test.go b/pkg/protocol/http1/ext/headerscanner_test.go
index fc1f921..0f8874d 100644
--- a/pkg/protocol/http1/ext/headerscanner_test.go
+++ b/pkg/protocol/http1/ext/headerscanner_test.go
@@ -42,8 +42,13 @@
 package ext
 
 import (
+	"bufio"
+	"errors"
+	"net/http"
+	"strings"
 	"testing"
 
+	errs "github.com/cloudwego/hertz/pkg/common/errors"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 )
 
@@ -52,3 +57,58 @@ func TestHasHeaderValue(t *testing.T) {
 	assert.True(t, HasHeaderValue(s, []byte("Connection: Keep-Alive")))
 	assert.False(t, HasHeaderValue(s, []byte("Connection: Keep-Alive1")))
 }
+
+func TestResponseHeaderMultiLineValue(t *testing.T) {
+	firstLine := "HTTP/1.1 200 OK\r\n"
+	rawHeaders := "EmptyValue1:\r\n" +
+		"Content-Type: foo/bar;\r\n\tnewline;\r\n another/newline\r\n" +
+		"Foo: Bar\r\n" +
+		"Multi-Line: one;\r\n two\r\n" +
+		"Values: v1;\r\n v2; v3;\r\n v4;\tv5\r\n" +
+		"\r\n"
+
+	// compared with http response
+	response, err := http.ReadResponse(bufio.NewReader(strings.NewReader(firstLine+rawHeaders)), nil)
+	assert.Nil(t, err)
+	defer func() { response.Body.Close() }()
+
+	hs := &HeaderScanner{}
+	hs.B = []byte(rawHeaders)
+	hs.DisableNormalizing = false
+	hmap := make(map[string]string, len(response.Header))
+	for hs.Next() {
+		if len(hs.Key) > 0 {
+			hmap[string(hs.Key)] = string(hs.Value)
+		}
+	}
+
+	for name, vals := range response.Header {
+		got := hmap[name]
+		want := vals[0]
+		assert.DeepEqual(t, want, got)
+	}
+}
+
+func TestHeaderScannerError(t *testing.T) {
+	t.Run("TestHeaderScannerErrorInvalidName", func(t *testing.T) {
+		rawHeaders := "Host: go.dev\r\nGopher-New-\r\n Line: This is a header on multiple lines\r\n\r\n"
+		testTestHeaderScannerError(t, rawHeaders, errInvalidName)
+	})
+	t.Run("TestHeaderScannerErrorNeedMore", func(t *testing.T) {
+		rawHeaders := "This is a header on multiple lines"
+		testTestHeaderScannerError(t, rawHeaders, errs.ErrNeedMore)
+
+		rawHeaders = "Gopher-New-\r\n Line"
+		testTestHeaderScannerError(t, rawHeaders, errs.ErrNeedMore)
+	})
+}
+
+func testTestHeaderScannerError(t *testing.T, rawHeaders string, expectError error) {
+	hs := &HeaderScanner{}
+	hs.B = []byte(rawHeaders)
+	hs.DisableNormalizing = false
+	for hs.Next() {
+	}
+	assert.NotNil(t, hs.Err)
+	assert.True(t, errors.Is(hs.Err, expectError))
+}
diff --git a/pkg/protocol/http1/ext/stream_test.go b/pkg/protocol/http1/ext/stream_test.go
index 57cad37..ccec83d 100644
--- a/pkg/protocol/http1/ext/stream_test.go
+++ b/pkg/protocol/http1/ext/stream_test.go
@@ -16,6 +16,8 @@
 package ext
 
 import (
+	"bytes"
+	"errors"
 	"fmt"
 	"io"
 	"testing"
@@ -88,3 +90,164 @@ func TestChunkedSkipRest(t *testing.T) {
 	// big body
 	testChunkedSkipRestWithBodySize(t, 3*1024*1024)
 }
+
+func TestBodyStream_Reset(t *testing.T) {
+	t.Parallel()
+	bs := bodyStream{
+		prefetchedBytes: bytes.NewReader([]byte("aaa")),
+		reader:          mock.NewZeroCopyReader("bbb"),
+		trailer:         &protocol.Trailer{},
+		offset:          10,
+		contentLength:   20,
+		chunkLeft:       50,
+		chunkEOF:        true,
+	}
+
+	bs.reset()
+
+	assert.Nil(t, bs.prefetchedBytes)
+	assert.Nil(t, bs.reader)
+	assert.Nil(t, bs.trailer)
+	assert.DeepEqual(t, 0, bs.offset)
+	assert.DeepEqual(t, 0, bs.contentLength)
+	assert.DeepEqual(t, 0, bs.chunkLeft)
+	assert.False(t, bs.chunkEOF)
+}
+
+func TestReadBodyWithStreaming(t *testing.T) {
+	t.Run("TestBodyFixedSize", func(t *testing.T) {
+		bodySize := 1024
+		body := mock.CreateFixedBody(bodySize)
+		reader := mock.NewZeroCopyReader(string(body))
+		dst, err := ReadBodyWithStreaming(reader, bodySize, -1, nil)
+		assert.Nil(t, err)
+		assert.DeepEqual(t, body, dst)
+	})
+
+	t.Run("TestBodyFixedSizeMaxContentLength", func(t *testing.T) {
+		bodySize := 8 * 1024 * 2
+		body := mock.CreateFixedBody(bodySize)
+		reader := mock.NewZeroCopyReader(string(body))
+		dst, err := ReadBodyWithStreaming(reader, bodySize, 8*1024*10, nil)
+		assert.Nil(t, err)
+		assert.DeepEqual(t, body[:maxContentLengthInStream], dst)
+	})
+
+	t.Run("TestBodyIdentity", func(t *testing.T) {
+		bodySize := 1024
+		body := mock.CreateFixedBody(bodySize)
+		reader := mock.NewZeroCopyReader(string(body))
+		dst, err := ReadBodyWithStreaming(reader, -2, 512, nil)
+		assert.Nil(t, err)
+		assert.DeepEqual(t, body, dst)
+	})
+
+	t.Run("TestErrBodyTooLarge", func(t *testing.T) {
+		bodySize := 2048
+		body := mock.CreateFixedBody(bodySize)
+		reader := mock.NewZeroCopyReader(string(body))
+		dst, err := ReadBodyWithStreaming(reader, bodySize, 1024, nil)
+		assert.True(t, errors.Is(err, errBodyTooLarge))
+		assert.DeepEqual(t, body[:len(dst)], dst)
+	})
+
+	t.Run("TestErrChunkedStream", func(t *testing.T) {
+		bodySize := 1024
+		body := mock.CreateFixedBody(bodySize)
+		reader := mock.NewZeroCopyReader(string(body))
+		dst, err := ReadBodyWithStreaming(reader, -1, bodySize, nil)
+		assert.True(t, errors.Is(err, errChunkedStream))
+		assert.Nil(t, dst)
+	})
+}
+
+func TestBodyStream(t *testing.T) {
+	t.Run("TestBodyStreamPrereadBuffer", func(t *testing.T) {
+		bodySize := 1024
+		body := mock.CreateFixedBody(bodySize)
+		byteBuffer := &bytebufferpool.ByteBuffer{}
+		byteBuffer.Set(body)
+
+		bs := AcquireBodyStream(byteBuffer, mock.NewSlowReadConn(""), nil, len(body))
+		defer func() {
+			ReleaseBodyStream(bs)
+		}()
+
+		b := make([]byte, bodySize)
+		err := bodyStreamRead(bs, b)
+		assert.Nil(t, err)
+		assert.DeepEqual(t, len(body), len(b))
+		assert.DeepEqual(t, string(body), string(b))
+	})
+
+	t.Run("TestBodyStreamRelease", func(t *testing.T) {
+		bodySize := 1024
+		body := mock.CreateFixedBody(bodySize)
+		byteBuffer := &bytebufferpool.ByteBuffer{}
+		byteBuffer.Set(body)
+		bs := AcquireBodyStream(byteBuffer, mock.NewSlowReadConn(string(body)), nil, bodySize*2)
+		err := ReleaseBodyStream(bs)
+		assert.Nil(t, err)
+	})
+
+	t.Run("TestBodyStreamChunked", func(t *testing.T) {
+		bodySize := 5
+		body := mock.CreateFixedBody(bodySize)
+		expectedTrailer := map[string]string{"Foo": "chunked shit"}
+		chunkedBody := mock.CreateChunkedBody(body, expectedTrailer, true)
+
+		byteBuffer := &bytebufferpool.ByteBuffer{}
+		byteBuffer.Set(chunkedBody)
+
+		bs := AcquireBodyStream(byteBuffer, mock.NewSlowReadConn(string(chunkedBody)), &protocol.Trailer{}, -1)
+		defer func() {
+			ReleaseBodyStream(bs)
+		}()
+
+		b := make([]byte, bodySize)
+		err := bodyStreamRead(bs, b)
+		assert.Nil(t, err)
+		assert.DeepEqual(t, len(body), len(b))
+		assert.DeepEqual(t, string(body), string(b))
+	})
+
+	t.Run("TestBodyStreamReadFromWire", func(t *testing.T) {
+		bodySize := 1024
+		body := mock.CreateFixedBody(bodySize)
+		byteBuffer := &bytebufferpool.ByteBuffer{}
+		byteBuffer.Set(body)
+
+		rcBodySize := 128
+		rcBody := mock.CreateFixedBody(rcBodySize)
+		bs := AcquireBodyStream(byteBuffer, mock.NewSlowReadConn(string(rcBody)), nil, -2)
+		defer func() {
+			ReleaseBodyStream(bs)
+		}()
+
+		b := make([]byte, bodySize)
+		err := bodyStreamRead(bs, b)
+		assert.Nil(t, err)
+		assert.DeepEqual(t, len(body), len(b))
+		assert.DeepEqual(t, string(body), string(b))
+	})
+}
+
+func bodyStreamRead(bs io.Reader, b []byte) (err error) {
+	nb := 0
+	for {
+		p := make([]byte, 64)
+		n, rErr := bs.Read(p)
+		if n > 0 {
+			copy(b[nb:], p[:])
+			nb = nb + n
+		}
+
+		if rErr != nil {
+			if rErr != io.EOF {
+				err = rErr
+			}
+			break
+		}
+	}
+	return
+}
diff --git a/pkg/protocol/http1/req/header_test.go b/pkg/protocol/http1/req/header_test.go
index fcf5f41..489e0aa 100644
--- a/pkg/protocol/http1/req/header_test.go
+++ b/pkg/protocol/http1/req/header_test.go
@@ -44,11 +44,13 @@ package req
 import (
 	"bufio"
 	"bytes"
+	"errors"
 	"fmt"
 	"net/http"
 	"strings"
 	"testing"
 
+	errs "github.com/cloudwego/hertz/pkg/common/errors"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"github.com/cloudwego/hertz/pkg/common/test/mock"
 	"github.com/cloudwego/hertz/pkg/protocol"
@@ -412,3 +414,10 @@ func TestRequestHeader_PeekIfExists(t *testing.T) {
 	assert.DeepEqual(t, []byte{}, rh.Peek("exists"))
 	assert.DeepEqual(t, []byte(nil), rh.Peek("non-exists"))
 }
+
+func TestRequestHeaderError(t *testing.T) {
+	er := mock.EOFReader{}
+	rh := protocol.RequestHeader{}
+	err := ReadHeader(&rh, &er)
+	assert.True(t, errors.Is(err, errs.ErrNothingRead))
+}
diff --git a/pkg/protocol/http1/req/request_test.go b/pkg/protocol/http1/req/request_test.go
index e123da4..0411187 100644
--- a/pkg/protocol/http1/req/request_test.go
+++ b/pkg/protocol/http1/req/request_test.go
@@ -44,6 +44,7 @@ package req
 import (
 	"bufio"
 	"bytes"
+	"encoding/base64"
 	"errors"
 	"fmt"
 	"io"
@@ -54,6 +55,7 @@ import (
 	"testing"
 
 	"github.com/cloudwego/hertz/internal/bytesconv"
+	"github.com/cloudwego/hertz/internal/bytestr"
 	"github.com/cloudwego/hertz/pkg/common/bytebufferpool"
 	"github.com/cloudwego/hertz/pkg/common/compress"
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
@@ -451,7 +453,7 @@ func TestRequestWriteRequestURINoHost(t *testing.T) {
 	t.Parallel()
 
 	var req protocol.Request
-	req.Header.SetRequestURI("http://google.com/foo/bar?baz=aaa")
+	req.Header.SetRequestURI("http://user:pass@google.com/foo/bar?baz=aaa")
 	var w bytes.Buffer
 	zw := netpoll.NewWriter(&w)
 	if err := Write(&req, zw); err != nil {
@@ -474,6 +476,16 @@ func TestRequestWriteRequestURINoHost(t *testing.T) {
 	if string(req.Header.RequestURI()) != "/foo/bar?baz=aaa" {
 		t.Fatalf("unexpected requestURI: %q. Expecting %q", req.Header.RequestURI(), "/foo/bar?baz=aaa")
 	}
+	// authorization
+	authorization := req.Header.Get(string(bytestr.StrAuthorization))
+	author, err := base64.StdEncoding.DecodeString(authorization[len(bytestr.StrBasicSpace):])
+	if err != nil {
+		t.Fatalf("expecting error")
+	}
+
+	if string(author) != "user:pass" {
+		t.Fatalf("unexpected Authorization: %q. Expecting %q", authorization, "user:pass")
+	}
 
 	// verify that Write returns error on non-absolute RequestURI
 	req.Reset()
@@ -484,6 +496,38 @@ func TestRequestWriteRequestURINoHost(t *testing.T) {
 	}
 }
 
+func TestRequestWriteMultipartFile(t *testing.T) {
+	t.Parallel()
+
+	var req protocol.Request
+	req.Header.SetHost("foobar.com")
+	req.Header.SetMethod(consts.MethodPost)
+	req.SetFileReader("filea", "filea.txt", bytes.NewReader([]byte("This is filea.")))
+	req.SetMultipartField("fileb", "fileb.txt", "text/plain", bytes.NewReader([]byte("This is fileb.")))
+
+	var w bytes.Buffer
+	zw := netpoll.NewWriter(&w)
+	if err := Write(&req, zw); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	if err := zw.Flush(); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+
+	var req1 protocol.Request
+	zr := mock.NewZeroCopyReader(w.String())
+	if err := Read(&req1, zr); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+
+	filea, err := req1.FormFile("filea")
+	assert.Nil(t, err)
+	assert.DeepEqual(t, "filea.txt", filea.Filename)
+	fileb, err := req1.FormFile("fileb")
+	assert.Nil(t, err)
+	assert.DeepEqual(t, "fileb.txt", fileb.Filename)
+}
+
 func TestSetRequestBodyStreamChunked(t *testing.T) {
 	t.Parallel()
 
@@ -1037,7 +1081,7 @@ func testRequestMultipartFormNotPreParse(t *testing.T, boundary string, formData
 func testReadBodyChunked(t *testing.T, bodySize int) {
 	body := mock.CreateFixedBody(bodySize)
 	expectedTrailer := map[string]string{"Foo": "chunked shit"}
-	chunkedBody := createChunkedBody(body, expectedTrailer, true)
+	chunkedBody := mock.CreateChunkedBody(body, expectedTrailer, true)
 
 	zr := mock.NewZeroCopyReader(string(chunkedBody))
 
@@ -1052,32 +1096,6 @@ func testReadBodyChunked(t *testing.T, bodySize int) {
 	verifyTrailer(t, zr, expectedTrailer)
 }
 
-func createChunkedBody(body []byte, trailer map[string]string, hasTrailer bool) []byte {
-	var b []byte
-	chunkSize := 1
-	for len(body) > 0 {
-		if chunkSize > len(body) {
-			chunkSize = len(body)
-		}
-		b = append(b, []byte(fmt.Sprintf("%x\r\n", chunkSize))...)
-		b = append(b, body[:chunkSize]...)
-		b = append(b, []byte("\r\n")...)
-		body = body[chunkSize:]
-		chunkSize++
-	}
-	if hasTrailer {
-		b = append(b, "0\r\n"...)
-		for k, v := range trailer {
-			b = append(b, k...)
-			b = append(b, ": "...)
-			b = append(b, v...)
-			b = append(b, "\r\n"...)
-		}
-		b = append(b, "\r\n"...)
-	}
-	return b
-}
-
 func testReadBodyFixedSize(t *testing.T, bodySize int) {
 	body := mock.CreateFixedBody(bodySize)
 
diff --git a/pkg/protocol/http1/resp/response_test.go b/pkg/protocol/http1/resp/response_test.go
index 01a38c1..0ff010f 100644
--- a/pkg/protocol/http1/resp/response_test.go
+++ b/pkg/protocol/http1/resp/response_test.go
@@ -57,6 +57,7 @@ import (
 	"github.com/cloudwego/hertz/pkg/common/utils"
 	"github.com/cloudwego/hertz/pkg/protocol"
 	"github.com/cloudwego/hertz/pkg/protocol/consts"
+	"github.com/cloudwego/hertz/pkg/protocol/http1/ext"
 	"github.com/cloudwego/netpoll"
 )
 
@@ -411,15 +412,16 @@ func TestResponseReadLimitBody(t *testing.T) {
 }
 
 func TestResponseReadWithoutBody(t *testing.T) {
-	t.Parallel()
-
 	var resp protocol.Response
 
 	testResponseReadWithoutBody(t, &resp, "HTTP/1.1 304 Not Modified\r\nContent-Type: aa\r\nContent-Encoding: gzip\r\nContent-Length: 1235\r\n\r\n", false,
 		consts.StatusNotModified, 1235, "aa", nil, "gzip", consts.HTTP11)
 
-	testResponseReadWithoutBody(t, &resp, "HTTP/1.1 204 Foo Bar\r\nContent-Type: aab\r\nTrailer: Foo\r\nContent-Encoding: deflate\r\nTransfer-Encoding: chunked\r\n\r\n0\r\nFoo: bar\r\n\r\n", false,
-		consts.StatusNoContent, -1, "aab", map[string]string{"Foo": "bar"}, "deflate", consts.HTTP11)
+	testResponseReadWithoutBody(t, &resp, "HTTP/1.1 200 Foo Bar\r\nContent-Type: aab\r\nTrailer: Foo\r\nContent-Encoding: deflate\r\nTransfer-Encoding: chunked\r\n\r\n0\r\nFoo: bar\r\n\r\nHTTP/1.2", false,
+		consts.StatusOK, 0, "aab", map[string]string{"Foo": "bar"}, "deflate", consts.HTTP11)
+
+	testResponseReadWithoutBody(t, &resp, "HTTP/1.1 204 Foo Bar\r\nContent-Type: aab\r\nTrailer: Foo\r\nContent-Encoding: deflate\r\nTransfer-Encoding: chunked\r\n\r\n0\r\nFoo: bar\r\n\r\nHTTP/1.2", true,
+		consts.StatusNoContent, -1, "aab", nil, "deflate", consts.HTTP11)
 
 	testResponseReadWithoutBody(t, &resp, "HTTP/1.1 123 AAA\r\nContent-Type: xxx\r\nContent-Encoding: gzip\r\nContent-Length: 3434\r\n\r\n", false,
 		123, 3434, "xxx", nil, "gzip", consts.HTTP11)
@@ -524,6 +526,12 @@ func verifyResponseTrailer(t *testing.T, h *protocol.ResponseHeader, expectedTra
 			t.Fatalf("Unexpected trailer %q. Expected %q. Got %q", k, v, got)
 		}
 	}
+
+	h.Trailer().VisitAll(func(key, value []byte) {
+		if v := expectedTrailers[string(key)]; string(value) != v {
+			t.Fatalf("Unexpected trailer %q. Expected %q. Got %q", string(key), v, string(value))
+		}
+	})
 }
 
 func testResponseReadLimitBodyError(t *testing.T, s string, maxBodySize int) {
@@ -607,6 +615,27 @@ func testResponseBodyStreamWithTrailer(t *testing.T, body []byte, disableNormali
 	}
 }
 
+func TestResponseReadBodyStreamBadReader(t *testing.T) {
+	t.Parallel()
+
+	resp := protocol.AcquireResponse()
+
+	errReader := mock.NewErrorReadConn(errors.New("test error"))
+
+	bodyBuf := resp.BodyBuffer()
+	bodyBuf.Reset()
+
+	bodyStream := ext.AcquireBodyStream(bodyBuf, errReader, resp.Header.Trailer(), 100)
+	resp.ConstructBodyStream(bodyBuf, convertClientRespStream(bodyStream, func(shouldClose bool) error {
+		assert.True(t, shouldClose)
+		return nil
+	}))
+
+	stBody := resp.BodyStream()
+	closer, _ := stBody.(io.Closer)
+	closer.Close()
+}
+
 func TestSetResponseBodyStreamFixedSize(t *testing.T) {
 	t.Parallel()
 
@@ -791,3 +820,9 @@ func TestResponseReadBodyStreamBadTrailer(t *testing.T) {
 	testResponseReadBodyStreamBadTrailer(t, resp, "HTTP/1.1 300 OK\r\nTransfer-Encoding: chunked\r\nContent-Type: bar\r\n\r\n5\r\n56789\r\n0\r\ncontent-type: bar\r\n\r\n")
 	testResponseReadBodyStreamBadTrailer(t, resp, "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nTransfer-Encoding: chunked\r\n\r\n4\r\nqwer\r\n2\r\nty\r\n0\r\nproxy-connection: bar2\r\n\r\n")
 }
+
+func TestResponseString(t *testing.T) {
+	resp := protocol.Response{}
+	resp.Header.Set("Location", "foo\r\nSet-Cookie: SESSIONID=MaliciousValue\r\n")
+	assert.True(t, strings.Contains(GetHTTP1Response(&resp).String(), "Location: foo\r\nSet-Cookie: SESSIONID=MaliciousValue\r\n"))
+}
diff --git a/pkg/protocol/http1/server_test.go b/pkg/protocol/http1/server_test.go
index dfa8b78..e34bb3f 100644
--- a/pkg/protocol/http1/server_test.go
+++ b/pkg/protocol/http1/server_test.go
@@ -20,8 +20,10 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	inStats "github.com/cloudwego/hertz/internal/stats"
 	"github.com/cloudwego/hertz/pkg/app"
@@ -33,6 +35,7 @@ import (
 	"github.com/cloudwego/hertz/pkg/common/tracer/traceinfo"
 	"github.com/cloudwego/hertz/pkg/network"
 	"github.com/cloudwego/hertz/pkg/protocol"
+	"github.com/cloudwego/hertz/pkg/protocol/consts"
 	"github.com/cloudwego/hertz/pkg/protocol/http1/resp"
 )
 
@@ -247,14 +250,124 @@ func TestHijackResponseWriter(t *testing.T) {
 	assert.True(t, isFinal)
 }
 
+func TestHijackHandler(t *testing.T) {
+	server := NewServer()
+	reqCtx := &app.RequestContext{}
+	originReadTimeout := time.Second
+	hijackReadTimeout := 200 * time.Millisecond
+	reqCtx.SetHijackHandler(func(c network.Conn) {
+		c.SetReadTimeout(hijackReadTimeout) // hijack read timeout
+	})
+
+	server.Core = &mockCore{
+		ctxPool: &sync.Pool{New: func() interface{} {
+			return reqCtx
+		}},
+	}
+
+	server.HijackConnHandle = func(c network.Conn, h app.HijackHandler) {
+		h(c)
+	}
+
+	defaultConn := mock.NewConn("GET / HTTP/1.1\nHost: foobar.com\n\n")
+	defaultConn.SetReadTimeout(originReadTimeout)
+	assert.DeepEqual(t, originReadTimeout, defaultConn.GetReadTimeout())
+	err := server.Serve(context.TODO(), defaultConn)
+	assert.True(t, errors.Is(err, errs.ErrHijacked))
+	assert.DeepEqual(t, hijackReadTimeout, defaultConn.GetReadTimeout())
+}
+
+func TestKeepAlive(t *testing.T) {
+	server := NewServer()
+	reqCtx := &app.RequestContext{}
+	times := 0
+	server.Core = &mockCore{
+		ctxPool: &sync.Pool{New: func() interface{} {
+			return reqCtx
+		}},
+		isRunning: true,
+		mockHandler: func(c context.Context, ctx *app.RequestContext) {
+			times++
+			if string(ctx.Path()) == "/close" {
+				ctx.SetConnectionClose()
+			}
+		},
+	}
+	server.IdleTimeout = time.Second
+
+	var s strings.Builder
+	s.WriteString("GET / HTTP/1.1\r\nHost: aaa\r\nConnection: keep-alive\r\n\r\n")
+	s.WriteString("GET /close HTTP/1.0\r\nHost: aaa\r\nConnection: keep-alive\r\n\r\n") // set connection close
+
+	defaultConn := mock.NewConn(s.String())
+	err := server.Serve(context.TODO(), defaultConn)
+	assert.True(t, errors.Is(err, errs.ErrShortConnection))
+	assert.DeepEqual(t, times, 2)
+}
+
+func TestExpect100Continue(t *testing.T) {
+	server := &Server{}
+	reqCtx := &app.RequestContext{}
+	server.Core = &mockCore{
+		ctxPool: &sync.Pool{New: func() interface{} {
+			return reqCtx
+		}},
+		mockHandler: func(c context.Context, ctx *app.RequestContext) {
+			data, err := ctx.Body()
+			if err == nil {
+				ctx.Write(data)
+			}
+		},
+	}
+
+	defaultConn := mock.NewConn("POST /foo HTTP/1.1\r\nHost: gle.com\r\nExpect: 100-continue\r\nContent-Length: 5\r\nContent-Type: a/b\r\n\r\n12345")
+	err := server.Serve(context.TODO(), defaultConn)
+	assert.True(t, errors.Is(err, errs.ErrShortConnection))
+	defaultResponseResult := defaultConn.WriterRecorder()
+	assert.DeepEqual(t, 0, defaultResponseResult.Len())
+	response := protocol.AcquireResponse()
+	resp.Read(response, defaultResponseResult)
+	assert.DeepEqual(t, "12345", string(response.Body()))
+}
+
+func TestExpect100ContinueHandler(t *testing.T) {
+	server := &Server{}
+	reqCtx := &app.RequestContext{}
+	server.Core = &mockCore{
+		ctxPool: &sync.Pool{New: func() interface{} {
+			return reqCtx
+		}},
+		mockHandler: func(c context.Context, ctx *app.RequestContext) {
+			data, err := ctx.Body()
+			if err == nil {
+				ctx.Write(data)
+			}
+		},
+	}
+	server.ContinueHandler = func(header *protocol.RequestHeader) bool {
+		return false
+	}
+
+	defaultConn := mock.NewConn("POST /foo HTTP/1.1\r\nHost: gle.com\r\nExpect: 100-continue\r\nContent-Length: 5\r\nContent-Type: a/b\r\n\r\n12345")
+	err := server.Serve(context.TODO(), defaultConn)
+	assert.True(t, errors.Is(err, errs.ErrShortConnection))
+	defaultResponseResult := defaultConn.WriterRecorder()
+	assert.DeepEqual(t, 0, defaultResponseResult.Len())
+	response := protocol.AcquireResponse()
+	resp.Read(response, defaultResponseResult)
+	assert.DeepEqual(t, consts.StatusExpectationFailed, response.StatusCode())
+	assert.DeepEqual(t, "", string(response.Body()))
+}
+
 type mockCore struct {
 	ctxPool     *sync.Pool
 	controller  tracer.Controller
 	mockHandler func(c context.Context, ctx *app.RequestContext)
+	isRunning   bool
 }
 
 func (m *mockCore) IsRunning() bool {
-	return false
+	return m.isRunning
 }
 
 func (m *mockCore) GetCtxPool() *sync.Pool {