Repository: apache/dubbo-go-pixiu

Issue:
Bug: when trying to hotreload config nil pointer panic occures


<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

-->


**What happened**:

When trying to hotreload config nil pointer panic occures

**How to reproduce it (as minimally and precisely as possible)**:

Run nacos-farconf sample, and try to update the config stored in nacos, pixiu will try to hotreload the config file. At this time, nil pointer panic occures.

**Anything else we need to know?**:
log:

```
2025-05-28T11:42:44.952+0800	INFO	config/config_load.go:60	[dubbopixiu go] load path:../../dubbo-go-pixiu-samples/dubbogo/simple/nacos_farconf//pixiu/conf.yaml
2025-05-28T11:42:44.952+0800	INFO	config/config_load.go:86	load config in YAML format from : /mnt/c/Users/Alan.lxt-redmi/Desktop/repoistories/dubbo-go-pixiu-samples/dubbogo/simple/nacos_farconf/pixiu/conf.yaml
2025-05-28T11:42:45.038+0800	INFO	config/config_load.go:137	http config:map[idle_timeout:4s read_timeout:5s write_timeout:6s] true
2025-05-28T11:42:45.038+0800	INFO	cmd/gateway.go:208	GOMAXPROCS set to 18
2025-05-28T11:42:45.038+0800	INFO	server/pixiu_start.go:127	[dubbopixiu go] start by config : &{StaticResources:{Listeners:[0xc000799d90] Clusters:[] Adapters:[] ShutdownConfig:0xc000797aa0 PprofConf:{Enable:false Address:{SocketAddress:{Address:0.0.0.0 Port:8881 ResolverName: Domains:[] CertsDir:} Name:}}} DynamicResources:<nil> Metric:{Enable:false PrometheusPort:0} Node:<nil> Trace:<nil> Wasm:<nil> Config:0xc0004fb5c0 Nacos:0xc00064f2c0 Log:<nil>}
2025-05-28 11:42:45	INFO	config/root_config.go:138	[Config Center] Config center doesn't start
2025/05/28 11:42:45 Connected to 127.0.0.1:2181
2025-05-28T11:42:45.042+0800	INFO	proxy/descriptor_source.go:51	[dubbo-go-pixiu] could not load protoset files: no protoset files provided
2025-05-28T11:42:45.042+0800	INFO	tracing/driver.go:76	[dubbo-go-pixiu] no trace configuration in conf.yaml
2025-05-28T11:42:45.042+0800	INFO	http/http_listener.go:157	[dubbo-go-server] httpListener start at : 0.0.0.0:8881
2025/05/28 11:42:45 Authenticated: id=72057594730840067, timeout=4000
2025/05/28 11:42:45 Re-submitting `0` credentials after reconnect
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x1172d34]

goroutine 89 [running]:
github.com/apache/dubbo-go-pixiu/pkg/hotreload.(*LoggerReloader).CheckUpdate(0x3749f60, 0xc00040dc00?, 0x0?)
	/mnt/c/Users/Alan.lxt-redmi/Desktop/repoistories/review/dubbo-go-pixiu/pkg/hotreload/logger.go:42 +0x74
github.com/apache/dubbo-go-pixiu/pkg/hotreload.(*Coordinator).hotReload(0x36f3f20, 0xc00040dc00)
	/mnt/c/Users/Alan.lxt-redmi/Desktop/repoistories/review/dubbo-go-pixiu/pkg/hotreload/hotreload.go:85 +0x97
github.com/apache/dubbo-go-pixiu/pkg/hotreload.(*Coordinator).HotReload(0x36f3f20)
	/mnt/c/Users/Alan.lxt-redmi/Desktop/repoistories/review/dubbo-go-pixiu/pkg/hotreload/hotreload.go:75 +0x52
created by github.com/apache/dubbo-go-pixiu/pkg/hotreload.StartHotReload in goroutine 1
	/mnt/c/Users/Alan.lxt-redmi/Desktop/repoistories/review/dubbo-go-pixiu/pkg/hotreload/hotreload.go:62 +0x7f
exit status 2
```

Discussion:

Solution patch:
diff --git a/configs/conf.yaml b/configs/conf.yaml
index 3f4f4a3a..5cfca4d3 100644
--- a/configs/conf.yaml
+++ b/configs/conf.yaml
@@ -50,9 +50,9 @@ static_resources:
                     allow_credentials: false
 
       config:
-        idle_timeout: 5s
-        read_timeout: 5s
-        write_timeout: 5s
+        idle_timeout: "5s"
+        read_timeout: "5s"
+        write_timeout: "5s"
   clusters:
     - name: "user"
       lb_policy: "lb"
diff --git a/pkg/client/http/http.go b/pkg/client/http/http.go
index c5e2dd12..3a8efce0 100644
--- a/pkg/client/http/http.go
+++ b/pkg/client/http/http.go
@@ -20,13 +20,13 @@ package http
 import (
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 	"sync"
 )
 
 import (
 	"github.com/pkg/errors"
-
 	"go.opentelemetry.io/otel"
 	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
 	"go.opentelemetry.io/otel/trace"
@@ -206,3 +206,51 @@ func IsSSEStream(resp *http.Response) bool {
 	contentType := resp.Header.Get(constant.HeaderKeyContextType)
 	return strings.Contains(contentType, constant.HeaderValueTextEventStream)
 }
+
+// IsStreamableResponse check if the response is streamable
+// Determine whether it is a streaming response based on the following conditions:
+// 1. Using Transfer-Encoding: chunked
+// 2. The content type indicates that this is a streamable response (e.g. text/event-stream, application/json, etc.)
+func IsStreamableResponse(resp *http.Response) bool {
+	if IsSSEStream(resp) {
+		return true
+	}
+
+	// check if the block encoded transfer is used
+	transferEncoding := resp.Header.Get(constant.HeaderKeyTransferEncoding)
+	if strings.Contains(strings.ToLower(transferEncoding), constant.HeaderValueChunked) {
+		return true
+	}
+
+	// check the content type
+	contentType := resp.Header.Get(constant.HeaderKeyContextType)
+
+	// check if it s a streamable content type
+	streamableTypes := []string{
+		constant.HeaderValueTextPrefix,
+		constant.HeaderValueApplicationJson,
+		constant.HeaderValueApplicationNDJson,
+		constant.HeaderValueApplicationOctetStream,
+	}
+
+	for _, streamableType := range streamableTypes {
+		if strings.HasPrefix(contentType, streamableType) {
+			// For these content types, if you don't have Content-Length set or if Content-Length is large,
+			// may be a good candidate for streaming
+			contentLength := resp.Header.Get(constant.HeaderKeyContentLength)
+
+			// If Content-Length is not specified, it is possible that the server is not aware of the content length
+			if contentLength == "" {
+				return true
+			}
+
+			// If the Content-Length is large (> 1MB), streaming is also suitable
+			length, err := strconv.ParseInt(contentLength, 10, 64)
+			if err == nil && length > 1024*1024 {
+				return true
+			}
+		}
+	}
+
+	return false
+}
diff --git a/pkg/client/response.go b/pkg/client/response.go
index fac5fd95..f9b6a81e 100644
--- a/pkg/client/response.go
+++ b/pkg/client/response.go
@@ -21,34 +21,27 @@ import (
 	"io"
 )
 
-type Response interface {
-	IsStream() bool
-}
-
 // UnaryResponse response from endpoint
 type UnaryResponse struct {
 	Data []byte
 }
 
-func (r *UnaryResponse) IsStream() bool {
-	return false
-}
-
-// NewByteResponse create response contains a []byte
-func NewByteResponse(data []byte) *UnaryResponse {
+// NewUnaryResponse create response contains a []byte
+func NewUnaryResponse(data []byte) *UnaryResponse {
 	return &UnaryResponse{Data: data}
 }
 
 // StreamResponse response from endpoint
 type StreamResponse struct {
-	Stream io.ReadCloser
+	Stream      io.ReadCloser
+	IsSSEStream bool
 }
 
-func (r *StreamResponse) IsStream() bool {
-	return true
+func (r *StreamResponse) IsSSE() bool {
+	return r.IsSSEStream
 }
 
 // NewStreamResponse create response contains a stream
-func NewStreamResponse(stream io.ReadCloser) *StreamResponse {
-	return &StreamResponse{Stream: stream}
+func NewStreamResponse(stream io.ReadCloser, isSSE bool) *StreamResponse {
+	return &StreamResponse{Stream: stream, IsSSEStream: isSSE}
 }
diff --git a/pkg/common/constant/http.go b/pkg/common/constant/http.go
index ac10c843..b76dfeb0 100644
--- a/pkg/common/constant/http.go
+++ b/pkg/common/constant/http.go
@@ -18,9 +18,11 @@
 package constant
 
 const (
-	HeaderKeyContextType  = "Content-Type"
-	HeaderKeyCacheControl = "Cache-Control"
-	HeaderKeyConnection   = "Connection"
+	HeaderKeyContextType      = "Content-Type"
+	HeaderKeyCacheControl     = "Cache-Control"
+	HeaderKeyConnection       = "Connection"
+	HeaderKeyTransferEncoding = "Transfer-Encoding"
+	HeaderKeyContentLength    = "Content-Length"
 
 	HeaderKeyAccessControlAllowOrigin      = "Access-Control-Allow-Origin"
 	HeaderKeyAccessControlAllowHeaders     = "Access-Control-Allow-Headers"
@@ -29,10 +31,15 @@ const (
 	HeaderKeyAccessControlMaxAge           = "Access-Control-Max-Age"
 	HeaderKeyAccessControlAllowCredentials = "Access-Control-Allow-Credentials"
 
-	HeaderValueJsonUtf8        = "application/json;charset=UTF-8"
-	HeaderValueTextPlain       = "text/plain"
-	HeaderValueTextEventStream = "text/event-stream"
-	HeaderValueApplicationJson = "application/json"
+	HeaderValueJsonUtf8               = "application/json;charset=UTF-8"
+	HeaderValueTextPlain              = "text/plain"
+	HeaderValueTextEventStream        = "text/event-stream"
+	HeaderValueApplicationJson        = "application/json"
+	HeaderValueApplicationOctetStream = "application/octet-stream"
+	HeaderValueApplicationNDJson      = "application/x-ndjson"
+	HeaderValueImageJpeg              = "image/jpeg"
+	HeaderValueChunked                = "chunked"
+	HeaderValueTextPrefix             = "text/"
 
 	HeaderValueKeepAlive = "keep-alive"
 	HeaderValueNoCache   = "no-cache"
diff --git a/pkg/common/http/manager.go b/pkg/common/http/manager.go
index 703e4b1c..ee9a5833 100644
--- a/pkg/common/http/manager.go
+++ b/pkg/common/http/manager.go
@@ -179,6 +179,11 @@ func (hcm *HttpConnectionManager) writeResponse(c *pch.HttpContext) {
 							_ = res.Stream.Close()
 							return
 						}
+
+						if flusher, ok := c.Writer.(stdHttp.Flusher); ok {
+							flusher.Flush()
+						}
+
 					case err := <-errC:
 						if err != nil && err != io.EOF {
 							logger.Errorf("Stream error: %v", err)
@@ -190,7 +195,6 @@ func (hcm *HttpConnectionManager) writeResponse(c *pch.HttpContext) {
 			default:
 				logger.Errorf("Unknown response type: %T", c.TargetResp)
 			}
-
 		}
 	}
 }
@@ -210,8 +214,8 @@ func (hcm *HttpConnectionManager) buildTargetResponse(c *pch.HttpContext) {
 		//status code
 		c.StatusCode(res.StatusCode)
 
-		if http.IsSSEStream(res) {
-			c.TargetResp = &client.StreamResponse{Stream: res.Body}
+		if http.IsStreamableResponse(res) {
+			c.TargetResp = client.NewStreamResponse(res.Body, http.IsSSEStream(res))
 		} else {
 			body, err := io.ReadAll(res.Body)
 			if err != nil {
@@ -219,7 +223,7 @@ func (hcm *HttpConnectionManager) buildTargetResponse(c *pch.HttpContext) {
 			}
 			//close body
 			_ = res.Body.Close()
-			c.TargetResp = &client.UnaryResponse{Data: body}
+			c.TargetResp = client.NewUnaryResponse(body)
 		}
 	case []byte:
 		c.StatusCode(stdHttp.StatusOK)
@@ -228,7 +232,7 @@ func (hcm *HttpConnectionManager) buildTargetResponse(c *pch.HttpContext) {
 		} else {
 			c.AddHeader(constant.HeaderKeyContextType, constant.HeaderValueTextPlain)
 		}
-		c.TargetResp = &client.UnaryResponse{Data: res}
+		c.TargetResp = client.NewUnaryResponse(res)
 	default:
 		//dubbo go generic invoke
 		response := util.NewDubboResponse(res, false)
diff --git a/pkg/context/http/context.go b/pkg/context/http/context.go
index 9674f1a0..68f47449 100644
--- a/pkg/context/http/context.go
+++ b/pkg/context/http/context.go
@@ -59,7 +59,7 @@ type HttpContext struct {
 	// localReplyBody: happen error
 	localReplyBody []byte
 	// the response context will return.
-	TargetResp client.Response
+	TargetResp any
 	// client call response.
 	SourceResp any
 
diff --git a/pkg/filter/metric/metric.go b/pkg/filter/metric/metric.go
index 6aaf600b..72fdf484 100644
--- a/pkg/filter/metric/metric.go
+++ b/pkg/filter/metric/metric.go
@@ -25,9 +25,7 @@ import (
 
 import (
 	"github.com/pkg/errors"
-
 	"go.opentelemetry.io/otel/attribute"
-
 	"go.opentelemetry.io/otel/metric/global"
 	"go.opentelemetry.io/otel/metric/instrument"
 	"go.opentelemetry.io/otel/metric/instrument/syncint64"
@@ -139,7 +137,7 @@ func (f *Filter) Encode(c *http.HttpContext) filter.FilterStatus {
 	return filter.Continue
 }
 
-func computeApproximateResponseSize(res client.Response) (int, error) {
+func computeApproximateResponseSize(res any) (int, error) {
 	if res == nil {
 		return 0, errors.New("client.UnaryResponse is null pointer ")
 	}
diff --git a/pkg/hotreload/logger.go b/pkg/hotreload/logger.go
index eb2a024f..6bc00424 100644
--- a/pkg/hotreload/logger.go
+++ b/pkg/hotreload/logger.go
@@ -17,6 +17,10 @@
 
 package hotreload
 
+import (
+	"go.uber.org/zap"
+)
+
 import (
 	"github.com/apache/dubbo-go-pixiu/pkg/logger"
 	"github.com/apache/dubbo-go-pixiu/pkg/model"
@@ -30,12 +34,12 @@ func (r *LoggerReloader) CheckUpdate(oldConfig, newConfig *model.Bootstrap) bool
 	oc := oldConfig.Log
 	nc := newConfig.Log
 
-	if oc == nil && nc != nil {
+	if nc == nil && oc != nil {
 		return true
-	}
-
-	if oc != nil && nc == nil {
+	} else if nc == nil && oc == nil {
 		return false
+	} else if nc != nil && oc == nil {
+		return true
 	}
 
 	// Check if any logger configuration fields have changed.
@@ -67,7 +71,11 @@ func (r *LoggerReloader) CheckUpdate(oldConfig, newConfig *model.Bootstrap) bool
 
 // HotReload applies the new logger configuration.
 func (r *LoggerReloader) HotReload(oldConfig, newConfig *model.Bootstrap) error {
-	if err := logger.HotReload(newConfig.Log.Build()); err != nil {
+	var newConfBuild *zap.Config
+	if newConfig.Log != nil {
+		newConfBuild = newConfig.Log.Build()
+	}
+	if err := logger.HotReload(newConfBuild); err != nil {
 		logger.Errorf("Failed to reload logger configuration: %v", err)
 		return err
 	}
diff --git a/pkg/listener/http/http_listener.go b/pkg/listener/http/http_listener.go
index d18c4157..b34cc684 100644
--- a/pkg/listener/http/http_listener.go
+++ b/pkg/listener/http/http_listener.go
@@ -20,7 +20,6 @@ package http
 import (
 	"context"
 	"fmt"
-	"log"
 	"net/http"
 	"strconv"
 	"sync"
@@ -110,7 +109,7 @@ func (ls *HttpListenerService) httpsListener() {
 	hl := createDefaultHttpWorker(ls)
 
 	// user customize http config
-	hc := model.MapInStruct(ls.Config)
+	hc := model.MapInStruct(ls.Config.Config)
 
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", hl.ServeHTTP)
@@ -139,7 +138,7 @@ func (ls *HttpListenerService) httpListener() {
 	hl := createDefaultHttpWorker(ls)
 
 	// user customize http config
-	hc := model.MapInStruct(ls.Config)
+	hc := model.MapInStruct(ls.Config.Config)
 
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", hl.ServeHTTP)
@@ -154,9 +153,16 @@ func (ls *HttpListenerService) httpListener() {
 		MaxHeaderBytes: resolveInt2IntProp(hc.MaxHeaderBytes, 1<<20),
 	}
 
-	logger.Infof("[dubbo-go-server] httpListener start at : %s", ls.srv.Addr)
+	logger.Infof("[dubbo-go-server] httpListener starting at %s with WriteTimeout: %v, IdleTimeout: %v, ReadTimeout: %v",
+		ls.srv.Addr, ls.srv.WriteTimeout, ls.srv.IdleTimeout, ls.srv.ReadTimeout)
 
-	log.Println(ls.srv.ListenAndServe())
+	err := ls.srv.ListenAndServe()
+	// Improved error logging and replace log.Println
+	if err != nil && !errors.Is(err, http.ErrServerClosed) {
+		logger.Errorf("[dubbo-go-server] httpListener ListenAndServe error: %v", err)
+	} else {
+		logger.Info("[dubbo-go-server] httpListener stopped gracefully.")
+	}
 }
 
 // createDefaultHttpWorker create http listener
@@ -184,6 +190,7 @@ func resolveStr2Time(currentV string, defaultV time.Duration) time.Duration {
 		return defaultV
 	} else {
 		if duration, err := time.ParseDuration(currentV); err != nil {
+			logger.Errorf("Parse duration failed, err: %v", err)
 			return 20 * time.Second
 		} else {
 			return duration
diff --git a/pkg/logger/controller.go b/pkg/logger/controller.go
index 8914cf5d..fb2d9708 100644
--- a/pkg/logger/controller.go
+++ b/pkg/logger/controller.go
@@ -31,7 +31,7 @@ import (
 type logController struct {
 	mu sync.RWMutex
 
-	logger *logger
+	logger *pixiuLogger
 }
 
 // setLoggerLevel safely changes the log level in a concurrent manner.
@@ -45,12 +45,12 @@ func (c *logController) setLoggerLevel(level string) bool {
 
 	c.logger.config.Level = *lvl
 	l, _ := c.logger.config.Build(zap.AddCallerSkip(2))
-	c.logger = &logger{SugaredLogger: l.Sugar(), config: c.logger.config}
+	c.logger = &pixiuLogger{SugaredLogger: l.Sugar(), config: c.logger.config}
 	return true
 }
 
 // updateLogger safely modifies the log object in a concurrent manner.
-func (c *logController) updateLogger(l *logger) {
+func (c *logController) updateLogger(l *pixiuLogger) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.logger = l
diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go
index 9fd73af1..f2db95e8 100644
--- a/pkg/logger/logger.go
+++ b/pkg/logger/logger.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"os"
 	"path"
+	"strings"
 )
 
 import (
@@ -35,7 +36,7 @@ import (
 
 var control *logController
 
-type logger struct {
+type pixiuLogger struct {
 	*zap.SugaredLogger
 	config *zap.Config
 }
@@ -48,6 +49,21 @@ func init() {
 	}
 }
 
+// PaddedCallerEncoder is a custom caller encoder that ensures that all file paths are displayed at the same length
+func PaddedCallerEncoder(caller zapcore.EntryCaller, enc zapcore.PrimitiveArrayEncoder) {
+
+	callerPath := caller.TrimmedPath()
+
+	// Set a fixed length, and if the path is too short, add a space after it
+	const fixedLength = 30
+	if len(callerPath) < fixedLength {
+		padding := strings.Repeat(" ", fixedLength-len(callerPath))
+		callerPath = callerPath + padding
+	}
+
+	enc.AppendString(callerPath)
+}
+
 // InitLog load from config path
 func InitLog(logConfFile string) error {
 	if logConfFile == "" {
@@ -69,7 +85,7 @@ func InitLog(logConfFile string) error {
 	err = yaml.UnmarshalYML(confFileStream, conf)
 	if err != nil {
 		InitLogger(nil)
-		return perrors.New(fmt.Sprintf("[Unmarshal]init logger error: %v", err))
+		return perrors.New(fmt.Sprintf("[Unmarshal]init pixiuLogger error: %v", err))
 	}
 
 	InitLogger(conf)
@@ -84,21 +100,24 @@ func InitLogger(conf *zap.Config) {
 		zapLoggerEncoderConfig := zapcore.EncoderConfig{
 			TimeKey:        "time",
 			LevelKey:       "level",
-			NameKey:        "logger",
+			NameKey:        "pixiuLogger",
 			CallerKey:      "caller",
 			MessageKey:     "message",
 			StacktraceKey:  "stacktrace",
 			EncodeLevel:    zapcore.CapitalColorLevelEncoder,
 			EncodeTime:     zapcore.ISO8601TimeEncoder,
 			EncodeDuration: zapcore.SecondsDurationEncoder,
-			EncodeCaller:   zapcore.ShortCallerEncoder,
+			EncodeCaller:   PaddedCallerEncoder,
+			// EncodeCaller:   zapcore.ShortCallerEncoder,
 		}
 		zapLoggerConfig.EncoderConfig = zapLoggerEncoderConfig
 	} else {
 		zapLoggerConfig = *conf
+		// Set up a custom encoder directly without checking the original value
+		zapLoggerConfig.EncoderConfig.EncodeCaller = PaddedCallerEncoder
 	}
 	zapLogger, _ := zapLoggerConfig.Build(zap.AddCallerSkip(2))
-	l := &logger{zapLogger.Sugar(), &zapLoggerConfig}
+	l := &pixiuLogger{zapLogger.Sugar(), &zapLoggerConfig}
 
 	control.updateLogger(l)
 }
diff --git a/pkg/model/http.go b/pkg/model/http.go
index e9bbfc47..e9f26716 100644
--- a/pkg/model/http.go
+++ b/pkg/model/http.go
@@ -114,11 +114,11 @@ type HttpConfig struct {
 }
 
 func MapInStruct(cfg any) *HttpConfig {
-	var hc *HttpConfig
+	var hc HttpConfig
 	if cfg != nil {
-		if ok := mapstructure.Decode(cfg, &hc); ok != nil {
-			logger.Error("Config error", ok)
+		if err := mapstructure.Decode(cfg, &hc); err != nil {
+			logger.Error("Config error", err)
 		}
 	}
-	return hc
+	return &hc
 }
diff --git a/pkg/prometheus/prometheus.go b/pkg/prometheus/prometheus.go
index de00261b..e7171b97 100644
--- a/pkg/prometheus/prometheus.go
+++ b/pkg/prometheus/prometheus.go
@@ -29,9 +29,7 @@ import (
 
 import (
 	"github.com/dubbo-go-pixiu/pixiu-api/pkg/context"
-
 	"github.com/prometheus/client_golang/prometheus"
-
 	"github.com/prometheus/common/expfmt"
 )
 



Test patch:
diff --git a/pkg/common/http/manager_test.go b/pkg/common/http/manager_test.go
index 480272ef..292fd752 100644
--- a/pkg/common/http/manager_test.go
+++ b/pkg/common/http/manager_test.go
@@ -34,6 +34,8 @@ import (
 )
 
 import (
+	clienthttp "github.com/apache/dubbo-go-pixiu/pkg/client/http"
+	"github.com/apache/dubbo-go-pixiu/pkg/common/constant"
 	"github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter"
 	commonmock "github.com/apache/dubbo-go-pixiu/pkg/common/mock"
 	"github.com/apache/dubbo-go-pixiu/pkg/common/router/trie"
@@ -171,8 +173,8 @@ func TestStreamingResponse(t *testing.T) {
 
 	// mock server
 	upstreamServer, _ := NewTestServerWithURL("localhost:8080", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Type", "text/event-stream")
-		w.Header().Set("Cache-Control", "no-cache")
+		w.Header().Set(constant.HeaderKeyContextType, constant.HeaderValueTextEventStream)
+		w.Header().Set(constant.HeaderKeyCacheControl, constant.HeaderValueNoCache)
 		flusher := w.(http.Flusher)
 
 		for i := 1; i <= 3; i++ {
@@ -229,7 +231,6 @@ func TestStreamingResponse(t *testing.T) {
 			return
 		}
 	}
-
 }
 
 // mock recorder
@@ -262,6 +263,9 @@ func (r *StreamRecorder) Write(data []byte) (int, error) {
 	return len(data), nil
 }
 
+func (r *StreamRecorder) Flush() {
+}
+
 func NewTestServerWithURL(URL string, handler http.Handler) (*httptest.Server, error) {
 	ts := httptest.NewUnstartedServer(handler)
 	if URL != "" {
@@ -278,3 +282,221 @@ func NewTestServerWithURL(URL string, handler http.Handler) (*httptest.Server, e
 	ts.Start()
 	return ts, nil
 }
+
+// StreamHTTPRecorder Used to capture and test streaming HTTP responses over channels
+type StreamHTTPRecorder struct {
+	http.ResponseWriter
+	receivedBuf []string
+	headers     http.Header
+	status      int
+	flushCount  int
+}
+
+func NewStreamHTTPRecorder() *StreamHTTPRecorder {
+	return &StreamHTTPRecorder{
+		receivedBuf: make([]string, 0),
+		headers:     make(http.Header),
+		flushCount:  0,
+	}
+}
+
+func (r *StreamHTTPRecorder) Header() http.Header {
+	return r.headers
+}
+
+func (r *StreamHTTPRecorder) WriteHeader(statusCode int) {
+	r.status = statusCode
+}
+
+func (r *StreamHTTPRecorder) Write(data []byte) (int, error) {
+	eventCh <- string(data)
+	r.receivedBuf = append(r.receivedBuf, string(data))
+	return len(data), nil
+}
+
+func (r *StreamHTTPRecorder) Flush() {
+	r.flushCount++
+}
+
+// Test a variety of common streaming HTTP response types
+func TestStreamableHTTPResponse(t *testing.T) {
+	// define the type of content you want to test
+	contentTypes := []string{
+		constant.HeaderValueTextPlain,
+		constant.HeaderValueApplicationJson,
+		constant.HeaderValueApplicationOctetStream,
+		constant.HeaderValueApplicationNDJson,
+	}
+
+	for _, contentType := range contentTypes {
+		t.Run(fmt.Sprintf("ContentType_%s", contentType), func(t *testing.T) {
+			testStreamableResponse(t, contentType)
+		})
+	}
+}
+
+func testStreamableResponse(t *testing.T, contentType string) {
+	hcmc := model.HttpConnectionManagerConfig{
+		RouteConfig: model.RouteConfiguration{
+			RouteTrie: trie.NewTrieWithDefault("GET/api/stream", model.RouteAction{
+				Cluster: "mock_stream_cluster",
+			}),
+		},
+		HTTPFilters: []*model.HTTPFilter{
+			{
+				Name: commonmock.Kind,
+			},
+		},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+
+	// Clear any data that may have been left over from the previous test
+	for len(eventCh) > 0 {
+		<-eventCh
+	}
+
+	// mock server
+	upstreamServer, _ := NewTestServerWithURL("localhost:8080", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set(constant.HeaderKeyContextType, contentType)
+		flusher := w.(http.Flusher)
+
+		// Generate appropriate test data based on content type
+		var data []byte
+		for i := 1; i <= 5; i++ {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				time.Sleep(10 * time.Millisecond)
+
+				switch contentType {
+				case constant.HeaderValueApplicationJson:
+					data = []byte(fmt.Sprintf(`{"id": %d, "message": "test chunk %d"}\n`, i, i))
+				case constant.HeaderValueApplicationNDJson:
+					data = []byte(fmt.Sprintf(`{"id": %d, "message": "test chunk %d"}\n`, i, i))
+				case constant.HeaderValueApplicationOctetStream:
+					data = []byte(fmt.Sprintf("CHUNK-%d", i))
+				default: // text/plain
+					data = []byte(fmt.Sprintf("Chunk %d\n", i))
+				}
+
+				_, _ = w.Write(data)
+				flusher.Flush()
+				logger.Info("Upstream sent chunk ", i)
+			}
+		}
+	}))
+	defer upstreamServer.Close()
+
+	req := httptest.NewRequest("GET", "http://localhost:8080/api/stream", nil).WithContext(ctx)
+	done := make(chan struct{})
+
+	httpCtx := &contexthttp.HttpContext{
+		Request: req,
+		Writer:  NewStreamHTTPRecorder(),
+		Ctx:     ctx,
+	}
+
+	go func() {
+		defer close(done)
+
+		hcm := CreateHttpConnectionManager(&hcmc)
+
+		if err := hcm.Handle(httpCtx); err != nil {
+			t.Errorf("Handle failed: %v", err)
+		}
+
+		// verify that targetResp exists
+		if httpCtx.TargetResp == nil {
+			t.Error("TargetResp is nil")
+			return
+		}
+	}()
+
+	// collect and validate responses
+	receivedChunks := 0
+	for {
+		receivedEvents := httpCtx.Writer.(*StreamHTTPRecorder).receivedBuf
+		select {
+		case event := <-eventCh:
+			logger.Info("Received chunk: %s", event)
+			receivedChunks++
+		case <-done:
+			assert.Equal(t, 5, len(receivedEvents), "Should receive 5 chunks")
+			return
+		case <-time.After(5 * time.Second):
+			t.Fatal("Test timeout")
+			return
+		}
+	}
+}
+
+// TestIsStreamableResponse Test whether it is a function that can be streamed and responded
+func TestIsStreamableResponse(t *testing.T) {
+	tests := []struct {
+		name     string
+		headers  map[string]string
+		expected bool
+	}{
+		{
+			name: "sseResponse",
+			headers: map[string]string{
+				constant.HeaderKeyContextType: constant.HeaderValueTextEventStream,
+			},
+			expected: true,
+		},
+		{
+			name: "chunkedEncodingResponses",
+			headers: map[string]string{
+				constant.HeaderKeyContextType:      constant.HeaderValueApplicationJson,
+				constant.HeaderKeyTransferEncoding: constant.HeaderValueChunked,
+			},
+			expected: true,
+		},
+		{
+			name: "JsonResponseWithoutContent-Length",
+			headers: map[string]string{
+				constant.HeaderKeyContextType: constant.HeaderValueApplicationJson,
+			},
+			expected: true,
+		},
+		{
+			name: "The text response is large Content-Length",
+			headers: map[string]string{
+				constant.HeaderKeyContextType:   constant.HeaderValueTextPlain,
+				constant.HeaderKeyContentLength: "2097152", // 2MB
+			},
+			expected: true,
+		},
+		{
+			name: "JSON response Content-Length",
+			headers: map[string]string{
+				constant.HeaderKeyContextType:   constant.HeaderValueApplicationJson,
+				constant.HeaderKeyContentLength: "1024", // 1KB
+			},
+			expected: false,
+		},
+		{
+			name: "no stream Content-Type",
+			headers: map[string]string{
+				constant.HeaderKeyContextType: constant.HeaderValueImageJpeg,
+			},
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resp := &http.Response{
+				Header: make(http.Header),
+			}
+			for k, v := range tt.headers {
+				resp.Header.Set(k, v)
+			}
+			result := clienthttp.IsStreamableResponse(resp)
+			assert.Equal(t, tt.expected, result, "IsStreamableResponse() return err")
+		})
+	}
+}
diff --git a/pkg/filter/accesslog/access_log_test.go b/pkg/filter/accesslog/access_log_test.go
index 1fa4164e..e8c4d5c9 100644
--- a/pkg/filter/accesslog/access_log_test.go
+++ b/pkg/filter/accesslog/access_log_test.go
@@ -64,7 +64,7 @@ func TestApply(t *testing.T) {
 
 	request, _ := http.NewRequest("POST", "http://www.dubbogopixiu.com/mock/test?name=tc", bytes.NewReader([]byte("{\"id\":\"12345\"}")))
 	ctx := mock.GetMockHTTPContext(request)
-	ctx.TargetResp = client.NewByteResponse([]byte(msg))
+	ctx.TargetResp = client.NewUnaryResponse([]byte(msg))
 
 	f.Decode(ctx)
 	f.Encode(ctx)
diff --git a/pkg/filter/prometheus/metric_test.go b/pkg/filter/prometheus/metric_test.go
index ab7eb65f..b563a31f 100644
--- a/pkg/filter/prometheus/metric_test.go
+++ b/pkg/filter/prometheus/metric_test.go
@@ -62,7 +62,7 @@ func TestCounterExporterApiMetric(t *testing.T) {
 			body, _ := json.Marshal(&data)
 			request, _ := http.NewRequest("POST", "/_api/health", bytes.NewBuffer(body))
 			ctx := mock.GetMockHTTPContext(request)
-			ctx.TargetResp = client.NewByteResponse([]byte(msg))
+			ctx.TargetResp = client.NewUnaryResponse([]byte(msg))
 			err := factory.PrepareFilterChain(ctx, chain)
 			assert.Nil(t, err)
 			chain.OnDecode(ctx)