diff --git a/rayforce-sys/src/lib.rs b/rayforce-sys/src/lib.rs index 79d7fd7..e6d6918 100644 --- a/rayforce-sys/src/lib.rs +++ b/rayforce-sys/src/lib.rs @@ -38,6 +38,17 @@ extern "C" { err: *mut ::std::os::raw::c_char, errlen: usize, ) -> *mut ray_t; + /// Decompress (if needed) and deserialize a response *body* (wire header + /// already stripped) into a freshly-owned object — possibly a `RAY_ERROR` + /// carrying a Q server-side error. Touches the symbol table: engine thread + /// only. Returns null on a decode failure with a short reason in `err`. + pub fn q_decode( + resp: *mut u8, + resp_len: i64, + compressed: ::std::os::raw::c_int, + err: *mut ::std::os::raw::c_char, + errlen: usize, + ) -> *mut ray_t; } /// `q_connect` failure codes (mirrors `q.h`). diff --git a/rayforce/src/q.rs b/rayforce/src/q.rs index 470c1e5..0809749 100644 --- a/rayforce/src/q.rs +++ b/rayforce/src/q.rs @@ -94,3 +94,53 @@ impl Drop for QConnection { unsafe { sys::q_close(self.fd) }; } } + +/// Decode a complete Q response message (8-byte wire header + body) that was +/// received by an **external transport** — e.g. a worker thread that owns a +/// plain `std::net::TcpStream` with read timeouts — into a [`Value`]. +/// +/// This is the engine-thread half of a split client: socket I/O is plain +/// bytes and thread-safe anywhere, while this call allocates engine objects +/// and must run on the thread that owns the [`crate::Runtime`] (like every +/// other constructor in this crate). A Q server-side error surfaces as `Err`. +pub fn decode_response(msg: &[u8]) -> Result { + // q_header_t (q.c): endianness, msgtype, compressed, reserved, u32 size. + // `size` counts the whole message, header included. Little-endian wire only. + const HEADER_LEN: usize = 8; + if msg.len() < HEADER_LEN { + return Err(RayError::binding("Q: message shorter than wire header")); + } + let size = u32::from_le_bytes([msg[4], msg[5], msg[6], msg[7]]) as usize; + if size != msg.len() { + return Err(RayError::binding( + "Q: message length does not match wire header", + )); + } + let compressed = i32::from(msg[2] != 0); + let body = &msg[HEADER_LEN..]; + if body.is_empty() { + return Err(RayError::binding("Q: empty response body")); + } + + let mut err = [0i8; 256]; + let res = unsafe { + sys::q_decode( + body.as_ptr() as *mut u8, + body.len() as i64, + compressed, + err.as_mut_ptr(), + err.len(), + ) + }; + if res.is_null() { + let msg = unsafe { std::ffi::CStr::from_ptr(err.as_ptr()) }.to_string_lossy(); + let msg = if msg.is_empty() { + "q_decode failed".into() + } else { + msg + }; + return Err(RayError::binding(format!("Q: {msg}"))); + } + let ok = unsafe { check(res)? }; + Ok(unsafe { Value::from_owned(ok) }) +}