Repository: cloudwego/hertz
Pull request #690 (2023-03-27 08:57:01)

chore: release v0.6.1


Patch:
diff --git a/cmd/hz/generator/layout.go b/cmd/hz/generator/layout.go
index 17bf4a6..3791955 100644
--- a/cmd/hz/generator/layout.go
+++ b/cmd/hz/generator/layout.go
@@ -138,7 +138,7 @@ func (lg *LayoutGenerator) GenerateByService(service Layout) error {
 		return err
 	}
 	if service.HasIdl {
-		for k, _ := range lg.tpls {
+		for k := range lg.tpls {
 			if strings.Contains(k, registerTplName) {
 				delete(lg.tpls, k)
 				break
diff --git a/cmd/hz/protobuf/tags.go b/cmd/hz/protobuf/tags.go
index ac23668..30ba1b8 100644
--- a/cmd/hz/protobuf/tags.go
+++ b/cmd/hz/protobuf/tags.go
@@ -270,10 +270,10 @@ func reflectJsonTag(f protoreflect.FieldDescriptor) (ret model.Tag) {
 	if v := checkFirstOption(api.E_Body, f.Options()); v != nil {
 		ret.Value += ",string"
 	}
-	if !unsetOmitempty && descriptorpb.FieldDescriptorProto_Label(f.Cardinality()) == descriptorpb.FieldDescriptorProto_LABEL_OPTIONAL {
-		ret.Value += ",omitempty"
-	} else if descriptorpb.FieldDescriptorProto_Label(f.Cardinality()) == descriptorpb.FieldDescriptorProto_LABEL_REQUIRED {
+	if descriptorpb.FieldDescriptorProto_Label(f.Cardinality()) == descriptorpb.FieldDescriptorProto_LABEL_REQUIRED {
 		ret.Value += ",required"
+	} else if !unsetOmitempty {
+		ret.Value += ",omitempty"
 	}
 	return
 }
@@ -383,10 +383,11 @@ func getStructJsonValue(f protoreflect.FieldDescriptor, val string) string {
 	if v := checkFirstOption(api.E_JsConv, f.Options()); v != nil {
 		val += ",string"
 	}
-	if !unsetOmitempty && descriptorpb.FieldDescriptorProto_Label(f.Cardinality()) == descriptorpb.FieldDescriptorProto_LABEL_OPTIONAL {
-		val += ",omitempty"
-	} else if descriptorpb.FieldDescriptorProto_Label(f.Cardinality()) == descriptorpb.FieldDescriptorProto_LABEL_REQUIRED {
+
+	if descriptorpb.FieldDescriptorProto_Label(f.Cardinality()) == descriptorpb.FieldDescriptorProto_LABEL_REQUIRED {
 		val += ",required"
+	} else if !unsetOmitempty {
+		val += ",omitempty"
 	}
 
 	return val
diff --git a/cmd/hz/thrift/ast.go b/cmd/hz/thrift/ast.go
index d1580e9..9622ed8 100644
--- a/cmd/hz/thrift/ast.go
+++ b/cmd/hz/thrift/ast.go
@@ -168,7 +168,7 @@ func parseAnnotationToClient(clientMethod *generator.ClientMethod, p *parser.Typ
 	}
 	scope, err := golang.BuildScope(thriftgoUtil, symbol.Scope)
 	if err != nil {
-		fmt.Errorf("can not build scope for %s", p.Name)
+		return fmt.Errorf("can not build scope for %s", p.Name)
 	}
 	thriftgoUtil.SetRootScope(scope)
 	st := scope.StructLike(typeName)
diff --git a/pkg/app/client/client.go b/pkg/app/client/client.go
index 1f98cbe..33f9fb5 100644
--- a/pkg/app/client/client.go
+++ b/pkg/app/client/client.go
@@ -66,7 +66,10 @@ import (
 	"github.com/cloudwego/hertz/pkg/protocol/suite"
 )
 
-var errorInvalidURI = errors.NewPublic("invalid uri")
+var (
+	errorInvalidURI          = errors.NewPublic("invalid uri")
+	errorLastMiddlewareExist = errors.NewPublic("last middleware already set")
+)
 
 // Do performs the given http request and fills the given http response.
 //
@@ -269,10 +272,11 @@ type Client struct {
 
 	clientFactory suite.ClientFactory
 
-	mLock sync.Mutex
-	m     map[string]client.HostClient
-	ms    map[string]client.HostClient
-	mws   Middleware
+	mLock          sync.Mutex
+	m              map[string]client.HostClient
+	ms             map[string]client.HostClient
+	mws            Middleware
+	lastMiddleware Middleware
 }
 
 func (c *Client) GetOptions() *config.ClientOptions {
@@ -454,6 +458,9 @@ func (c *Client) Do(ctx context.Context, req *protocol.Request, resp *protocol.R
 	if c.mws == nil {
 		return c.do(ctx, req, resp)
 	}
+	if c.lastMiddleware != nil {
+		return c.mws(c.lastMiddleware(c.do))(ctx, req, resp)
+	}
 	return c.mws(c.do)(ctx, req, resp)
 }
 
@@ -616,6 +623,28 @@ func (c *Client) Use(mws ...Middleware) {
 	c.mws = chain(middlewares...)
 }
 
+// UseAsLast is used to add middleware to the end of the middleware chain.
+//
+// Will return an error if last middleware has been set before, to ensure all middleware has the change to work,
+// Please use `TakeOutLastMiddleware` to take out the already set middleware.
+// Chain the middleware after or before is both Okay - but remember to put it back.
+func (c *Client) UseAsLast(mw Middleware) error {
+	if c.lastMiddleware != nil {
+		return errorLastMiddlewareExist
+	}
+	c.lastMiddleware = mw
+	return nil
+}
+
+// TakeOutLastMiddleware will return the set middleware and remove it from client.
+//
+// Remember to set it back after chain it with other middleware.
+func (c *Client) TakeOutLastMiddleware() Middleware {
+	last := c.lastMiddleware
+	c.lastMiddleware = nil
+	return last
+}
+
 func newHttp1OptionFromClient(c *Client) *http1.ClientOptions {
 	return &http1.ClientOptions{
 		Name:                          c.options.Name,
diff --git a/pkg/common/config/request_option.go b/pkg/common/config/request_option.go
index 3737ac5..ed6dfc5 100644
--- a/pkg/common/config/request_option.go
+++ b/pkg/common/config/request_option.go
@@ -27,6 +27,10 @@ type RequestOptions struct {
 	dialTimeout  time.Duration
 	readTimeout  time.Duration
 	writeTimeout time.Duration
+	// Request timeout. Usually set by DoDeadline or DoTimeout
+	// if <= 0, means not set
+	requestTimeout time.Duration
+	start          time.Time
 }
 
 // RequestOption is the only struct to set request-level options.
@@ -95,6 +99,16 @@ func WithWriteTimeout(t time.Duration) RequestOption {
 	}}
 }
 
+// WithRequestTimeout sets whole request timeout. If it reaches timeout,
+// the client will return.
+//
+// This is the request level configuration.
+func WithRequestTimeout(t time.Duration) RequestOption {
+	return RequestOption{F: func(o *RequestOptions) {
+		o.requestTimeout = t
+	}}
+}
+
 func (o *RequestOptions) Apply(opts []RequestOption) {
 	for _, op := range opts {
 		op.F(o)
@@ -125,6 +139,23 @@ func (o *RequestOptions) WriteTimeout() time.Duration {
 	return o.writeTimeout
 }
 
+func (o *RequestOptions) RequestTimeout() time.Duration {
+	return o.requestTimeout
+}
+
+// StartRequest records the start time of the request.
+//
+// Note: Users should not call this method.
+func (o *RequestOptions) StartRequest() {
+	if o.requestTimeout > 0 {
+		o.start = time.Now()
+	}
+}
+
+func (o *RequestOptions) StartTime() time.Time {
+	return o.start
+}
+
 func (o *RequestOptions) CopyTo(dst *RequestOptions) {
 	if dst.tags == nil {
 		dst.tags = make(map[string]string)
@@ -138,6 +169,8 @@ func (o *RequestOptions) CopyTo(dst *RequestOptions) {
 	dst.readTimeout = o.readTimeout
 	dst.writeTimeout = o.writeTimeout
 	dst.dialTimeout = o.dialTimeout
+	dst.requestTimeout = o.requestTimeout
+	dst.start = o.start
 }
 
 // SetPreDefinedOpts Pre define some RequestOption here
diff --git a/pkg/network/netpoll/dial.go b/pkg/network/netpoll/dial.go
index 86a371e..ee1feb0 100644
--- a/pkg/network/netpoll/dial.go
+++ b/pkg/network/netpoll/dial.go
@@ -45,6 +45,9 @@ func (d dialer) DialConnection(n, address string, timeout time.Duration, tlsConf
 }
 
 func (d dialer) DialTimeout(network, address string, timeout time.Duration, tlsConfig *tls.Config) (conn net.Conn, err error) {
+	if tlsConfig != nil {
+		return nil, errNotSupportTLS
+	}
 	conn, err = d.Dialer.DialTimeout(network, address, timeout)
 	return
 }
diff --git a/pkg/network/standard/connection.go b/pkg/network/standard/connection.go
index 5388c38..5ab9a8c 100644
--- a/pkg/network/standard/connection.go
+++ b/pkg/network/standard/connection.go
@@ -26,6 +26,7 @@ import (
 	"time"
 
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
+	"github.com/cloudwego/hertz/pkg/common/hlog"
 	"github.com/cloudwego/hertz/pkg/network"
 )
 
@@ -526,6 +527,14 @@ func (c *Conn) Flush() (err error) {
 	return nil
 }
 
+func (c *Conn) HandleSpecificError(err error, rip string) (needIgnore bool) {
+	if errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET) {
+		hlog.SystemLogger().Debugf("Go net library error=%s, remoteAddr=%s", err.Error(), rip)
+		return true
+	}
+	return false
+}
+
 func (c *TLSConn) Handshake() error {
 	return c.c.(network.ConnTLSer).Handshake()
 }
diff --git a/pkg/protocol/client/client.go b/pkg/protocol/client/client.go
index e770547..3715255 100644
--- a/pkg/protocol/client/client.go
+++ b/pkg/protocol/client/client.go
@@ -63,8 +63,6 @@ var (
 	errTooManyRedirects = errors.NewPublic("too many redirects detected when doing the request")
 
 	clientURLResponseChPool sync.Pool
-
-	errorChPool sync.Pool
 )
 
 type HostClient interface {
@@ -279,76 +277,20 @@ func getRedirectURL(baseURL string, location []byte) string {
 }
 
 func DoTimeout(ctx context.Context, req *protocol.Request, resp *protocol.Response, timeout time.Duration, c Doer) error {
-	deadline := time.Now().Add(timeout)
-	return DoDeadline(ctx, req, resp, deadline, c)
+	if timeout <= 0 {
+		return errTimeout
+	}
+	// Note: it will overwrite the reqTimeout.
+	req.SetOptions(config.WithRequestTimeout(timeout))
+	return c.Do(ctx, req, resp)
 }
 
 func DoDeadline(ctx context.Context, req *protocol.Request, resp *protocol.Response, deadline time.Time, c Doer) error {
-	timeout := -time.Since(deadline)
+	timeout := time.Until(deadline)
 	if timeout <= 0 {
 		return errTimeout
 	}
-
-	var ch chan error
-	chv := errorChPool.Get()
-	if chv == nil {
-		chv = make(chan error, 1)
-	}
-	ch = chv.(chan error)
-
-	// Make req and resp copies, since on timeout they no longer
-	// may be accessed.
-	reqCopy := protocol.AcquireRequest()
-	req.CopyToSkipBody(reqCopy)
-	protocol.SwapRequestBody(req, reqCopy)
-	respCopy := protocol.AcquireResponse()
-	if resp != nil {
-		// Not calling resp.copyToSkipBody(respCopy) here to avoid
-		// unexpected messing with headers
-		respCopy.SkipBody = resp.SkipBody
-	}
-
-	// Note that the request continues execution on errTimeout until
-	// client-specific ReadTimeout exceeds. This helps limiting load
-	// on slow hosts by MaxConns* concurrent requests.
-	//
-	// Without this 'hack' the load on slow host could exceed MaxConns*
-	// concurrent requests, since timed out requests on client side
-	// usually continue execution on the host.
-
-	var mu sync.Mutex
-	var timedout bool
-
-	go func() {
-		errDo := c.Do(ctx, reqCopy, respCopy)
-		mu.Lock()
-		if !timedout {
-			if resp != nil {
-				respCopy.CopyToSkipBody(resp)
-				protocol.SwapResponseBody(resp, respCopy)
-			}
-			protocol.SwapRequestBody(reqCopy, req)
-			ch <- errDo
-		}
-		mu.Unlock()
-
-		protocol.ReleaseResponse(respCopy)
-		protocol.ReleaseRequest(reqCopy)
-	}()
-
-	tc := timer.AcquireTimer(timeout)
-	var err error
-	select {
-	case err = <-ch:
-	case <-tc.C:
-		mu.Lock()
-		timedout = true
-		err = errTimeout
-		mu.Unlock()
-	}
-	timer.ReleaseTimer(tc)
-
-	errorChPool.Put(chv)
-
-	return err
+	// Note: it will overwrite the reqTimeout.
+	req.SetOptions(config.WithRequestTimeout(timeout))
+	return c.Do(ctx, req, resp)
 }
diff --git a/pkg/protocol/http1/client.go b/pkg/protocol/http1/client.go
index 4caddb0..8a35aed 100644
--- a/pkg/protocol/http1/client.go
+++ b/pkg/protocol/http1/client.go
@@ -73,6 +73,7 @@ import (
 
 var errConnectionClosed = errs.NewPublic("the server closed connection before returning the first response byte. " +
 	"Make sure the server returns 'Connection: close' response header before closing the connection")
+var errTimeout = errs.New(errs.ErrTimeout, errs.ErrorTypePublic, "host client")
 
 // HostClient balances http requests among hosts listed in Addr.
 //
@@ -377,7 +378,7 @@ func (c *HostClient) Do(ctx context.Context, req *protocol.Request, resp *protoc
 	}
 
 	atomic.AddInt32(&c.pendingRequests, 1)
-
+	req.Options().StartRequest()
 	for {
 		canIdempotentRetry, err = c.do(req, resp)
 		if err == nil {
@@ -465,6 +466,26 @@ func (c *HostClient) preHandleConfig(o *config.RequestOptions) requestConfig {
 	return rc
 }
 
+func updateReqTimeout(reqTimeout, compareTimeout time.Duration, before time.Time) (shouldCloseConn bool, timeout time.Duration) {
+	if reqTimeout <= 0 {
+		return false, compareTimeout
+	}
+	left := reqTimeout - time.Since(before)
+	if left <= 0 {
+		return true, 0
+	}
+
+	if compareTimeout <= 0 {
+		return false, left
+	}
+
+	if left > compareTimeout {
+		return false, compareTimeout
+	}
+
+	return false, left
+}
+
 func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Response) (bool, error) {
 	if req == nil {
 		panic("BUG: req cannot be nil")
@@ -487,7 +508,14 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 	if c.DisablePathNormalizing {
 		req.URI().DisablePathNormalizing = true
 	}
-	cc, err := c.acquireConn(rc.dialTimeout)
+	reqTimeout := req.Options().RequestTimeout()
+	begin := req.Options().StartTime()
+
+	dialTimeout := rc.dialTimeout
+	if reqTimeout < dialTimeout || dialTimeout == 0 {
+		dialTimeout = reqTimeout
+	}
+	cc, err := c.acquireConn(dialTimeout)
 	// if getting connection error, fast fail
 	if err != nil {
 		return false, err
@@ -495,7 +523,6 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 	conn := cc.c
 
 	usingProxy := false
-
 	if c.ProxyURI != nil && bytes.Equal(req.Scheme(), bytestr.StrHTTP) {
 		usingProxy = true
 		proxy.SetProxyAuthHeader(&req.Header, c.ProxyURI)
@@ -503,12 +530,16 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 
 	resp.ParseNetAddr(conn)
 
-	if rc.writeTimeout > 0 {
-		if err = conn.SetWriteTimeout(rc.writeTimeout); err != nil {
-			c.closeConn(cc)
-			// try another connection if retry is enabled
-			return true, err
-		}
+	shouldClose, timeout := updateReqTimeout(reqTimeout, rc.writeTimeout, begin)
+	if shouldClose {
+		c.closeConn(cc)
+		return false, errTimeout
+	}
+
+	if err = conn.SetWriteTimeout(timeout); err != nil {
+		c.closeConn(cc)
+		// try another connection if retry is enabled
+		return true, err
 	}
 
 	resetConnection := false
@@ -566,14 +597,18 @@ func (c *HostClient) doNonNilReqResp(req *protocol.Request, resp *protocol.Respo
 		return true, err
 	}
 
-	if rc.readTimeout > 0 {
-		// Set Deadline every time, since golang has fixed the performance issue
-		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
-		if err = conn.SetReadTimeout(rc.readTimeout); err != nil {
-			c.closeConn(cc)
-			// try another connection if retry is enabled
-			return true, err
-		}
+	shouldClose, timeout = updateReqTimeout(reqTimeout, rc.readTimeout, begin)
+	if shouldClose {
+		c.closeConn(cc)
+		return false, errTimeout
+	}
+
+	// Set Deadline every time, since golang has fixed the performance issue
+	// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
+	if err = conn.SetReadTimeout(timeout); err != nil {
+		c.closeConn(cc)
+		// try another connection if retry is enabled
+		return true, err
 	}
 
 	if customSkipBody || req.Header.IsHead() || req.Header.IsConnect() {
diff --git a/pkg/protocol/request.go b/pkg/protocol/request.go
index f6c138a..d10edd2 100644
--- a/pkg/protocol/request.go
+++ b/pkg/protocol/request.go
@@ -383,6 +383,7 @@ func (req *Request) CopyToSkipBody(dst *Request) {
 		dst.options = &config.RequestOptions{}
 		req.options.CopyTo(dst.options)
 	}
+
 	// do not copy multipartForm - it will be automatically
 	// re-created on the first call to MultipartForm.
 }
diff --git a/version.go b/version.go
index 16b3997..b0b607b 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.0"
+	Version = "v0.6.1"
 )


Test patch:
diff --git a/pkg/app/client/client_test.go b/pkg/app/client/client_test.go
index 1754044..b6eb263 100644
--- a/pkg/app/client/client_test.go
+++ b/pkg/app/client/client_test.go
@@ -55,6 +55,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"reflect"
 	"regexp"
 	"strings"
 	"sync"
@@ -1803,13 +1804,82 @@ func TestClientMiddleware(t *testing.T) {
 	client.Use(mw1)
 	client.Use(mw2)
 
-	req, resp := protocol.AcquireRequest(), protocol.AcquireResponse()
-	err := client.Do(context.Background(), req, resp)
+	request, response := protocol.AcquireRequest(), protocol.AcquireResponse()
+	defer func() {
+		protocol.ReleaseRequest(request)
+		protocol.ReleaseResponse(response)
+	}()
+	err := client.Do(context.Background(), request, response)
 	if err != nil {
 		t.Errorf("unexpected error: %s", err.Error())
 	}
 }
 
+func TestClientLastMiddleware(t *testing.T) {
+	client, _ := NewClient()
+	mw0 := func(next Endpoint) Endpoint {
+		return func(ctx context.Context, req *protocol.Request, resp *protocol.Response) (err error) {
+			finalValue0 := ctx.Value("final0")
+			assert.DeepEqual(t, "final3", finalValue0)
+			finalValue1 := ctx.Value("final1")
+			assert.DeepEqual(t, "final1", finalValue1)
+			finalValue2 := ctx.Value("final2")
+			assert.DeepEqual(t, "final2", finalValue2)
+			return nil
+		}
+	}
+	mw1 := func(next Endpoint) Endpoint {
+		return func(ctx context.Context, req *protocol.Request, resp *protocol.Response) (err error) {
+			ctx = context.WithValue(ctx, "final0", "final0")
+			return next(ctx, req, resp)
+		}
+	}
+	mw2 := func(next Endpoint) Endpoint {
+		return func(ctx context.Context, req *protocol.Request, resp *protocol.Response) (err error) {
+			ctx = context.WithValue(ctx, "final1", "final1")
+			return next(ctx, req, resp)
+		}
+	}
+	mw3 := func(next Endpoint) Endpoint {
+		return func(ctx context.Context, req *protocol.Request, resp *protocol.Response) (err error) {
+			ctx = context.WithValue(ctx, "final2", "final2")
+			return next(ctx, req, resp)
+		}
+	}
+	mw4 := func(next Endpoint) Endpoint {
+		return func(ctx context.Context, req *protocol.Request, resp *protocol.Response) (err error) {
+			ctx = context.WithValue(ctx, "final0", "final3")
+			return next(ctx, req, resp)
+		}
+	}
+	err := client.UseAsLast(mw0)
+	assert.Nil(t, err)
+	err = client.UseAsLast(func(endpoint Endpoint) Endpoint {
+		return nil
+	})
+	assert.DeepEqual(t, errorLastMiddlewareExist, err)
+	client.Use(mw1)
+	client.Use(mw2)
+	client.Use(mw3)
+	client.Use(mw4)
+
+	request, response := protocol.AcquireRequest(), protocol.AcquireResponse()
+	defer func() {
+		protocol.ReleaseRequest(request)
+		protocol.ReleaseResponse(response)
+	}()
+	err = client.Do(context.Background(), request, response)
+	if err != nil {
+		t.Errorf("unexpected error: %s", err.Error())
+	}
+
+	last := client.TakeOutLastMiddleware()
+
+	assert.DeepEqual(t, reflect.ValueOf(last).Pointer(), reflect.ValueOf(mw0).Pointer())
+	last = client.TakeOutLastMiddleware()
+	assert.Nil(t, last)
+}
+
 func TestClientReadResponseBodyStreamWithDoubleRequest(t *testing.T) {
 	part1 := ""
 	for i := 0; i < 8192; i++ {
diff --git a/pkg/common/test/assert/assert.go b/pkg/common/test/assert/assert.go
index 4d52201..9ed544f 100644
--- a/pkg/common/test/assert/assert.go
+++ b/pkg/common/test/assert/assert.go
@@ -51,6 +51,7 @@ func DeepEqual(t testing.TB, expected, actual interface{}) {
 }
 
 func Nil(t testing.TB, data interface{}) {
+	t.Helper()
 	if data == nil {
 		return
 	}
@@ -60,6 +61,7 @@ func Nil(t testing.TB, data interface{}) {
 }
 
 func NotNil(t testing.TB, data interface{}) {
+	t.Helper()
 	if data == nil {
 		return
 	}
diff --git a/pkg/common/test/mock/network.go b/pkg/common/test/mock/network.go
index 22c8266..0375982 100644
--- a/pkg/common/test/mock/network.go
+++ b/pkg/common/test/mock/network.go
@@ -42,7 +42,7 @@ type Recorder interface {
 
 func (m *Conn) SetWriteTimeout(t time.Duration) error {
 	// TODO implement me
-	panic("implement me")
+	return nil
 }
 
 type SlowReadConn struct {
@@ -50,8 +50,12 @@ type SlowReadConn struct {
 }
 
 func (m *SlowReadConn) SetWriteTimeout(t time.Duration) error {
-	// TODO implement me
-	panic("implement me")
+	return nil
+}
+
+func (m *SlowReadConn) SetReadTimeout(t time.Duration) error {
+	m.Conn.readTimeout = t
+	return nil
 }
 
 func SlowReadDialer(addr string) (network.Conn, error) {
@@ -133,9 +137,12 @@ func (r *recorder) WroteLen() int {
 
 func (m *SlowReadConn) Peek(i int) ([]byte, error) {
 	b, err := m.zr.Peek(i)
-	time.Sleep(100 * time.Millisecond)
-	if err != nil || len(b) != i {
+	if m.readTimeout > 0 {
 		time.Sleep(m.readTimeout)
+	} else {
+		time.Sleep(100 * time.Millisecond)
+	}
+	if err != nil || len(b) != i {
 		return nil, errs.ErrReadTimeout
 	}
 	return b, err
@@ -152,7 +159,7 @@ func NewConn(source string) *Conn {
 }
 
 func NewSlowReadConn(source string) *SlowReadConn {
-	return &SlowReadConn{NewConn(source)}
+	return &SlowReadConn{Conn: NewConn(source)}
 }
 
 type SlowWriteConn struct {
diff --git a/pkg/common/test/mock/network_test.go b/pkg/common/test/mock/network_test.go
new file mode 100644
index 0000000..84df1cd
--- /dev/null
+++ b/pkg/common/test/mock/network_test.go
@@ -0,0 +1,184 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package mock
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	errs "github.com/cloudwego/hertz/pkg/common/errors"
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+	"github.com/cloudwego/netpoll"
+)
+
+func TestConn(t *testing.T) {
+	t.Run("TestReader", func(t *testing.T) {
+		s1 := "abcdef4343"
+		conn1 := NewConn(s1)
+		err := conn1.SetReadDeadline(time.Now().Add(time.Millisecond * 100))
+		assert.DeepEqual(t, nil, err)
+		err = conn1.SetReadTimeout(time.Millisecond * 100)
+		assert.DeepEqual(t, nil, err)
+
+		// Peek Skip Read
+		b, _ := conn1.Peek(1)
+		assert.DeepEqual(t, []byte{'a'}, b)
+		conn1.Skip(1)
+		readByte, _ := conn1.ReadByte()
+		assert.DeepEqual(t, byte('b'), readByte)
+
+		p := make([]byte, 100)
+		n, err := conn1.Read(p)
+		assert.DeepEqual(t, nil, err)
+		assert.DeepEqual(t, s1[2:], string(p[:n]))
+
+		_, err = conn1.Peek(1)
+		assert.DeepEqual(t, errs.ErrTimeout, err)
+
+		conn2 := NewConn(s1)
+		p, _ = conn2.ReadBinary(len(s1))
+		assert.DeepEqual(t, s1, string(p))
+		assert.DeepEqual(t, 0, conn2.Len())
+		// Reader
+		assert.DeepEqual(t, conn2.zr, conn2.Reader())
+	})
+
+	t.Run("TestReadWriter", func(t *testing.T) {
+		s1 := "abcdef4343"
+		conn := NewConn(s1)
+		p, err := conn.ReadBinary(len(s1))
+		assert.DeepEqual(t, nil, err)
+		assert.DeepEqual(t, s1, string(p))
+
+		wr := conn.WriterRecorder()
+		s2 := "efghljk"
+		// WriteBinary
+		n, err := conn.WriteBinary([]byte(s2))
+		assert.DeepEqual(t, nil, err)
+		assert.DeepEqual(t, len(s2), n)
+		assert.DeepEqual(t, len(s2), wr.WroteLen())
+
+		// Flush
+		p, _ = wr.ReadBinary(len(s2))
+		assert.DeepEqual(t, len(p), 0)
+
+		conn.Flush()
+		p, _ = wr.ReadBinary(len(s2))
+		assert.DeepEqual(t, s2, string(p))
+
+		// Write
+		s3 := "foobarbaz"
+		n, err = conn.Write([]byte(s3))
+		assert.DeepEqual(t, nil, err)
+		assert.DeepEqual(t, len(s3), n)
+		p, _ = wr.ReadBinary(len(s3))
+		assert.DeepEqual(t, s3, string(p))
+
+		// Malloc
+		buf, _ := conn.Malloc(10)
+		assert.DeepEqual(t, 10, len(buf))
+		// Writer
+		assert.DeepEqual(t, conn.zw, conn.Writer())
+
+		_, err = DialerFun("")
+		assert.DeepEqual(t, nil, err)
+	})
+
+	t.Run("TestNotImplement", func(t *testing.T) {
+		conn := NewConn("")
+		t1 := time.Now().Add(time.Millisecond)
+		du1 := time.Second
+		assert.DeepEqual(t, nil, conn.Release())
+		assert.DeepEqual(t, nil, conn.Close())
+		assert.DeepEqual(t, nil, conn.LocalAddr())
+		assert.DeepEqual(t, nil, conn.RemoteAddr())
+		assert.DeepEqual(t, nil, conn.SetIdleTimeout(du1))
+		assert.Panic(t, func() {
+			conn.SetDeadline(t1)
+		})
+		assert.Panic(t, func() {
+			conn.SetWriteDeadline(t1)
+		})
+		assert.Panic(t, func() {
+			conn.IsActive()
+		})
+		assert.Panic(t, func() {
+			conn.SetOnRequest(func(ctx context.Context, connection netpoll.Connection) error {
+				return nil
+			})
+		})
+		assert.Panic(t, func() {
+			conn.AddCloseCallback(func(connection netpoll.Connection) error {
+				return nil
+			})
+		})
+	})
+}
+
+func TestSlowConn(t *testing.T) {
+	t.Run("TestSlowReadConn", func(t *testing.T) {
+		s1 := "abcdefg"
+		conn := NewSlowReadConn(s1)
+		b, err := conn.Peek(4)
+		assert.DeepEqual(t, nil, err)
+		assert.DeepEqual(t, s1[:4], string(b))
+		conn.Skip(len(s1))
+		_, err = conn.Peek(1)
+		assert.DeepEqual(t, errs.ErrReadTimeout, err)
+		_, err = SlowReadDialer("")
+		assert.DeepEqual(t, nil, err)
+	})
+
+	t.Run("TestSlowWriteConn", func(t *testing.T) {
+		conn, err := SlowWriteDialer("")
+		assert.DeepEqual(t, nil, err)
+		conn.SetWriteTimeout(time.Millisecond * 100)
+		err = conn.Flush()
+		assert.DeepEqual(t, errs.ErrWriteTimeout, err)
+	})
+}
+
+func TestStreamConn(t *testing.T) {
+	t.Run("TestStreamConn", func(t *testing.T) {
+		conn := NewStreamConn()
+		_, err := conn.Peek(10)
+		assert.DeepEqual(t, nil, err)
+		conn.Skip(conn.Len())
+		assert.DeepEqual(t, 0, conn.Len())
+		_, err = conn.Peek(10)
+		assert.DeepEqual(t, "not enough data", err.Error())
+		_, err = conn.Peek(1)
+		assert.DeepEqual(t, nil, err)
+		assert.DeepEqual(t, cap(conn.Data), conn.Len())
+		err = conn.Skip(conn.Len() + 1)
+		assert.DeepEqual(t, "not enough data", err.Error())
+	})
+
+	t.Run("TestNotImplement", func(t *testing.T) {
+		conn := NewStreamConn()
+		assert.Panic(t, func() {
+			conn.Release()
+		})
+		assert.Panic(t, func() {
+			conn.ReadByte()
+		})
+		assert.Panic(t, func() {
+			conn.ReadBinary(10)
+		})
+	})
+}
diff --git a/pkg/common/test/mock/reader_test.go b/pkg/common/test/mock/reader_test.go
new file mode 100644
index 0000000..c0a2eea
--- /dev/null
+++ b/pkg/common/test/mock/reader_test.go
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package mock
+
+import (
+	"bufio"
+	"io"
+	"testing"
+	"testing/iotest"
+
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+)
+
+func TestZeroCopyReader(t *testing.T) {
+	// raw
+	r := "abcdef4343"
+	zr := NewZeroCopyReader(r)
+	rs := readBytes(zr.Reader)
+	assert.DeepEqual(t, rs, r)
+
+	// peek
+	zr = NewZeroCopyReader(r)
+	s, err := zr.Peek(1)
+	assert.DeepEqual(t, nil, err)
+	assert.DeepEqual(t, "a", string(s))
+
+	s, err = zr.Peek(4)
+	assert.DeepEqual(t, nil, err)
+	assert.DeepEqual(t, "abcd", string(s))
+
+	// https://github.com/golang/go/issues/50569
+	ezr := NewZeroCopyReader("")
+	s, err = ezr.Peek(32)
+	assert.DeepEqual(t, io.EOF, err)
+	assert.DeepEqual(t, "", string(s))
+
+	// skip
+	err = zr.Skip(1)
+	assert.DeepEqual(t, nil, err)
+
+	s, err = zr.Peek(4)
+	assert.DeepEqual(t, nil, err)
+	assert.DeepEqual(t, "bcde", string(s))
+
+	// len
+	assert.DeepEqual(t, len(r)-1, zr.Len())
+	assert.DeepEqual(t, nil, ezr.Release())
+	assert.Panic(t, func() { // not implement
+		zr.ReadBinary(10)
+	})
+}
+
+func TestEOFReader(t *testing.T) {
+	r := &EOFReader{}
+	s, err := r.Peek(1)
+	assert.DeepEqual(t, io.EOF, err)
+	assert.DeepEqual(t, "", string(s))
+	assert.DeepEqual(t, nil, r.Skip(1))
+	assert.DeepEqual(t, 0, r.Len())
+
+	_, err = r.ReadByte()
+	assert.DeepEqual(t, io.EOF, err)
+	_, err = r.ReadBinary(10)
+	assert.DeepEqual(t, io.EOF, err)
+	_, err = r.Read(s)
+	assert.DeepEqual(t, io.EOF, err)
+	assert.DeepEqual(t, nil, r.Release())
+}
+
+func readBytes(buf *bufio.Reader) string {
+	var b [1000]byte
+	nb := 0
+	for {
+		c, err := buf.ReadByte()
+		if err == io.EOF {
+			break
+		}
+		if err == nil {
+			b[nb] = c
+			nb++
+		} else if err != iotest.ErrTimeout {
+			panic("Data: " + err.Error())
+		}
+	}
+	return string(b[0:nb])
+}
diff --git a/pkg/common/test/mock/writer_test.go b/pkg/common/test/mock/writer_test.go
new file mode 100644
index 0000000..f03d69b
--- /dev/null
+++ b/pkg/common/test/mock/writer_test.go
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package mock
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+)
+
+func TestExtWriter(t *testing.T) {
+	b1 := []byte("abcdef4343")
+	buf := new(bytes.Buffer)
+	isFinal := false
+	w := &ExtWriter{
+		Buf:     buf,
+		IsFinal: &isFinal,
+	}
+
+	// write
+	n, err := w.Write(b1)
+	assert.DeepEqual(t, nil, err)
+	assert.DeepEqual(t, len(b1), n)
+
+	// flush
+	err = w.Flush()
+	assert.DeepEqual(t, nil, err)
+	assert.DeepEqual(t, b1, w.Buf.Bytes())
+
+	// setbody
+	b2 := []byte("abc")
+	w.SetBody([]byte(b2))
+	err = w.Flush()
+	assert.DeepEqual(t, nil, err)
+	assert.DeepEqual(t, b2, w.Buf.Bytes())
+
+	w.Finalize()
+	assert.DeepEqual(t, true, *(w.IsFinal))
+}
diff --git a/pkg/network/netpoll/dial_test.go b/pkg/network/netpoll/dial_test.go
new file mode 100644
index 0000000..1830d1e
--- /dev/null
+++ b/pkg/network/netpoll/dial_test.go
@@ -0,0 +1,71 @@
+// Copyright 2023 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+//go:build !windows
+// +build !windows
+
+package netpoll
+
+import (
+	"context"
+	"crypto/tls"
+	"testing"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/common/config"
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+	"github.com/cloudwego/hertz/pkg/common/test/mock"
+)
+
+func TestDial(t *testing.T) {
+	t.Run("NetpollDial", func(t *testing.T) {
+		const nw = "tcp"
+		const addr = "localhost:10100"
+		transporter := NewTransporter(&config.Options{
+			Addr:    addr,
+			Network: nw,
+		})
+		go transporter.ListenAndServe(func(ctx context.Context, conn interface{}) error {
+			return nil
+		})
+		defer transporter.Close()
+		time.Sleep(100 * time.Millisecond)
+
+		dial := NewDialer()
+		// DialConnection
+		_, err := dial.DialConnection("tcp", "localhost:10101", time.Second, nil) // wrong addr
+		assert.NotNil(t, err)
+
+		nwConn, err := dial.DialConnection(nw, addr, time.Second, nil)
+		assert.Nil(t, err)
+		defer nwConn.Close()
+		_, err = nwConn.Write([]byte("abcdef"))
+		assert.Nil(t, err)
+		// DialTimeout
+		nConn, err := dial.DialTimeout(nw, addr, time.Second, nil)
+		assert.Nil(t, err)
+		defer nConn.Close()
+	})
+
+	t.Run("NotSupportTLS", func(t *testing.T) {
+		dial := NewDialer()
+		_, err := dial.AddTLS(mock.NewConn(""), nil)
+		assert.DeepEqual(t, errNotSupportTLS, err)
+		_, err = dial.DialConnection("tcp", "localhost:10102", time.Microsecond, &tls.Config{})
+		assert.DeepEqual(t, errNotSupportTLS, err)
+		_, err = dial.DialTimeout("tcp", "localhost:10102", time.Microsecond, &tls.Config{})
+		assert.DeepEqual(t, errNotSupportTLS, err)
+	})
+}
diff --git a/pkg/network/netpoll/transport_test.go b/pkg/network/netpoll/transport_test.go
new file mode 100644
index 0000000..d8a0609
--- /dev/null
+++ b/pkg/network/netpoll/transport_test.go
@@ -0,0 +1,100 @@
+// Copyright 2023 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+//go:build !windows
+// +build !windows
+
+package netpoll
+
+import (
+	"context"
+	"net"
+	"sync/atomic"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/common/config"
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+	"github.com/cloudwego/hertz/pkg/network"
+	"golang.org/x/sys/unix"
+)
+
+func TestTransport(t *testing.T) {
+	const nw = "tcp"
+	const addr = "localhost:10103"
+	t.Run("TestDefault", func(t *testing.T) {
+		var onConnFlag, onAcceptFlag, onDataFlag int32
+		transporter := NewTransporter(&config.Options{
+			Addr:    addr,
+			Network: nw,
+			OnConnect: func(ctx context.Context, conn network.Conn) context.Context {
+				atomic.StoreInt32(&onConnFlag, 1)
+				return ctx
+			},
+			WriteTimeout: time.Second,
+			OnAccept: func(conn net.Conn) context.Context {
+				atomic.StoreInt32(&onAcceptFlag, 1)
+				return context.Background()
+			},
+		})
+		go transporter.ListenAndServe(func(ctx context.Context, conn interface{}) error {
+			atomic.StoreInt32(&onDataFlag, 1)
+			return nil
+		})
+		defer transporter.Close()
+		time.Sleep(100 * time.Millisecond)
+
+		dial := NewDialer()
+		conn, err := dial.DialConnection(nw, addr, time.Second, nil)
+		assert.Nil(t, err)
+		_, err = conn.Write([]byte("123"))
+		assert.Nil(t, err)
+		time.Sleep(100 * time.Millisecond)
+
+		assert.Assert(t, atomic.LoadInt32(&onConnFlag) == 1)
+		assert.Assert(t, atomic.LoadInt32(&onAcceptFlag) == 1)
+		assert.Assert(t, atomic.LoadInt32(&onDataFlag) == 1)
+	})
+
+	t.Run("TestListenConfig", func(t *testing.T) {
+		listenCfg := &net.ListenConfig{Control: func(network, address string, c syscall.RawConn) error {
+			return c.Control(func(fd uintptr) {
+				syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, unix.SO_REUSEADDR, 1)
+				syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, unix.SO_REUSEPORT, 1)
+			})
+		}}
+		transporter := NewTransporter(&config.Options{
+			Addr:         addr,
+			Network:      nw,
+			ListenConfig: listenCfg,
+		})
+		go transporter.ListenAndServe(func(ctx context.Context, conn interface{}) error {
+			return nil
+		})
+		defer transporter.Close()
+	})
+
+	t.Run("TestExceptionCase", func(t *testing.T) {
+		assert.Panic(t, func() { // listen err
+			transporter := NewTransporter(&config.Options{
+				Network: "unknow",
+			})
+			transporter.ListenAndServe(func(ctx context.Context, conn interface{}) error {
+				return nil
+			})
+		})
+	})
+}
diff --git a/pkg/network/standard/connection_test.go b/pkg/network/standard/connection_test.go
index 8108f1c..9b0fafd 100644
--- a/pkg/network/standard/connection_test.go
+++ b/pkg/network/standard/connection_test.go
@@ -23,6 +23,7 @@ import (
 	"io"
 	"net"
 	"strings"
+	"syscall"
 	"testing"
 	"time"
 
@@ -263,6 +264,12 @@ func TestInitializeTLSConn(t *testing.T) {
 	assert.DeepEqual(t, tls.ConnectionState{}, tlsConn.ConnectionState())
 }
 
+func TestHandleSpecificError(t *testing.T) {
+	conn := &Conn{}
+	assert.DeepEqual(t, false, conn.HandleSpecificError(nil, ""))
+	assert.DeepEqual(t, true, conn.HandleSpecificError(syscall.EPIPE, ""))
+}
+
 type mockConn struct {
 	buffer     bytes.Buffer
 	localAddr  net.Addr
diff --git a/pkg/network/standard/dial_test.go b/pkg/network/standard/dial_test.go
new file mode 100644
index 0000000..7d9f3a0
--- /dev/null
+++ b/pkg/network/standard/dial_test.go
@@ -0,0 +1,174 @@
+/*
+ * Copyright 2023 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package standard
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/common/config"
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+)
+
+func TestDial(t *testing.T) {
+	const nw = "tcp"
+	const addr = "localhost:10104"
+	transporter := NewTransporter(&config.Options{
+		Addr:    addr,
+		Network: nw,
+	})
+
+	go transporter.ListenAndServe(func(ctx context.Context, conn interface{}) error {
+		return nil
+	})
+	defer transporter.Close()
+	time.Sleep(time.Millisecond * 100)
+
+	dial := NewDialer()
+	_, err := dial.DialConnection(nw, addr, time.Second, nil)
+	assert.Nil(t, err)
+
+	nConn, err := dial.DialTimeout(nw, addr, time.Second, nil)
+	assert.Nil(t, err)
+	defer nConn.Close()
+}
+
+func TestDialTLS(t *testing.T) {
+	const nw = "tcp"
+	const addr = "localhost:10105"
+	data := []byte("abcdefg")
+	listened := make(chan struct{})
+	go func() {
+		mockTLSServe(nw, addr, func(conn net.Conn) {
+			defer conn.Close()
+			_, err := conn.Write(data)
+			assert.Nil(t, err)
+		}, listened)
+	}()
+
+	select {
+	case <-listened:
+	case <-time.After(time.Second * 5):
+		t.Fatalf("timeout")
+	}
+
+	dial := NewDialer()
+	_, err := dial.DialConnection(nw, addr, time.Second, &tls.Config{
+		InsecureSkipVerify: true,
+	})
+	assert.Nil(t, err)
+
+	conn, err := dial.DialConnection(nw, addr, time.Second, nil)
+	assert.Nil(t, err)
+	nConn, err := dial.AddTLS(conn, &tls.Config{
+		InsecureSkipVerify: true,
+	})
+	assert.Nil(t, err)
+
+	buf := make([]byte, len(data))
+	_, err = nConn.Read(buf)
+	assert.Nil(t, err)
+	assert.DeepEqual(t, string(data), string(buf))
+}
+
+func mockTLSServe(nw, addr string, handle func(conn net.Conn), listened chan struct{}) (err error) {
+	certData, keyData, err := generateTestCertificate("")
+	if err != nil {
+		return
+	}
+
+	cert, err := tls.X509KeyPair(certData, keyData)
+	if err != nil {
+		return
+	}
+
+	tlsConfig := &tls.Config{
+		Certificates: []tls.Certificate{cert},
+	}
+	ln, err := tls.Listen(nw, addr, tlsConfig)
+	if err != nil {
+		return
+	}
+	defer ln.Close()
+
+	listened <- struct{}{}
+	for {
+		conn, err := ln.Accept()
+		if err != nil {
+			continue
+		}
+		go handle(conn)
+	}
+}
+
+// generateTestCertificate generates a test certificate and private key based on the given host.
+func generateTestCertificate(host string) ([]byte, []byte, error) {
+	priv, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	serialNumberLimit := new(big.Int).Lsh(big.NewInt(1), 128)
+	serialNumber, err := rand.Int(rand.Reader, serialNumberLimit)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	cert := &x509.Certificate{
+		SerialNumber: serialNumber,
+		Subject: pkix.Name{
+			Organization: []string{"fasthttp test"},
+		},
+		NotBefore:             time.Now(),
+		NotAfter:              time.Now().Add(365 * 24 * time.Hour),
+		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
+		SignatureAlgorithm:    x509.SHA256WithRSA,
+		DNSNames:              []string{host},
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+	}
+
+	certBytes, err := x509.CreateCertificate(
+		rand.Reader, cert, cert, &priv.PublicKey, priv,
+	)
+
+	p := pem.EncodeToMemory(
+		&pem.Block{
+			Type:  "PRIVATE KEY",
+			Bytes: x509.MarshalPKCS1PrivateKey(priv),
+		},
+	)
+
+	b := pem.EncodeToMemory(
+		&pem.Block{
+			Type:  "CERTIFICATE",
+			Bytes: certBytes,
+		},
+	)
+
+	return b, p, err
+}
diff --git a/pkg/protocol/http1/client_unix_test.go b/pkg/protocol/http1/client_unix_test.go
index ff65bc4..ba616ad 100644
--- a/pkg/protocol/http1/client_unix_test.go
+++ b/pkg/protocol/http1/client_unix_test.go
@@ -32,9 +32,9 @@ import (
 
 func TestGcBodyStream(t *testing.T) {
 	srv := &http.Server{Addr: ":11001", Handler: http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-		w.Write([]byte("hello\n"))
-		time.Sleep(time.Second)
-		w.Write([]byte("world\n"))
+		for range [1024]int{} {
+			w.Write([]byte("hello world\n"))
+		}
 	})}
 	go srv.ListenAndServe()
 	time.Sleep(100 * time.Millisecond)
@@ -58,6 +58,8 @@ func TestGcBodyStream(t *testing.T) {
 	}
 
 	runtime.GC()
+	// wait for gc
+	time.Sleep(100 * time.Millisecond)
 	c.CloseIdleConnections()
 	assert.DeepEqual(t, 0, c.ConnPoolState().TotalConnNum)
 }