| 173 | } |
| 174 | |
| 175 | async fn post_streaming( |
| 176 | &self, |
| 177 | url: &str, |
| 178 | headers: Vec<(&str, &str)>, |
| 179 | body: &serde_json::Value, |
| 180 | cancel_token: CancellationToken, |
| 181 | ) -> Result<StreamingHttpResponse> { |
| 182 | let start = std::time::Instant::now(); |
| 183 | let request_body = serde_json::to_string(body).unwrap_or_default(); |
| 184 | let request_bytes = request_body.len() as u64; |
| 185 | |
| 186 | let mut request = self.client.post(url); |
| 187 | for (key, value) in headers { |
| 188 | request = request.header(key, value); |
| 189 | } |
| 190 | request = request.json(body); |
| 191 | |
| 192 | let response = tokio::select! { |
| 193 | _ = cancel_token.cancelled() => { |
| 194 | anyhow::bail!("HTTP streaming request cancelled"); |
| 195 | } |
| 196 | result = request.send() => { |
| 197 | result.context(format!("Failed to send streaming request to {}", url))? |
| 198 | } |
| 199 | }; |
| 200 | |
| 201 | let status = response.status().as_u16(); |
| 202 | let retry_after = response |
| 203 | .headers() |
| 204 | .get("retry-after") |
| 205 | .and_then(|v| v.to_str().ok()) |
| 206 | .map(String::from); |
| 207 | |
| 208 | // For streaming, we record metrics after sending but before consuming the stream |
| 209 | // Note: response_bytes is estimated as we can't know the full stream size upfront |
| 210 | let duration_ms = start.elapsed().as_secs_f64() * 1000.0; |
| 211 | maybe_record_metrics(HttpMetricsRecord { |
| 212 | url: url.to_string(), |
| 213 | method: "POST".to_string(), |
| 214 | status, |
| 215 | duration_ms, |
| 216 | request_bytes, |
| 217 | response_bytes: 0, // Unknown for streaming |
| 218 | streaming: true, |
| 219 | }); |
| 220 | |
| 221 | if (200..300).contains(&status) { |
| 222 | let byte_stream = response |
| 223 | .bytes_stream() |
| 224 | .map(|r| r.map_err(|e| anyhow::anyhow!("Stream error: {}", e))); |
| 225 | Ok(StreamingHttpResponse { |
| 226 | status, |
| 227 | retry_after, |
| 228 | byte_stream: Box::pin(byte_stream), |
| 229 | error_body: String::new(), |
| 230 | }) |
| 231 | } else { |
| 232 | let error_body = response.text().await.unwrap_or_default(); |