mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-09-21 17:38:00 +00:00
fix(gateway): forward Flush through statusRecorder so SSE streams in real time
statusRecorder (the access-log wrapper) did not implement http.Flusher, so w.(http.Flusher) inside streamChat/streamChatAuto returned nil and every SSE chunk stayed buffered until the response ended. Add Flush() that delegates to the underlying writer when it supports flushing. Regression test: TestStatusRecorderFlusher pins the interface assertion and the forwarding path.
This commit is contained in:
@ -671,3 +671,34 @@ func TestHasScopeModelWithSourcePrefix(t *testing.T) {
|
|||||||
t.Error("AUTO scope should allow any prefixed model")
|
t.Error("AUTO scope should allow any prefixed model")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestStatusRecorderFlusher pins the SSE-critical contract: the access-log
|
||||||
|
// wrapper must implement http.Flusher, otherwise the streaming chat handlers'
|
||||||
|
// w.(http.Flusher) assertion yields nil and chunks are never flushed until
|
||||||
|
// the response ends (regression for the buffered-SSE bug).
|
||||||
|
func TestStatusRecorderFlusher(t *testing.T) {
|
||||||
|
var inner *flushRecorder
|
||||||
|
rr := httptest.NewRecorder()
|
||||||
|
inner = &flushRecorder{ResponseWriter: rr}
|
||||||
|
sr := &statusRecorder{ResponseWriter: inner}
|
||||||
|
|
||||||
|
fl, ok := any(sr).(http.Flusher)
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("statusRecorder does not implement http.Flusher — SSE streaming is broken")
|
||||||
|
}
|
||||||
|
if inner.flushed {
|
||||||
|
t.Fatal("Flush called before Flush()")
|
||||||
|
}
|
||||||
|
fl.Flush()
|
||||||
|
if !inner.flushed {
|
||||||
|
t.Fatal("statusRecorder.Flush did not forward to the underlying writer")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// flushRecorder records whether Flush was forwarded.
|
||||||
|
type flushRecorder struct {
|
||||||
|
http.ResponseWriter
|
||||||
|
flushed bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f *flushRecorder) Flush() { f.flushed = true }
|
||||||
|
|||||||
@ -113,6 +113,17 @@ func (s *statusRecorder) Write(b []byte) (int, error) {
|
|||||||
return s.ResponseWriter.Write(b)
|
return s.ResponseWriter.Write(b)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Flush forwards to the underlying writer so SSE handlers can stream
|
||||||
|
// incrementally through the access-log wrapper. Without this method the
|
||||||
|
// w.(http.Flusher) assertion inside the streaming chat handlers fails (the
|
||||||
|
// embedded ResponseWriter interface does not carry Flush into the method set),
|
||||||
|
// and every chunk stays buffered until the response ends.
|
||||||
|
func (s *statusRecorder) Flush() {
|
||||||
|
if f, ok := s.ResponseWriter.(http.Flusher); ok {
|
||||||
|
f.Flush()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (g *Gateway) routes(w http.ResponseWriter, r *http.Request) {
|
func (g *Gateway) routes(w http.ResponseWriter, r *http.Request) {
|
||||||
switch {
|
switch {
|
||||||
case r.URL.Path == "/v1/chat/completions":
|
case r.URL.Path == "/v1/chat/completions":
|
||||||
|
|||||||
Reference in New Issue
Block a user