Function bridge_forward_tcp_stream

crates/openshell-server/src/grpc/sandbox.rs:1149–1215

(
    mut inbound: tonic::Streaming<TcpForwardFrame>,
    relay_stream: tokio::io::DuplexStream,
    tx: mpsc::Sender<Result<TcpForwardFrame, Status>>,
    sandbox_id: &str,
    channel_id: &str,
)

}

async fn bridge_forward_tcp_stream(
    mut inbound: tonic::Streaming<TcpForwardFrame>,
    relay_stream: tokio::io::DuplexStream,
    tx: mpsc::Sender<Result<TcpForwardFrame, Status>>,
    sandbox_id: &str,
    channel_id: &str,
) {
    let (mut relay_read, mut relay_write) = tokio::io::split(relay_stream);

    let sandbox_id_in = sandbox_id.to_string();
    let channel_id_in = channel_id.to_string();
    tokio::spawn(async move {
        loop {
            match inbound.message().await {
                Ok(Some(frame)) => {
                    let Some(openshell_core::proto::tcp_forward_frame::Payload::Data(data)) =
                        frame.payload
                    else {
                        warn!(sandbox_id = %sandbox_id_in, channel_id = %channel_id_in, "ForwardTcp: received non-data frame after init");
                        break;
                    };
                    if data.is_empty() {
                        continue;
                    }
                    if let Err(err) =
                        tokio::io::AsyncWriteExt::write_all(&mut relay_write, &data).await
                    {
                        warn!(sandbox_id = %sandbox_id_in, channel_id = %channel_id_in, error = %err, "ForwardTcp: write to relay failed");
                        break;
                    }
                }
                Ok(None) => break,
                Err(err) => {
                    debug!(sandbox_id = %sandbox_id_in, channel_id = %channel_id_in, error = %err, "ForwardTcp: inbound stream ended");
                    break;
                }
            }
        }
        let _ = tokio::io::AsyncWriteExt::shutdown(&mut relay_write).await;
    });

    let mut buf = vec![0u8; TCP_FORWARD_CHUNK_SIZE];
    loop {
        match tokio::io::AsyncReadExt::read(&mut relay_read, &mut buf).await {
            Ok(0) => break,
            Ok(n) => {
                let frame = TcpForwardFrame {
                    payload: Some(openshell_core::proto::tcp_forward_frame::Payload::Data(
                        buf[..n].to_vec(),
                    )),
                };
                if tx.send(Ok(frame)).await.is_err() {
                    break;
                }
            }
            Err(err) => {
                warn!(sandbox_id = %sandbox_id, channel_id = %channel_id, error = %err, "ForwardTcp: read from relay failed");
                let _ = tx