Function stream_anthropic_sse

src-tauri/src/services/chat_service.rs:725–783

(
    response: reqwest::Response,
    on_token: &Channel<String>,
    cancel_token: &CancellationToken,
)

}

async fn stream_anthropic_sse(
    response: reqwest::Response,
    on_token: &Channel<String>,
    cancel_token: &CancellationToken,
) -> AppResult<(String, Option<TokenUsage>)> {
    let mut stream = response.bytes_stream();
    let mut line_buffer = String::new();
    let mut output = String::new();
    // Track the most recent `event:` line so we can use it when parsing the
    // subsequent `data:` line.  Some gateways omit the `"type"` field from
    // the JSON payload, so we fall back to the SSE event name.
    let mut current_event = String::new();
    let mut usage = UsageAccumulator::default();
    let mut message_stop_received = false;

    'outer: loop {
        tokio::select! {
            _ = cancel_token.cancelled() => {
                return Err(AppError::Cancelled);
            }
            chunk = stream.next() => {
                match chunk {
                    Some(Ok(bytes)) => {
                        line_buffer.push_str(&String::from_utf8_lossy(&bytes));

                        while let Some(pos) = line_buffer.find('\n') {
                            let mut line = line_buffer[..pos].to_string();
                            line_buffer.drain(..=pos);
                            if line.ends_with('\r') {
                                line.pop();
                            }

                            if parse_anthropic_sse_line(
                                &line,
                                &mut current_event,
                                on_token,
                                &mut output,
                                &mut usage,
                            )? {
                                message_stop_received = true;
                                break 'outer;
                            }
                        }
                    }
                    Some(Err(e)) => return Err(AppError::Http(e.to_string())),
                    None => break,
                }
            }
        }
    }

    if !message_stop_received {
        return Err(AppError::Http(
            "Stream ended without completion signal — connection may have been interrupted. Please retry.".to_string(),
        ));
    }

    Ok((output, usage.finish()))