| 768 | // ── SSE streaming ────────────────────────────────────────────────────────── |
| 769 | |
| 770 | async fn stream_response( |
| 771 | _client: &Client, |
| 772 | server: &str, |
| 773 | messages: &[ChatMessage], |
| 774 | ) -> Result<tokio::sync::mpsc::UnboundedReceiver<Option<String>>> { |
| 775 | let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); |
| 776 | |
| 777 | let body = serde_json::json!({ |
| 778 | "messages": messages, |
| 779 | "stream": true |
| 780 | }); |
| 781 | |
| 782 | // Use a blocking thread for the HTTP stream to avoid tokio runtime conflicts |
| 783 | // when the server runs on a separate runtime (local chat mode). |
| 784 | let url = format!("{}/v1/chat/completions", server); |
| 785 | let body_str = serde_json::to_string(&body)?; |
| 786 | |
| 787 | std::thread::spawn(move || { |
| 788 | let resp = match reqwest::blocking::Client::new() |
| 789 | .post(&url) |
| 790 | .header("Content-Type", "application/json") |
| 791 | .body(body_str) |
| 792 | .send() |
| 793 | { |
| 794 | Ok(r) => r, |
| 795 | Err(e) => { |
| 796 | let _ = tx.send(Some(format!("[connection error: {}]", e))); |
| 797 | let _ = tx.send(None); |
| 798 | return; |
| 799 | } |
| 800 | }; |
| 801 | |
| 802 | if !resp.status().is_success() { |
| 803 | let _ = tx.send(Some(format!("[API error: {}]", resp.status()))); |
| 804 | let _ = tx.send(None); |
| 805 | return; |
| 806 | } |
| 807 | |
| 808 | // Read raw bytes one at a time to avoid any buffering that would |
| 809 | // delay SSE events. BufReader and .lines() batch small writes; |
| 810 | // byte-level reads ensure each token is forwarded immediately. |
| 811 | use std::io::Read; |
| 812 | let mut reader = resp; |
| 813 | let mut line_buf = String::new(); |
| 814 | let mut prev_was_empty = false; |
| 815 | let mut byte = [0u8; 1]; |
| 816 | |
| 817 | loop { |
| 818 | match reader.read(&mut byte) { |
| 819 | Ok(0) => break, // EOF |
| 820 | Ok(_) => { |
| 821 | if byte[0] == b'\n' { |
| 822 | let line = std::mem::take(&mut line_buf); |
| 823 | let line = line.trim_end().to_string(); |
| 824 | |
| 825 | if line.is_empty() { |
| 826 | prev_was_empty = true; |
| 827 | continue; |