(mut socket: WebSocket, sub: PtySubscription, cursor_param: i64)
| 3486 | } |
| 3487 | |
| 3488 | async fn handle_pty_websocket(mut socket: WebSocket, sub: PtySubscription, cursor_param: i64) { |
| 3489 | // --- Phase 1: Replay buffered output --- |
| 3490 | // Determine the byte offset to start replaying from. |
| 3491 | let from = if cursor_param == -1 { |
| 3492 | // Skip all buffered output. |
| 3493 | sub.cursor |
| 3494 | } else if cursor_param > 0 { |
| 3495 | cursor_param as usize |
| 3496 | } else { |
| 3497 | 0 |
| 3498 | }; |
| 3499 | |
| 3500 | if from < sub.cursor && !sub.buffer.is_empty() { |
| 3501 | let offset = from.saturating_sub(sub.buffer_start); |
| 3502 | if offset < sub.buffer.len() { |
| 3503 | let replay = &sub.buffer[offset..]; |
| 3504 | // Send in 64 KiB chunks to avoid oversized frames (matching TS). |
| 3505 | for chunk in replay.chunks(64 * 1024) { |
| 3506 | let bytes = axum::body::Bytes::copy_from_slice(chunk); |
| 3507 | if socket.send(Message::Binary(bytes)).await.is_err() { |
| 3508 | return; |
| 3509 | } |
| 3510 | } |
| 3511 | } |
| 3512 | } |
| 3513 | |
| 3514 | // Send metadata frame: 0x00 byte prefix + JSON `{"cursor":<n>}`. |
| 3515 | { |
| 3516 | let meta_json = format!("{{\"cursor\":{}}}", sub.cursor); |
| 3517 | let json_bytes = meta_json.as_bytes(); |
| 3518 | let mut frame = Vec::with_capacity(1 + json_bytes.len()); |
| 3519 | frame.push(0x00); |
| 3520 | frame.extend_from_slice(json_bytes); |
| 3521 | let bytes = axum::body::Bytes::from(frame); |
| 3522 | if socket.send(Message::Binary(bytes)).await.is_err() { |
| 3523 | return; |
| 3524 | } |
| 3525 | } |
| 3526 | |
| 3527 | // --- Phase 2: Bridge live I/O --- |
| 3528 | // Use a channel to decouple the broadcast receiver from the socket send |
| 3529 | // loop, since WebSocket requires &mut self for both send and recv. |
| 3530 | let (ws_tx, mut ws_rx) = mpsc::channel::<Vec<u8>>(256); |
| 3531 | let mut rx = sub.rx; |
| 3532 | let writer = sub.writer; |
| 3533 | |
| 3534 | // Task: forward broadcast PTY output into the mpsc channel. |
| 3535 | let forward_task = tokio::spawn(async move { |
| 3536 | loop { |
| 3537 | match rx.recv().await { |
| 3538 | Ok(data) => { |
| 3539 | if ws_tx.send(data).await.is_err() { |
| 3540 | break; |
| 3541 | } |
| 3542 | } |
| 3543 | Err(broadcast::error::RecvError::Lagged(_)) => continue, |
| 3544 | Err(broadcast::error::RecvError::Closed) => break, |
| 3545 | } |
no test coverage detected