mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
fix: add X-Accel-Buffering header to SSE endpoints (#277)
Nginx buffers responses by default, breaking SSE connections that go through a reverse proxy. Add X-Accel-Buffering: no header to chat and log SSE handlers to match what compose-api and chat-api already do.
This commit is contained in:
+15
-11
@@ -564,9 +564,13 @@ pub async fn clear_auth_mode(state: &GatewayState) {
|
||||
async fn chat_events_handler(
|
||||
State(state): State<Arc<GatewayState>>,
|
||||
) -> Result<impl IntoResponse, (StatusCode, String)> {
|
||||
state.sse.subscribe().ok_or((
|
||||
let sse = state.sse.subscribe().ok_or((
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"Too many connections".to_string(),
|
||||
))?;
|
||||
Ok((
|
||||
[("X-Accel-Buffering", "no"), ("Cache-Control", "no-cache")],
|
||||
sse,
|
||||
))
|
||||
}
|
||||
|
||||
@@ -1592,10 +1596,7 @@ async fn job_files_read_handler(
|
||||
|
||||
async fn logs_events_handler(
|
||||
State(state): State<Arc<GatewayState>>,
|
||||
) -> Result<
|
||||
Sse<impl futures::Stream<Item = Result<Event, Infallible>> + Send + 'static>,
|
||||
(StatusCode, String),
|
||||
> {
|
||||
) -> Result<impl IntoResponse, (StatusCode, String)> {
|
||||
let broadcaster = state.log_broadcaster.as_ref().ok_or((
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"Log broadcaster not available".to_string(),
|
||||
@@ -1608,22 +1609,25 @@ async fn logs_events_handler(
|
||||
|
||||
let history_stream = futures::stream::iter(history).map(|entry| {
|
||||
let data = serde_json::to_string(&entry).unwrap_or_default();
|
||||
Ok(Event::default().event("log").data(data))
|
||||
Ok::<_, Infallible>(Event::default().event("log").data(data))
|
||||
});
|
||||
|
||||
let live_stream = tokio_stream::wrappers::BroadcastStream::new(rx)
|
||||
.filter_map(|result| result.ok())
|
||||
.map(|entry| {
|
||||
let data = serde_json::to_string(&entry).unwrap_or_default();
|
||||
Ok(Event::default().event("log").data(data))
|
||||
Ok::<_, Infallible>(Event::default().event("log").data(data))
|
||||
});
|
||||
|
||||
let stream = history_stream.chain(live_stream);
|
||||
|
||||
Ok(Sse::new(stream).keep_alive(
|
||||
KeepAlive::new()
|
||||
.interval(std::time::Duration::from_secs(30))
|
||||
.text(""),
|
||||
Ok((
|
||||
[("X-Accel-Buffering", "no"), ("Cache-Control", "no-cache")],
|
||||
Sse::new(stream).keep_alive(
|
||||
KeepAlive::new()
|
||||
.interval(std::time::Duration::from_secs(30))
|
||||
.text(""),
|
||||
),
|
||||
))
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user