Method send

nodedb-client/src/native/connection/mod.rs:354–423

Send a request and read the response.

(
        &mut self,
        op: OpCode,
        fields: TextFields,
    )

    /// Send a request and read the response.
    pub(crate) async fn send(
        &mut self,
        op: OpCode,
        fields: TextFields,
    ) -> NodeDbResult<NativeResponse> {
        let req = NativeRequest {
            op,
            seq: self.next_seq(),
            fields: RequestFields::Text(fields),
        };

        let payload = zerompk::to_msgpack_vec(&req)
            .map_err(|e| NodeDbError::serialization("msgpack", format!("request encode: {e}")))?;

        let len = payload.len() as u32;
        self.stream
            .write_all(&len.to_be_bytes())
            .await
            .map_err(io_err)?;
        self.stream.write_all(&payload).await.map_err(io_err)?;
        self.stream.flush().await.map_err(io_err)?;

        let mut combined_rows: Vec<Vec<nodedb_types::Value>> = Vec::new();
        let mut final_resp: Option<NativeResponse> = None;

        loop {
            let mut len_buf = [0u8; FRAME_HEADER_LEN];
            self.stream.read_exact(&mut len_buf).await.map_err(io_err)?;
            let resp_len = u32::from_be_bytes(len_buf);
            if resp_len > MAX_FRAME_SIZE {
                return Err(NodeDbError::internal(format!(
                    "response frame too large: {resp_len}"
                )));
            }

            let mut resp_buf = vec![0u8; resp_len as usize];
            self.stream
                .read_exact(&mut resp_buf)
                .await
                .map_err(io_err)?;

            let resp: NativeResponse = zerompk::from_msgpack(&resp_buf).map_err(|e| {
                NodeDbError::serialization("msgpack", format!("response decode: {e}"))
            })?;

            if resp.status == ResponseStatus::Partial {
                if let Some(rows) = resp.rows {
                    combined_rows.extend(rows);
                }
                if final_resp.is_none() {
                    final_resp = Some(NativeResponse { rows: None, ..resp });
                }
            } else {
                if combined_rows.is_empty() {
                    final_resp = Some(resp);
                } else {
                    if let Some(ref rows) = resp.rows {
                        combined_rows.extend(rows.iter().cloned());