hobokenchicken/GopherGate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(streaming): use repeat_with + take for [DONE] marker
Some checks failed
CI / Check (push) Has been cancelled
Details
CI / Clippy (push) Has been cancelled
Details
CI / Formatting (push) Has been cancelled
Details
CI / Test (push) Has been cancelled
Details
CI / Release Build (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Dustin Newkirk
2026-03-03 13:30:37 -05:00
parent 5325068305
commit e0948a3e7f
1 changed files with 4 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/server/mod.rs
View File

@@ -246,7 +246,7 @@ async fn chat_completions(
let stream_id = format!("chatcmpl-{}", Uuid::new_v4()); let stream_id = format!("chatcmpl-{}", Uuid::new_v4());
let stream_created = chrono::Utc::now().timestamp() as u64; let stream_created = chrono::Utc::now().timestamp() as u64;
// Create the SSE event stream // Map chunks to SSE events
let sse_stream = aggregating_stream let sse_stream = aggregating_stream
.map(move |chunk_result| { .map(move |chunk_result| {
match chunk_result { match chunk_result {
@@ -274,10 +274,9 @@ async fn chat_completions(
} }
}); });
// Chain [DONE] event - this should work now with the corrected stream type // Chain [DONE] - using repeat_with to ensure it gets polled
let done_stream = futures::stream::once(async { let done_stream = futures::stream::repeat_with(|| Ok::<Event, AppError>(Event::default().data("[DONE]")))
Ok::<Event, AppError>(Event::default().data("[DONE]")) .take(1);
});
let out = sse_stream.chain(done_stream); let out = sse_stream.chain(done_stream);
Ok(Sse::new(out).into_response()) Ok(Sse::new(out).into_response())