cratestack_client_rust/
streaming_callback.rs

1// -----------------------------------------------------------------------------
2// FFI / callback-shaped streaming consumer
3//
4// Synchronous from the caller's perspective: pass a callback, return
5// when the stream is done. The callback receives raw item bytes
6// (one CBOR-encoded value per call) so the FFI side can decode using
7// whatever native CBOR library it prefers.
8// -----------------------------------------------------------------------------
9
10use serde::{Deserialize, Serialize};
11
12use crate::codec::{CBOR_SEQUENCE_CONTENT_TYPE, media_type_matches};
13use crate::error::ClientError;
14use crate::runtime::wire::{RuntimeErrorCode, RuntimeErrorWire};
15use crate::streaming::CborSeqChunkDecoder;
16
17/// FFI-shaped chunk delivered to the `execute_streamed` callback.
18/// `Item` carries one CBOR-encoded item's raw bytes; `Error` is
19/// terminal; `End` is terminal and indicates a clean stream close.
20#[derive(Debug, Clone, Serialize, Deserialize)]
21pub enum RuntimeChunkWire {
22    /// One complete cbor-seq item. The bytes are CBOR-encoded — decode
23    /// on the FFI side with whatever native library the host has.
24    Item(Vec<u8>),
25    /// Terminal: the stream ended cleanly. No further chunks follow.
26    End,
27    /// Terminal: the stream failed mid-flight. No further chunks follow.
28    Error(RuntimeErrorWire),
29}
30
31/// Drive a streaming response through a callback. Used by
32/// `RuntimeHandle::execute_streamed` — the callback returns `false` to
33/// cancel the stream early. The function returns once the stream is
34/// done (by completion, error, or cancellation).
35///
36/// This path forwards raw item bytes to an FFI-side decoder rather than
37/// deserializing to a concrete `T` itself, so unlike
38/// `pump_streamed_response_typed` it has no buffered-cbor fallback to
39/// degrade to. It still checks `Content-Type` up front and fails with a
40/// clear message naming what was received instead of letting a
41/// non-cbor-seq body (e.g. negotiation picked buffered `application/cbor`)
42/// crash the frame decoder on the first chunk.
43///
44/// Its only caller, `RuntimeHandle`, is native-only (cratestack#1104).
45#[cfg_attr(target_arch = "wasm32", allow(dead_code))]
46pub(crate) async fn pump_streamed_response_callback<F>(
47    response: reqwest::Response,
48    mut on_chunk: F,
49) -> Result<(), RuntimeErrorWire>
50where
51    F: FnMut(RuntimeChunkWire) -> bool,
52{
53    use futures_util::StreamExt;
54
55    let content_type = response
56        .headers()
57        .get(reqwest::header::CONTENT_TYPE)
58        .and_then(|value| value.to_str().ok())
59        .unwrap_or_default()
60        .to_owned();
61    if !media_type_matches(&content_type, CBOR_SEQUENCE_CONTENT_TYPE) {
62        let err = RuntimeErrorWire {
63            code: RuntimeErrorCode::InvalidResponse,
64            http_status: None,
65            message: format!(
66                "streaming response had Content-Type {content_type:?}, needed \
67                 {CBOR_SEQUENCE_CONTENT_TYPE}",
68            ),
69            remote_code: None,
70            remote_body: None,
71        };
72        on_chunk(RuntimeChunkWire::Error(err.clone()));
73        return Err(err);
74    }
75
76    let mut byte_stream = response.bytes_stream();
77    let mut decoder = CborSeqChunkDecoder::new();
78    loop {
79        let chunk_result = byte_stream.next().await;
80        let chunk = match chunk_result {
81            Some(Ok(c)) => c,
82            Some(Err(error)) => {
83                let err = RuntimeErrorWire::from(ClientError::Transport(error.into()));
84                on_chunk(RuntimeChunkWire::Error(err.clone()));
85                return Err(err);
86            }
87            None => {
88                if decoder.pending_len() > 0 {
89                    let err = RuntimeErrorWire {
90                        code: RuntimeErrorCode::InvalidResponse,
91                        http_status: None,
92                        message: format!(
93                            "stream ended with {} bytes buffered (incomplete final item)",
94                            decoder.pending_len(),
95                        ),
96                        remote_code: None,
97                        remote_body: None,
98                    };
99                    on_chunk(RuntimeChunkWire::Error(err.clone()));
100                    return Err(err);
101                }
102                on_chunk(RuntimeChunkWire::End);
103                return Ok(());
104            }
105        };
106        let items = match decoder.feed_chunk(&chunk) {
107            Ok(items) => items,
108            Err(error) => {
109                let err = RuntimeErrorWire::from(ClientError::Codec(error));
110                on_chunk(RuntimeChunkWire::Error(err.clone()));
111                return Err(err);
112            }
113        };
114        for item_bytes in items {
115            if !on_chunk(RuntimeChunkWire::Item(item_bytes)) {
116                // Caller cancelled.
117                return Ok(());
118            }
119        }
120    }
121}
cratestack_client_rust/streaming_callback.rs

cratestack_client_rust/
streaming_callback.rs
