oracledb_protocol/thin/
aq.rs

1#![forbid(unsafe_code)]
2
3//! Sans-io codecs for Oracle Advanced Queuing (AQ) enqueue/dequeue operations.
4//!
5//! Mirrors the reference thin driver's `impl/thin/messages/aq_*.pyx`:
6//! - [`build_aq_enq_payload`] / [`parse_aq_enq_response`]  — FUNC 121 (single enqueue)
7//! - [`build_aq_deq_payload`] / [`parse_aq_deq_response`]  — FUNC 122 (single dequeue)
8//! - [`build_aq_array_enq_payload`] / [`build_aq_array_deq_payload`]
9//!   / [`parse_aq_array_response`]                          — FUNC 145 (bulk enqueue/dequeue)
10//!
11//! The message-property / payload codecs are shared between all three so the
12//! wire encoding is byte-identical to python-oracledb (golden traces under
13//! `tests/golden/aq_*.txt`). Object payloads reuse the `dbobject` codec and JSON
14//! payloads reuse [`crate::oson`].
15
16use super::*;
17use crate::oson::{decode_oson_with_limits, encode_oson, OsonValue};
18use crate::wire::ProtocolLimits;
19
20/// Payload classification for a queue. Determines the TOID sentinel and the
21/// payload-encoding branch taken during enqueue/dequeue.
22#[derive(Clone, Debug, PartialEq, Eq)]
23pub enum AqPayloadKind {
24    /// RAW / bytes payload. TOID sentinel `bytes([0]*15 + [0x17])`.
25    Raw,
26    /// JSON payload (OSON). TOID sentinel `bytes([0]*15 + [0x47])`.
27    Json,
28    /// Object payload of a named type. TOID is the type's OID.
29    Object,
30}
31
32/// Static description of an AQ queue, used by both enqueue and dequeue.
33#[derive(Clone, Debug)]
34pub struct AqQueueDesc {
35    pub name: String,
36    pub kind: AqPayloadKind,
37    /// TOID of the payload: 16-byte sentinel for RAW/JSON, the type OID for
38    /// object queues.
39    pub payload_toid: Vec<u8>,
40}
41
42impl AqQueueDesc {
43    /// Builds the queue descriptor, deriving the payload TOID from the kind.
44    /// For object queues `object_oid` must carry the type's OID.
45    pub fn new(name: String, kind: AqPayloadKind, object_oid: Option<Vec<u8>>) -> Self {
46        let payload_toid = match kind {
47            AqPayloadKind::Raw => raw_payload_toid(),
48            AqPayloadKind::Json => json_payload_toid(),
49            AqPayloadKind::Object => object_oid.unwrap_or_default(),
50        };
51        Self {
52            name,
53            kind,
54            payload_toid,
55        }
56    }
57}
58
59fn raw_payload_toid() -> Vec<u8> {
60    let mut toid = vec![0u8; 15];
61    toid.push(0x17);
62    toid
63}
64
65fn json_payload_toid() -> Vec<u8> {
66    let mut toid = vec![0u8; 15];
67    toid.push(0x47);
68    toid
69}
70
71/// The payload value carried by a message being enqueued.
72#[derive(Clone, Debug)]
73pub enum AqPayloadValue {
74    /// RAW bytes (already UTF-8 encoded if originally a string).
75    Raw(Vec<u8>),
76    /// JSON value encoded as OSON.
77    Json(OsonValue),
78    /// Pre-packed object image (the body produced by `DbObjectImpl::pack_image`).
79    Object { oid: Vec<u8>, image: Vec<u8> },
80}
81
82/// Mutable message properties (reference `ThinMsgPropsImpl`). Defaults match
83/// the reference: `delay=0`, `expiration=-1`, `priority=0`, `state=0`.
84#[derive(Clone, Debug)]
85pub struct AqMsgProps {
86    pub priority: i32,
87    pub delay: i32,
88    pub expiration: i32,
89    pub correlation: Option<String>,
90    pub exception_queue: Option<String>,
91    pub state: i32,
92    pub enq_txn_id: Option<Vec<u8>>,
93    /// Recipient names (multi-consumer enqueue). `None` => no recipient list.
94    pub recipients: Option<Vec<String>>,
95    /// Payload to enqueue. Required for enqueue; ignored for dequeue requests
96    /// (where defaults are written).
97    pub payload: Option<AqPayloadValue>,
98}
99
100impl Default for AqMsgProps {
101    fn default() -> Self {
102        Self {
103            priority: 0,
104            delay: 0,
105            expiration: -1,
106            correlation: None,
107            exception_queue: None,
108            state: 0,
109            enq_txn_id: None,
110            // Reference `ThinMsgPropsImpl.__init__` defaults recipients to an
111            // empty list (not None): an empty list still writes pointer=1 with
112            // a zero count, whereas None writes pointer=0.
113            recipients: Some(Vec::new()),
114            payload: None,
115        }
116    }
117}
118
119/// Enqueue options (reference `ThinEnqOptionsImpl`). `visibility` defaults to
120/// `ENQ_ON_COMMIT (2)`, `delivery_mode` to `PERSISTENT (1)`.
121#[derive(Clone, Debug)]
122pub struct AqEnqOptions {
123    pub visibility: u32,
124    pub delivery_mode: u16,
125}
126
127impl Default for AqEnqOptions {
128    fn default() -> Self {
129        Self {
130            visibility: 2,
131            delivery_mode: TNS_AQ_MSG_PERSISTENT,
132        }
133    }
134}
135
136/// Dequeue options (reference `ThinDeqOptionsImpl`). Defaults: `mode=REMOVE(3)`,
137/// `navigation=NEXT_MSG(3)`, `visibility=ON_COMMIT(2)`, `wait=WAIT_FOREVER`,
138/// `delivery_mode=PERSISTENT(1)`.
139#[derive(Clone, Debug)]
140pub struct AqDeqOptions {
141    pub condition: Option<String>,
142    pub consumer_name: Option<String>,
143    pub correlation: Option<String>,
144    pub delivery_mode: u16,
145    pub mode: i32,
146    pub msgid: Option<Vec<u8>>,
147    pub navigation: i32,
148    pub visibility: i32,
149    pub wait: u32,
150}
151
152impl Default for AqDeqOptions {
153    fn default() -> Self {
154        Self {
155            condition: None,
156            consumer_name: None,
157            correlation: None,
158            delivery_mode: TNS_AQ_MSG_PERSISTENT,
159            mode: 3,
160            msgid: None,
161            navigation: 3,
162            visibility: 2,
163            wait: 0xFFFF_FFFF,
164        }
165    }
166}
167
168/// A message returned by dequeue (reference fields read by `_process_msg_props`
169/// / `_process_payload`).
170#[derive(Clone, Debug, Default)]
171pub struct AqDeqMessage {
172    pub priority: i32,
173    pub delay: i32,
174    pub expiration: i32,
175    pub correlation: Option<String>,
176    pub num_attempts: i32,
177    pub exception_queue: Option<String>,
178    pub state: i32,
179    /// Oracle enqueue time decoded to a naive datetime, or `None`.
180    pub enq_time: Option<QueryValue>,
181    pub delivery_mode: u16,
182    pub msgid: Option<Vec<u8>>,
183    /// Decoded payload. `None` for an empty-payload message.
184    pub payload: Option<AqDeqPayload>,
185}
186
187/// A decoded dequeue payload.
188#[derive(Clone, Debug)]
189pub enum AqDeqPayload {
190    /// RAW bytes (may be empty for `DEQ_REMOVE_NODATA`).
191    Raw(Vec<u8>),
192    /// JSON decoded from OSON.
193    Json(OsonValue),
194    /// Object payload: the raw packed image (unpacked by the shim against the
195    /// queue's payload type).
196    Object(Vec<u8>),
197}
198
199// ---------------------------------------------------------------------------
200// Shared message-property and payload codecs (reference aq_base.pyx).
201// ---------------------------------------------------------------------------
202
203/// Writes the TTC function-code preamble: message-type/function/seq plus the
204/// `token_num` ub8 the server expects when the negotiated field version is at
205/// least `23.1 EXT 1` (reference `Message._write_function_code`).
206fn write_aq_function_code(
207    writer: &mut TtcWriter,
208    function_code: u8,
209    seq_num: u8,
210    ttc_field_version: u8,
211) {
212    writer.write_function_header(function_code, seq_num, ttc_field_version);
213}
214
215fn write_value_with_length(writer: &mut TtcWriter, value: Option<&[u8]>) -> Result<()> {
216    match value {
217        None => {
218            writer.write_ub4(0);
219            Ok(())
220        }
221        Some(bytes) => writer.write_bytes_with_two_lengths(Some(bytes)),
222    }
223}
224
225/// Writes the AQ message-property block (`_write_msg_props`).
226fn write_msg_props(
227    writer: &mut TtcWriter,
228    props: &AqMsgProps,
229    ttc_field_version: u8,
230) -> Result<()> {
231    writer.write_ub4(props.priority as u32);
232    writer.write_ub4(props.delay as u32);
233    writer.write_sb4(props.expiration);
234    write_value_with_length(writer, props.correlation.as_deref().map(str::as_bytes))?;
235    writer.write_ub4(0); // number of attempts
236    write_value_with_length(writer, props.exception_queue.as_deref().map(str::as_bytes))?;
237    writer.write_ub4(props.state as u32);
238    writer.write_ub4(0); // enqueue time length
239    write_value_with_length(writer, props.enq_txn_id.as_deref())?;
240    writer.write_ub4(4); // number of extensions
241    writer.write_u8(0x0e); // unknown extra byte
242    writer.write_keyword_value_pair(None, None, TNS_AQ_EXT_KEYWORD_AGENT_NAME)?;
243    writer.write_keyword_value_pair(None, None, TNS_AQ_EXT_KEYWORD_AGENT_ADDRESS)?;
244    writer.write_keyword_value_pair(None, Some(b"\x00"), TNS_AQ_EXT_KEYWORD_AGENT_PROTOCOL)?;
245    writer.write_keyword_value_pair(None, None, TNS_AQ_EXT_KEYWORD_ORIGINAL_MSGID)?;
246    writer.write_ub4(0); // user property
247    writer.write_ub4(0); // cscn
248    writer.write_ub4(0); // dscn
249    writer.write_ub4(0); // flags
250    if version_gates::carries_aq_shard_id(ttc_field_version) {
251        writer.write_ub4(0xFFFF_FFFF); // shard id
252    }
253    Ok(())
254}
255
256/// Writes the recipient-list key/value pairs (`_write_recipients`).
257fn write_recipients(writer: &mut TtcWriter, recipients: &[String]) -> Result<()> {
258    let mut index: u16 = 0;
259    for recipient in recipients {
260        writer.write_keyword_value_pair(Some(recipient.as_bytes()), None, index)?;
261        writer.write_keyword_value_pair(None, None, index + 1)?;
262        writer.write_keyword_value_pair(None, Some(b"\x00"), index + 2)?;
263        index += 3;
264    }
265    Ok(())
266}
267
268/// Writes the message payload (`_write_payload`).
269fn write_payload(
270    writer: &mut TtcWriter,
271    payload: &AqPayloadValue,
272    supports_oson_long_fnames: bool,
273) -> Result<()> {
274    match payload {
275        AqPayloadValue::Json(value) => {
276            // write_oson(..., write_length=False): a QLocator (no chunk-length
277            // prefix) followed by the OSON image as a length-prefixed chunk.
278            let image = encode_oson(value, supports_oson_long_fnames)?;
279            crate::vector::write_oson_aq_payload(writer, &image)
280        }
281        AqPayloadValue::Object { oid, image } => write_dbobject_bind(writer, oid, image),
282        AqPayloadValue::Raw(bytes) => {
283            writer.write_raw(bytes);
284            Ok(())
285        }
286    }
287}
288
289// ---------------------------------------------------------------------------
290// FUNC 121 — single enqueue
291// ---------------------------------------------------------------------------
292
293/// Builds the AQ enqueue (FUNC 121) request payload (reference
294/// `AqEnqMessage._write_message`).
295pub fn build_aq_enq_payload(
296    queue: &AqQueueDesc,
297    props: &AqMsgProps,
298    enq_options: &AqEnqOptions,
299    seq_num: u8,
300    ttc_field_version: u8,
301    supports_oson_long_fnames: bool,
302) -> Result<Vec<u8>> {
303    let payload = props
304        .payload
305        .as_ref()
306        .ok_or(ProtocolError::TtcDecode("AQ enqueue has no payload"))?;
307    let queue_name = queue.name.as_bytes();
308    let mut writer = TtcWriter::new();
309    write_aq_function_code(&mut writer, TNS_FUNC_AQ_ENQ, seq_num, ttc_field_version);
310    writer.write_u8(1); // queue name (pointer)
311    writer.write_ub4(queue_name.len() as u32);
312    write_msg_props(&mut writer, props, ttc_field_version)?;
313    match props.recipients.as_ref() {
314        None => {
315            writer.write_u8(0); // recipients (pointer)
316            writer.write_ub4(0); // number of key/value pairs
317        }
318        Some(recipients) => {
319            writer.write_u8(1);
320            writer.write_ub4(3 * recipients.len() as u32);
321        }
322    }
323    writer.write_ub4(enq_options.visibility);
324    writer.write_u8(0); // relative message id (pointer)
325    writer.write_ub4(0); // relative message length
326    writer.write_ub4(0); // sequence deviation
327    writer.write_u8(1); // TOID of payload (pointer)
328    writer.write_ub4(16); // TOID of payload length
329    writer.write_ub2(TNS_AQ_MESSAGE_VERSION);
330    match queue.kind {
331        AqPayloadKind::Json => {
332            writer.write_u8(0); // payload (pointer)
333            writer.write_u8(0); // RAW payload (pointer)
334            writer.write_ub4(0); // RAW payload length
335        }
336        AqPayloadKind::Object => {
337            writer.write_u8(1); // payload (pointer)
338            writer.write_u8(0); // RAW payload (pointer)
339            writer.write_ub4(0); // RAW payload length
340        }
341        AqPayloadKind::Raw => {
342            let raw_len = match payload {
343                AqPayloadValue::Raw(bytes) => bytes.len() as u32,
344                _ => return Err(ProtocolError::TtcDecode("RAW queue requires RAW payload")),
345            };
346            writer.write_u8(0); // payload (pointer)
347            writer.write_u8(1); // RAW payload (pointer)
348            writer.write_ub4(raw_len);
349        }
350    }
351    writer.write_u8(1); // return message id (pointer)
352    writer.write_ub4(TNS_AQ_MESSAGE_ID_LENGTH as u32);
353    let mut enq_flags = 0u32;
354    if enq_options.delivery_mode == TNS_AQ_MSG_BUFFERED {
355        enq_flags |= TNS_KPD_AQ_BUFMSG;
356    }
357    writer.write_ub4(enq_flags); // enqueue flags
358    writer.write_u8(0); // extensions 1 (pointer)
359    writer.write_ub4(0); // number of extensions 1
360    writer.write_u8(0); // extensions 2 (pointer)
361    writer.write_ub4(0); // number of extensions 2
362    writer.write_u8(0); // source sequence number
363    writer.write_ub4(0); // source sequence length
364    writer.write_u8(0); // max sequence number
365    writer.write_ub4(0); // max sequence length
366    writer.write_u8(0); // output ack length
367    writer.write_u8(0); // correlation (pointer)
368    writer.write_ub4(0); // correlation length
369    writer.write_u8(0); // sender name (pointer)
370    writer.write_ub4(0); // sender name length
371    writer.write_u8(0); // sender address (pointer)
372    writer.write_ub4(0); // sender address length
373    writer.write_u8(0); // sender charset id (pointer)
374    writer.write_u8(0); // sender ncharset id (pointer)
375    if version_gates::writes_aq_json_payload(ttc_field_version) {
376        // JSON payload (pointer)
377        writer.write_u8(u8::from(queue.kind == AqPayloadKind::Json));
378    }
379
380    writer.write_bytes_with_length(queue_name)?;
381    if let Some(recipients) = props.recipients.as_ref() {
382        write_recipients(&mut writer, recipients)?;
383    }
384    writer.write_raw(&queue.payload_toid);
385    write_payload(&mut writer, payload, supports_oson_long_fnames)?;
386    Ok(writer.into_bytes())
387}
388
389/// Parses an AQ enqueue (FUNC 121) response, returning the assigned 16-byte
390/// message id (reference `AqEnqMessage._process_return_parameters`).
391pub fn parse_aq_enq_response(
392    payload: &[u8],
393    capabilities: ClientCapabilities,
394) -> Result<Option<Vec<u8>>> {
395    parse_aq_enq_response_with_limits(payload, capabilities, ProtocolLimits::DEFAULT)
396}
397
398pub fn parse_aq_enq_response_with_limits(
399    payload: &[u8],
400    capabilities: ClientCapabilities,
401    limits: ProtocolLimits,
402) -> Result<Option<Vec<u8>>> {
403    let mut reader = TtcReader::with_limits(payload, limits)?;
404    let mut msgid: Option<Vec<u8>> = None;
405    while reader.remaining() > 0 {
406        let message_type = reader.read_u8()?;
407        match message_type {
408            0 => {}
409            TNS_MSG_TYPE_PARAMETER => {
410                let id = reader.read_raw(TNS_AQ_MESSAGE_ID_LENGTH)?.to_vec();
411                let _ext_len = reader.read_ub2()?;
412                msgid = Some(id);
413            }
414            TNS_MSG_TYPE_STATUS => {
415                let _call_status = reader.read_ub4()?;
416                let _seq = reader.read_ub2()?;
417            }
418            TNS_MSG_TYPE_SERVER_SIDE_PIGGYBACK => {
419                let _ = skip_server_side_piggyback(&mut reader)?;
420            }
421            TNS_MSG_TYPE_END_OF_RESPONSE => break,
422            TNS_MSG_TYPE_ERROR => {
423                let info = parse_server_error_info(&mut reader, capabilities.ttc_field_version)?;
424                if info.number != 0 {
425                    return Err(ProtocolError::ServerErrorInfo(Box::new(
426                        info.into_details(),
427                    )));
428                }
429            }
430            _ => {
431                return Err(ProtocolError::UnknownMessageType {
432                    message_type,
433                    position: reader.position().saturating_sub(1),
434                })
435            }
436        }
437    }
438    Ok(msgid)
439}
440
441// ---------------------------------------------------------------------------
442// FUNC 122 — single dequeue
443// ---------------------------------------------------------------------------
444
445/// Builds the AQ dequeue (FUNC 122) request payload (reference
446/// `AqDeqMessage._write_message`).
447pub fn build_aq_deq_payload(
448    queue: &AqQueueDesc,
449    deq_options: &AqDeqOptions,
450    seq_num: u8,
451    ttc_field_version: u8,
452) -> Result<Vec<u8>> {
453    let queue_name = queue.name.as_bytes();
454    let mut writer = TtcWriter::new();
455    write_aq_function_code(&mut writer, TNS_FUNC_AQ_DEQ, seq_num, ttc_field_version);
456    writer.write_u8(1); // queue name (pointer)
457    writer.write_ub4(queue_name.len() as u32);
458    writer.write_u8(1); // message properties
459    writer.write_u8(1); // msg props length
460    writer.write_u8(1); // recipient list
461    writer.write_u8(1); // recipient list length
462    let consumer_name = deq_options
463        .consumer_name
464        .as_ref()
465        .filter(|name| !name.is_empty());
466    match consumer_name {
467        Some(name) => {
468            writer.write_u8(1);
469            writer.write_ub4(name.len() as u32);
470        }
471        None => {
472            writer.write_u8(0);
473            writer.write_ub4(0);
474        }
475    }
476    writer.write_sb4(deq_options.mode);
477    writer.write_sb4(deq_options.navigation);
478    writer.write_sb4(deq_options.visibility);
479    writer.write_sb4(deq_options.wait as i32);
480    let msgid = deq_options.msgid.as_ref().filter(|id| !id.is_empty());
481    match msgid {
482        Some(_) => {
483            writer.write_u8(1);
484            writer.write_ub4(TNS_AQ_MESSAGE_ID_LENGTH as u32);
485        }
486        None => {
487            writer.write_u8(0);
488            writer.write_ub4(0);
489        }
490    }
491    let correlation = deq_options.correlation.as_ref().filter(|c| !c.is_empty());
492    match correlation {
493        Some(c) => {
494            writer.write_u8(1);
495            writer.write_ub4(c.len() as u32);
496        }
497        None => {
498            writer.write_u8(0);
499            writer.write_ub4(0);
500        }
501    }
502    writer.write_u8(1); // toid of payload
503    writer.write_ub4(16); // toid length
504    writer.write_ub2(TNS_AQ_MESSAGE_VERSION);
505    writer.write_u8(1); // payload
506    writer.write_u8(1); // return msg id
507    writer.write_ub4(TNS_AQ_MESSAGE_ID_LENGTH as u32);
508    let mut deq_flags = 0u32;
509    match deq_options.delivery_mode {
510        TNS_AQ_MSG_BUFFERED => deq_flags |= TNS_KPD_AQ_BUFMSG,
511        TNS_AQ_MSG_PERSISTENT_OR_BUFFERED => deq_flags |= TNS_KPD_AQ_EITHER,
512        _ => {}
513    }
514    writer.write_ub4(deq_flags);
515    let condition = deq_options.condition.as_ref().filter(|c| !c.is_empty());
516    match condition {
517        Some(c) => {
518            writer.write_u8(1);
519            writer.write_ub4(c.len() as u32);
520        }
521        None => {
522            writer.write_u8(0);
523            writer.write_ub4(0);
524        }
525    }
526    writer.write_u8(0); // extensions
527    writer.write_ub4(0); // number of extensions
528    if version_gates::writes_aq_json_payload(ttc_field_version) {
529        writer.write_u8(0); // JSON payload
530    }
531    if version_gates::carries_aq_shard_id(ttc_field_version) {
532        writer.write_ub4(0xFFFF_FFFF); // shard id (-1)
533    }
534
535    writer.write_bytes_with_length(queue_name)?;
536    if let Some(name) = consumer_name {
537        writer.write_bytes_with_length(name.as_bytes())?;
538    }
539    if let Some(id) = msgid {
540        let mut id = id.clone();
541        id.truncate(16);
542        if id.len() < 16 {
543            id.resize(16, 0);
544        }
545        writer.write_raw(&id);
546    }
547    if let Some(c) = correlation {
548        writer.write_bytes_with_length(c.as_bytes())?;
549    }
550    writer.write_raw(&queue.payload_toid);
551    if let Some(c) = condition {
552        writer.write_bytes_with_length(c.as_bytes())?;
553    }
554    Ok(writer.into_bytes())
555}
556
557/// Outcome of a single dequeue.
558#[derive(Clone, Debug, Default)]
559pub struct AqDeqResult {
560    /// The dequeued message, or `None` when the queue was empty (ORA-25228).
561    pub message: Option<AqDeqMessage>,
562}
563
564/// Parses an AQ dequeue (FUNC 122) response (reference
565/// `AqDeqMessage._process_return_parameters`).
566pub fn parse_aq_deq_response(
567    payload: &[u8],
568    capabilities: ClientCapabilities,
569    kind: &AqPayloadKind,
570) -> Result<AqDeqResult> {
571    parse_aq_deq_response_with_limits(payload, capabilities, kind, ProtocolLimits::DEFAULT)
572}
573
574pub fn parse_aq_deq_response_with_limits(
575    payload: &[u8],
576    capabilities: ClientCapabilities,
577    kind: &AqPayloadKind,
578    limits: ProtocolLimits,
579) -> Result<AqDeqResult> {
580    let mut reader = TtcReader::with_limits(payload, limits)?;
581    let mut result = AqDeqResult::default();
582    let mut no_msg_found = false;
583    while reader.remaining() > 0 {
584        let message_type = reader.read_u8()?;
585        match message_type {
586            0 => {}
587            TNS_MSG_TYPE_PARAMETER => {
588                let num_bytes = reader.read_ub4()?;
589                if num_bytes > 0 {
590                    let mut message = AqDeqMessage::default();
591                    process_msg_props(&mut reader, &mut message, capabilities.ttc_field_version)?;
592                    process_recipients(&mut reader)?;
593                    message.payload = process_payload(&mut reader, kind)?;
594                    message.msgid = Some(process_msg_id(&mut reader)?);
595                    result.message = Some(message);
596                }
597            }
598            TNS_MSG_TYPE_STATUS => {
599                let _call_status = reader.read_ub4()?;
600                let _seq = reader.read_ub2()?;
601            }
602            TNS_MSG_TYPE_SERVER_SIDE_PIGGYBACK => {
603                let _ = skip_server_side_piggyback(&mut reader)?;
604            }
605            TNS_MSG_TYPE_END_OF_RESPONSE => break,
606            TNS_MSG_TYPE_ERROR => {
607                let info = parse_server_error_info(&mut reader, capabilities.ttc_field_version)?;
608                if info.number == TNS_ERR_NO_MESSAGES_FOUND as u32 {
609                    no_msg_found = true;
610                } else if info.number != 0 {
611                    return Err(ProtocolError::ServerErrorInfo(Box::new(
612                        info.into_details(),
613                    )));
614                }
615            }
616            _ => {
617                return Err(ProtocolError::UnknownMessageType {
618                    message_type,
619                    position: reader.position().saturating_sub(1),
620                })
621            }
622        }
623    }
624    if no_msg_found {
625        result.message = None;
626    }
627    Ok(result)
628}
629
630// ---------------------------------------------------------------------------
631// FUNC 145 — bulk enqueue / dequeue
632// ---------------------------------------------------------------------------
633
634/// Builds the AQ array enqueue (FUNC 145, op=ENQ) request payload (reference
635/// `AqArrayMessage._write_message` + `_write_array_enq`).
636pub fn build_aq_array_enq_payload(
637    queue: &AqQueueDesc,
638    props_list: &[AqMsgProps],
639    enq_options: &AqEnqOptions,
640    seq_num: u8,
641    ttc_field_version: u8,
642    supports_oson_long_fnames: bool,
643) -> Result<Vec<u8>> {
644    let num_iters = props_list.len() as u32;
645    let queue_name = queue.name.as_bytes();
646    let mut writer = TtcWriter::new();
647    write_aq_function_code(&mut writer, TNS_FUNC_AQ_ARRAY, seq_num, ttc_field_version);
648    writer.write_u8(0); // input params (pointer)
649    writer.write_ub4(0); // length
650    writer.write_ub4(TNS_AQ_ARRAY_FLAGS_RETURN_MESSAGE_ID);
651    writer.write_u8(1); // output params (pointer)
652    writer.write_u8(0); // length
653    writer.write_sb4(TNS_AQ_ARRAY_ENQ);
654    writer.write_u8(1); // num iters (pointer)
655    if version_gates::carries_aq_shard_id(ttc_field_version) {
656        writer.write_ub4(0xFFFF); // shard id
657    }
658    writer.write_ub4(num_iters);
659
660    let mut flags = 0u32;
661    if enq_options.delivery_mode == TNS_AQ_MSG_BUFFERED {
662        flags |= TNS_KPD_AQ_BUFMSG;
663    }
664    writer.write_ub4(0); // rel msgid len
665    writer.write_u8(TNS_MSG_TYPE_ROW_HEADER);
666    writer.write_bytes_with_two_lengths(Some(queue_name))?;
667    writer.write_raw(&queue.payload_toid);
668    writer.write_ub2(TNS_AQ_MESSAGE_VERSION);
669    writer.write_ub4(flags);
670    for props in props_list {
671        let payload = props
672            .payload
673            .as_ref()
674            .ok_or(ProtocolError::TtcDecode("AQ array enqueue has no payload"))?;
675        writer.write_u8(TNS_MSG_TYPE_ROW_DATA);
676        writer.write_ub4(flags); // aqi flags
677        write_msg_props(&mut writer, props, ttc_field_version)?;
678        match props.recipients.as_ref() {
679            None => writer.write_ub4(0),
680            Some(recipients) => {
681                writer.write_ub4(3 * recipients.len() as u32);
682                write_recipients(&mut writer, recipients)?;
683            }
684        }
685        writer.write_sb4(enq_options.visibility as i32);
686        writer.write_ub4(0); // relative msg id
687        writer.write_sb4(0); // seq deviation
688        if matches!(queue.kind, AqPayloadKind::Raw) {
689            let raw_len = match payload {
690                AqPayloadValue::Raw(bytes) => bytes.len() as u32,
691                _ => return Err(ProtocolError::TtcDecode("RAW queue requires RAW payload")),
692            };
693            writer.write_ub4(raw_len);
694        }
695        write_payload(&mut writer, payload, supports_oson_long_fnames)?;
696    }
697    writer.write_u8(TNS_MSG_TYPE_STATUS);
698    Ok(writer.into_bytes())
699}
700
701/// Builds the AQ array dequeue (FUNC 145, op=DEQ) request payload (reference
702/// `AqArrayMessage._write_message` + `_write_array_deq`).
703pub fn build_aq_array_deq_payload(
704    queue: &AqQueueDesc,
705    deq_options: &AqDeqOptions,
706    num_iters: u32,
707    seq_num: u8,
708    ttc_field_version: u8,
709) -> Result<Vec<u8>> {
710    let queue_name = queue.name.as_bytes();
711    let mut writer = TtcWriter::new();
712    write_aq_function_code(&mut writer, TNS_FUNC_AQ_ARRAY, seq_num, ttc_field_version);
713    writer.write_u8(1); // input params (pointer)
714    writer.write_ub4(num_iters);
715    writer.write_ub4(TNS_AQ_ARRAY_FLAGS_RETURN_MESSAGE_ID);
716    writer.write_u8(1); // output params (pointer)
717    writer.write_u8(1); // length
718    writer.write_sb4(TNS_AQ_ARRAY_DEQ);
719    writer.write_u8(0); // num iters (pointer)
720    if version_gates::carries_aq_shard_id(ttc_field_version) {
721        writer.write_ub4(0xFFFF); // shard id
722    }
723
724    let mut flags = 0u32;
725    match deq_options.delivery_mode {
726        TNS_AQ_MSG_BUFFERED => flags |= TNS_KPD_AQ_BUFMSG,
727        TNS_AQ_MSG_PERSISTENT_OR_BUFFERED => flags |= TNS_KPD_AQ_EITHER,
728        _ => {}
729    }
730    let consumer_name = deq_options
731        .consumer_name
732        .as_ref()
733        .filter(|name| !name.is_empty())
734        .map(|name| name.as_bytes());
735    let correlation = deq_options
736        .correlation
737        .as_ref()
738        .filter(|c| !c.is_empty())
739        .map(|c| c.as_bytes());
740    let condition = deq_options
741        .condition
742        .as_ref()
743        .filter(|c| !c.is_empty())
744        .map(|c| c.as_bytes());
745    let props = AqMsgProps::default();
746    for _ in 0..num_iters {
747        writer.write_bytes_with_two_lengths(Some(queue_name))?;
748        write_msg_props(&mut writer, &props, ttc_field_version)?;
749        writer.write_ub4(0); // num recipients
750        write_value_with_length(&mut writer, consumer_name)?;
751        writer.write_sb4(deq_options.mode);
752        writer.write_sb4(deq_options.navigation);
753        writer.write_sb4(deq_options.visibility);
754        writer.write_sb4(deq_options.wait as i32);
755        write_value_with_length(&mut writer, deq_options.msgid.as_deref())?;
756        write_value_with_length(&mut writer, correlation)?;
757        write_value_with_length(&mut writer, condition)?;
758        writer.write_ub4(0); // extensions
759        writer.write_ub4(0); // rel msg id
760        writer.write_sb4(0); // seq deviation
761        writer.write_bytes_with_two_lengths(Some(&queue.payload_toid))?;
762        writer.write_ub2(TNS_AQ_MESSAGE_VERSION);
763        writer.write_ub4(0); // payload length
764        writer.write_ub4(0); // raw pay length
765        writer.write_ub4(0);
766        writer.write_ub4(flags);
767        writer.write_ub4(0); // extensions len
768        writer.write_ub4(0); // source seq len
769    }
770    Ok(writer.into_bytes())
771}
772
773/// Result of a bulk operation: enqueue returns assigned msgids per message;
774/// dequeue returns the dequeued messages (already truncated to `num_iters`).
775#[derive(Clone, Debug, Default)]
776pub struct AqArrayResult {
777    /// For enqueue: assigned msgid per input message, in order.
778    pub enq_msgids: Vec<Vec<u8>>,
779    /// For dequeue: the dequeued messages.
780    pub deq_messages: Vec<AqDeqMessage>,
781}
782
783/// Parses an AQ array (FUNC 145) response for either operation (reference
784/// `AqArrayMessage._process_return_parameters`).
785///
786/// `props_count` is the number of message-property slots prepared client-side
787/// (`num_iters` for enqueue, `max_num_messages` for dequeue).
788pub fn parse_aq_array_response(
789    payload: &[u8],
790    capabilities: ClientCapabilities,
791    operation: i32,
792    props_count: u32,
793    kind: &AqPayloadKind,
794) -> Result<AqArrayResult> {
795    parse_aq_array_response_with_limits(
796        payload,
797        capabilities,
798        operation,
799        props_count,
800        kind,
801        ProtocolLimits::DEFAULT,
802    )
803}
804
805pub fn parse_aq_array_response_with_limits(
806    payload: &[u8],
807    capabilities: ClientCapabilities,
808    operation: i32,
809    props_count: u32,
810    kind: &AqPayloadKind,
811    limits: ProtocolLimits,
812) -> Result<AqArrayResult> {
813    limits.check_batch_rows(props_count as usize)?;
814    let mut reader = TtcReader::with_limits(payload, limits)?;
815    let mut result = AqArrayResult::default();
816    let mut messages: Vec<AqDeqMessage> = Vec::new();
817    let mut enq_msgid_blob: Option<Vec<u8>> = None;
818    let mut response_num_iters: u32 = 0;
819    let mut no_msg_found = false;
820    while reader.remaining() > 0 {
821        let message_type = reader.read_u8()?;
822        match message_type {
823            0 => {}
824            TNS_MSG_TYPE_PARAMETER => {
825                let num_iters = reader.read_ub4()?;
826                reader.limits().check_batch_rows(num_iters as usize)?;
827                response_num_iters = num_iters;
828                for i in 0..num_iters {
829                    let mut message = AqDeqMessage::default();
830                    let props_len = reader.read_ub2()?;
831                    if props_len > 0 {
832                        reader.read_u8()?; // skip_ub1
833                        process_msg_props(
834                            &mut reader,
835                            &mut message,
836                            capabilities.ttc_field_version,
837                        )?;
838                    }
839                    process_recipients(&mut reader)?;
840                    let payload_len = reader.read_ub2()?;
841                    if payload_len > 0 {
842                        message.payload = process_payload(&mut reader, kind)?;
843                    }
844                    let msgid = reader.read_bytes_with_length()?.unwrap_or_default();
845                    if operation == TNS_AQ_ARRAY_ENQ {
846                        enq_msgid_blob = Some(msgid);
847                    } else {
848                        message.msgid = Some(msgid);
849                    }
850                    let ext_len = reader.read_ub2()?;
851                    if ext_len > 0 {
852                        return Err(ProtocolError::UnsupportedFeature("AQ array extensions"));
853                    }
854                    let _output_ack = reader.read_ub2()?;
855                    if operation != TNS_AQ_ARRAY_ENQ {
856                        let _ = i;
857                        messages.push(message);
858                    }
859                }
860                if operation == TNS_AQ_ARRAY_ENQ {
861                    response_num_iters = reader.read_ub4()?;
862                    reader
863                        .limits()
864                        .check_batch_rows(response_num_iters as usize)?;
865                }
866            }
867            TNS_MSG_TYPE_STATUS => {
868                let _call_status = reader.read_ub4()?;
869                let _seq = reader.read_ub2()?;
870            }
871            TNS_MSG_TYPE_SERVER_SIDE_PIGGYBACK => {
872                let _ = skip_server_side_piggyback(&mut reader)?;
873            }
874            TNS_MSG_TYPE_END_OF_RESPONSE => break,
875            TNS_MSG_TYPE_ERROR => {
876                let info = parse_server_error_info(&mut reader, capabilities.ttc_field_version)?;
877                if info.number == TNS_ERR_NO_MESSAGES_FOUND as u32 {
878                    no_msg_found = true;
879                } else if info.number != 0 {
880                    return Err(ProtocolError::ServerErrorInfo(Box::new(
881                        info.into_details(),
882                    )));
883                }
884            }
885            _ => {
886                return Err(ProtocolError::UnknownMessageType {
887                    message_type,
888                    position: reader.position().saturating_sub(1),
889                })
890            }
891        }
892    }
893    if operation == TNS_AQ_ARRAY_ENQ {
894        if let Some(blob) = enq_msgid_blob {
895            let count = props_count as usize;
896            result.enq_msgids = (0..count)
897                .map(|j| {
898                    let start = j * 16;
899                    let end = start + 16;
900                    blob.get(start..end).map(<[u8]>::to_vec).unwrap_or_default()
901                })
902                .collect();
903        }
904    } else if no_msg_found {
905        result.deq_messages = Vec::new();
906    } else {
907        let keep = response_num_iters as usize;
908        messages.truncate(keep);
909        result.deq_messages = messages;
910    }
911    Ok(result)
912}
913
914// ---------------------------------------------------------------------------
915// Shared response decoders (reference aq_base.pyx).
916// ---------------------------------------------------------------------------
917
918fn process_msg_props(
919    reader: &mut TtcReader<'_>,
920    message: &mut AqDeqMessage,
921    ttc_field_version: u8,
922) -> Result<()> {
923    message.priority = reader.read_sb4()?;
924    message.delay = reader.read_sb4()?;
925    message.expiration = reader.read_sb4()?;
926    message.correlation = reader.read_string_with_length()?;
927    message.num_attempts = reader.read_sb4()?;
928    message.exception_queue = reader.read_string_with_length()?;
929    message.state = reader.read_sb4()?;
930    message.enq_time = process_date(reader)?;
931    let _enq_txn_id = reader.read_bytes_with_length()?;
932    process_extensions(reader)?;
933    let user_props = reader.read_ub4()?;
934    if user_props > 0 {
935        return Err(ProtocolError::UnsupportedFeature("AQ user properties"));
936    }
937    let _csn = reader.read_ub4()?;
938    let _dsn = reader.read_ub4()?;
939    let flags = reader.read_ub4()?;
940    message.delivery_mode = if flags == TNS_KPD_AQ_BUFMSG {
941        TNS_AQ_MSG_BUFFERED
942    } else {
943        TNS_AQ_MSG_PERSISTENT
944    };
945    if version_gates::carries_aq_shard_id(ttc_field_version) {
946        let _shard = reader.read_ub4()?;
947    }
948    Ok(())
949}
950
951/// Reads an Oracle date the way the reference `_process_date` does: a ub4
952/// presence flag, then `read_raw_bytes_and_length` (a single u8 length byte
953/// followed by that many raw date bytes).
954fn process_date(reader: &mut TtcReader<'_>) -> Result<Option<QueryValue>> {
955    let num_bytes = reader.read_ub4()?;
956    if num_bytes == 0 {
957        return Ok(None);
958    }
959    let len = usize::from(reader.read_u8()?);
960    if len == 0 {
961        return Ok(None);
962    }
963    let bytes = reader.read_raw(len)?;
964    Ok(Some(decode_datetime_value(bytes)?))
965}
966
967fn process_extensions(reader: &mut TtcReader<'_>) -> Result<()> {
968    let num_extensions = reader.read_ub4()?;
969    if num_extensions > 0 {
970        reader.read_u8()?; // skip_ub1
971        for _ in 0..num_extensions {
972            let _text = reader.read_bytes_with_length()?;
973            let _binary = reader.read_bytes_with_length()?;
974            let _keyword = reader.read_ub2()?;
975        }
976    }
977    Ok(())
978}
979
980fn process_recipients(reader: &mut TtcReader<'_>) -> Result<()> {
981    let count = reader.read_ub4()?;
982    if count > 0 {
983        return Err(ProtocolError::UnsupportedFeature(
984            "AQ recipients on dequeue",
985        ));
986    }
987    Ok(())
988}
989
990fn process_msg_id(reader: &mut TtcReader<'_>) -> Result<Vec<u8>> {
991    Ok(reader.read_raw(TNS_AQ_MESSAGE_ID_LENGTH)?.to_vec())
992}
993
994/// Decodes the payload of a dequeued message (reference `_process_payload`).
995fn process_payload(
996    reader: &mut TtcReader<'_>,
997    kind: &AqPayloadKind,
998) -> Result<Option<AqDeqPayload>> {
999    if matches!(kind, AqPayloadKind::Object) {
1000        // Object branch (reference `read_dbobject`): TOID/OID/snapshot
1001        // (length-prefixed each), version ub2, image-len ub4, flags ub2, then
1002        // the packed image as a bare length-prefixed chunk (`read_bytes`).
1003        let _toid = reader.read_bytes_with_length()?;
1004        let _oid = reader.read_bytes_with_length()?;
1005        let _snapshot = reader.read_bytes_with_length()?;
1006        let _version = reader.read_ub2()?;
1007        let image_length = reader.read_ub4()?;
1008        reader
1009            .limits()
1010            .check_response_bytes(image_length as usize)?;
1011        let _flags = reader.read_ub2()?;
1012        if image_length == 0 {
1013            return Ok(None);
1014        }
1015        let image = reader
1016            .read_bytes()?
1017            .ok_or(ProtocolError::TtcDecode("AQ object payload missing"))?;
1018        if image.len() != image_length as usize {
1019            return Err(ProtocolError::TtcDecode(
1020                "AQ object payload length differs from declared image length",
1021            ));
1022        }
1023        return Ok(Some(AqDeqPayload::Object(image)));
1024    }
1025    // RAW / JSON branch.
1026    let _toid = reader.read_bytes_with_length()?;
1027    let _oid = reader.read_bytes_with_length()?;
1028    let _snapshot = reader.read_bytes_with_length()?;
1029    let _version = reader.read_ub2()?;
1030    let image_length = reader.read_ub4()? as usize;
1031    reader.limits().check_response_bytes(image_length)?;
1032    let _flags = reader.read_ub2()?;
1033    if image_length > 0 {
1034        // reference: payload = read_bytes()[4:image_length]
1035        if image_length < 4 {
1036            return Err(ProtocolError::TtcDecode(
1037                "AQ payload image shorter than header",
1038            ));
1039        }
1040        let raw = reader
1041            .read_bytes()?
1042            .ok_or(ProtocolError::TtcDecode("AQ payload missing"))?;
1043        if raw.len() != image_length {
1044            return Err(ProtocolError::TtcDecode(
1045                "AQ payload length differs from declared image length",
1046            ));
1047        }
1048        let end = image_length;
1049        let start = 4.min(end);
1050        let payload = raw.get(start..end).unwrap_or_default().to_vec();
1051        if matches!(kind, AqPayloadKind::Json) {
1052            let value = decode_oson_with_limits(&payload, reader.limits())?;
1053            return Ok(Some(AqDeqPayload::Json(value)));
1054        }
1055        return Ok(Some(AqDeqPayload::Raw(payload)));
1056    }
1057    if matches!(kind, AqPayloadKind::Raw) {
1058        return Ok(Some(AqDeqPayload::Raw(Vec::new())));
1059    }
1060    Ok(None)
1061}
1062
1063#[cfg(test)]
1064mod tests {
1065    use super::*;
1066
1067    // Field version negotiated against the 23ai container (lane-1523). >= EXT_1
1068    // (18) so token_num is emitted; >= 21_1 (16) so shard id is emitted.
1069    const FV: u8 = 24;
1070
1071    fn caps() -> ClientCapabilities {
1072        ClientCapabilities {
1073            ttc_field_version: FV,
1074            max_string_size: 32767,
1075            charset_id: 873,
1076        }
1077    }
1078
1079    // Golden RAW enqueue request (FUNC 121), captured from python-oracledb 4.0.1
1080    // against lane-1523: msgproperties(payload=b"sample raw data 1",
1081    // correlation="CORR1", priority=2); enqone. TTC payload (offset 10 past the
1082    // packet header), seq_num=4. See tests/golden/aq_raw.txt.
1083    const GOLDEN_RAW_ENQ: &[u8] = &[
1084        0x03, 0x79, 0x04, 0x00, 0x01, 0x01, 0x0e, 0x01, 0x02, 0x00, 0x81, 0x01, 0x01, 0x05, 0x05,
1085        0x43, 0x4f, 0x52, 0x52, 0x31, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x04, 0x0e, 0x00, 0x00,
1086        0x01, 0x40, 0x00, 0x00, 0x01, 0x41, 0x00, 0x01, 0x01, 0x01, 0x00, 0x01, 0x42, 0x00, 0x00,
1087        0x01, 0x45, 0x00, 0x00, 0x00, 0x00, 0x04, 0xff, 0xff, 0xff, 0xff, 0x01, 0x00, 0x01, 0x02,
1088        0x00, 0x00, 0x00, 0x01, 0x01, 0x10, 0x01, 0x01, 0x00, 0x01, 0x01, 0x11, 0x01, 0x01, 0x10,
1089        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1090        0x00, 0x00, 0x00, 0x00, 0x0e, 0x54, 0x45, 0x53, 0x54, 0x5f, 0x52, 0x41, 0x57, 0x5f, 0x51,
1091        0x55, 0x45, 0x55, 0x45, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1092        0x00, 0x00, 0x00, 0x00, 0x17, 0x73, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x20, 0x72, 0x61, 0x77,
1093        0x20, 0x64, 0x61, 0x74, 0x61, 0x20, 0x31,
1094    ];
1095
1096    // Golden RAW dequeue request (FUNC 122): deqoptions wait=NO_WAIT,
1097    // navigation=DEQ_FIRST_MSG; deqone. seq_num=6.
1098    const GOLDEN_RAW_DEQ: &[u8] = &[
1099        0x03, 0x7a, 0x06, 0x00, 0x01, 0x01, 0x0e, 0x01, 0x01, 0x01, 0x01, 0x00, 0x00, 0x01, 0x03,
1100        0x01, 0x01, 0x01, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x01, 0x10, 0x01, 0x01, 0x01,
1101        0x01, 0x01, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0xff, 0xff, 0xff, 0xff, 0x0e,
1102        0x54, 0x45, 0x53, 0x54, 0x5f, 0x52, 0x41, 0x57, 0x5f, 0x51, 0x55, 0x45, 0x55, 0x45, 0x00,
1103        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x17,
1104    ];
1105
1106    #[test]
1107    fn raw_enqueue_request_matches_golden() {
1108        let queue = AqQueueDesc::new("TEST_RAW_QUEUE".to_string(), AqPayloadKind::Raw, None);
1109        let props = AqMsgProps {
1110            priority: 2,
1111            correlation: Some("CORR1".to_string()),
1112            payload: Some(AqPayloadValue::Raw(b"sample raw data 1".to_vec())),
1113            ..AqMsgProps::default()
1114        };
1115        let bytes = build_aq_enq_payload(&queue, &props, &AqEnqOptions::default(), 4, FV, false)
1116            .expect("build enqueue");
1117        assert_eq!(bytes, GOLDEN_RAW_ENQ);
1118    }
1119
1120    #[test]
1121    fn raw_dequeue_request_matches_golden() {
1122        let queue = AqQueueDesc::new("TEST_RAW_QUEUE".to_string(), AqPayloadKind::Raw, None);
1123        let deq = AqDeqOptions {
1124            wait: 0,
1125            navigation: 1,
1126            ..AqDeqOptions::default()
1127        };
1128        let bytes = build_aq_deq_payload(&queue, &deq, 6, FV).expect("build dequeue");
1129        assert_eq!(bytes, GOLDEN_RAW_DEQ);
1130    }
1131
1132    #[test]
1133    fn empty_queue_dequeue_yields_no_message() {
1134        // ORA-25228 is cleared and surfaces as no message. We can't synthesize a
1135        // full error packet trivially here, so just confirm an empty response
1136        // (status-only) yields None without error.
1137        let caps = caps();
1138        let res = parse_aq_deq_response(&[], caps, &AqPayloadKind::Raw).expect("parse");
1139        assert!(res.message.is_none());
1140    }
1141
1142    fn deq_response_with_raw_image(image_length: u32, raw_image: &[u8]) -> Vec<u8> {
1143        let mut writer = TtcWriter::new();
1144        writer.write_u8(TNS_MSG_TYPE_PARAMETER);
1145        writer.write_ub4(1);
1146        write_msg_props(&mut writer, &AqMsgProps::default(), FV).expect("write message props");
1147        writer.write_ub4(0); // recipients
1148        writer.write_ub4(0); // TOID
1149        writer.write_ub4(0); // OID
1150        writer.write_ub4(0); // snapshot
1151        writer.write_ub2(0); // version
1152        writer.write_ub4(image_length);
1153        writer.write_ub2(0); // flags
1154        writer
1155            .write_bytes_with_length(raw_image)
1156            .expect("write raw image field");
1157        writer.write_raw(&[0u8; TNS_AQ_MESSAGE_ID_LENGTH]);
1158        writer.write_u8(TNS_MSG_TYPE_END_OF_RESPONSE);
1159        writer.into_bytes()
1160    }
1161
1162    fn deq_response_with_object_image(image_length: u32, object_image: &[u8]) -> Vec<u8> {
1163        let mut writer = TtcWriter::new();
1164        writer.write_u8(TNS_MSG_TYPE_PARAMETER);
1165        writer.write_ub4(1);
1166        write_msg_props(&mut writer, &AqMsgProps::default(), FV).expect("write message props");
1167        writer.write_ub4(0); // recipients
1168        writer.write_ub4(0); // TOID
1169        writer.write_ub4(0); // OID
1170        writer.write_ub4(0); // snapshot
1171        writer.write_ub2(0); // version
1172        writer.write_ub4(image_length);
1173        writer.write_ub2(0); // flags
1174        writer
1175            .write_bytes_with_length(object_image)
1176            .expect("write object image field");
1177        writer.write_raw(&[0u8; TNS_AQ_MESSAGE_ID_LENGTH]);
1178        writer.write_u8(TNS_MSG_TYPE_END_OF_RESPONSE);
1179        writer.into_bytes()
1180    }
1181
1182    #[test]
1183    fn raw_dequeue_rejects_declared_image_length_shortfall() {
1184        let response = deq_response_with_raw_image(8, &[0, 0, 0, 0, b'A', b'B']);
1185        let err = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Raw)
1186            .expect_err("short RAW image must fail");
1187        assert!(matches!(err, ProtocolError::TtcDecode(_)), "got {err:?}");
1188    }
1189
1190    #[test]
1191    fn json_dequeue_rejects_declared_image_length_shortfall() {
1192        let response = deq_response_with_raw_image(8, &[0, 0, 0, 0, b'A', b'B']);
1193        let err = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Json)
1194            .expect_err("short JSON image must fail");
1195        assert!(matches!(err, ProtocolError::TtcDecode(_)), "got {err:?}");
1196    }
1197
1198    #[test]
1199    fn raw_dequeue_rejects_nonzero_image_length_below_header() {
1200        let response = deq_response_with_raw_image(3, &[0, 0, 0]);
1201        let err = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Raw)
1202            .expect_err("nonzero RAW image shorter than header must fail");
1203        assert!(matches!(err, ProtocolError::TtcDecode(_)), "got {err:?}");
1204    }
1205
1206    #[test]
1207    fn raw_dequeue_rejects_declared_image_length_trailing_bytes() {
1208        let response = deq_response_with_raw_image(6, &[0, 0, 0, 0, b'A', b'B', b'X']);
1209        let err = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Raw)
1210            .expect_err("trailing RAW image bytes must fail");
1211        assert!(matches!(err, ProtocolError::TtcDecode(_)), "got {err:?}");
1212    }
1213
1214    #[test]
1215    fn raw_dequeue_accepts_exact_declared_image_length() {
1216        let response = deq_response_with_raw_image(6, &[0, 0, 0, 0, b'A', b'B']);
1217        let res = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Raw)
1218            .expect("exact RAW image should parse");
1219        let message = res.message.expect("message present");
1220        match message.payload {
1221            Some(AqDeqPayload::Raw(payload)) => assert_eq!(payload, vec![b'A', b'B']),
1222            other => panic!("unexpected payload {other:?}"),
1223        }
1224    }
1225
1226    #[test]
1227    fn object_dequeue_rejects_declared_image_length_mismatch() {
1228        let response = deq_response_with_object_image(2, b"ABC");
1229        let err = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Object)
1230            .expect_err("mismatched object image length must fail");
1231        assert!(matches!(err, ProtocolError::TtcDecode(_)), "got {err:?}");
1232    }
1233
1234    #[test]
1235    fn object_dequeue_accepts_exact_declared_image_length() {
1236        let response = deq_response_with_object_image(3, b"ABC");
1237        let res = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Object)
1238            .expect("exact object image should parse");
1239        let message = res.message.expect("message present");
1240        match message.payload {
1241            Some(AqDeqPayload::Object(payload)) => assert_eq!(payload, b"ABC"),
1242            other => panic!("unexpected payload {other:?}"),
1243        }
1244    }
1245
1246    #[test]
1247    fn aq_array_response_respects_protocol_batch_limit() {
1248        let limits = ProtocolLimits {
1249            max_batch_rows: 1,
1250            ..ProtocolLimits::DEFAULT
1251        };
1252        let err = parse_aq_array_response_with_limits(
1253            &[],
1254            caps(),
1255            TNS_AQ_ARRAY_ENQ,
1256            2,
1257            &AqPayloadKind::Raw,
1258            limits,
1259        )
1260        .expect_err("client-side AQ batch count above policy must fail");
1261        assert!(
1262            matches!(
1263                err,
1264                ProtocolError::ResourceLimit {
1265                    limit: "batch_rows",
1266                    observed: 2,
1267                    maximum: 1,
1268                }
1269            ),
1270            "got {err:?}"
1271        );
1272    }
1273
1274    // Golden JSON enqueue (FUNC 121): msgproperties(payload=dict(name="John",
1275    // age=30, city="NY")); enqone against TEST_JSON_QUEUE, seq_num=4. The OSON
1276    // image (after `ff 4a 5a 01`) also exercises encode_oson byte-parity.
1277    const GOLDEN_JSON_ENQ: &[u8] = &[
1278        0x03, 0x79, 0x04, 0x00, 0x01, 0x01, 0x0f, 0x00, 0x00, 0x81, 0x01, 0x00, 0x00, 0x00, 0x00,
1279        0x00, 0x00, 0x01, 0x04, 0x0e, 0x00, 0x00, 0x01, 0x40, 0x00, 0x00, 0x01, 0x41, 0x00, 0x01,
1280        0x01, 0x01, 0x00, 0x01, 0x42, 0x00, 0x00, 0x01, 0x45, 0x00, 0x00, 0x00, 0x00, 0x04, 0xff,
1281        0xff, 0xff, 0xff, 0x01, 0x00, 0x01, 0x02, 0x00, 0x00, 0x00, 0x01, 0x01, 0x10, 0x01, 0x01,
1282        0x00, 0x00, 0x00, 0x01, 0x01, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1283        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x0f, 0x54, 0x45, 0x53, 0x54,
1284        0x5f, 0x4a, 0x53, 0x4f, 0x4e, 0x5f, 0x51, 0x55, 0x45, 0x55, 0x45, 0x00, 0x00, 0x00, 0x00,
1285        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x47, 0x01, 0x28, 0x00,
1286        0x26, 0x00, 0x04, 0x61, 0x08, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1287        0x00, 0x43, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1288        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x43, 0xff, 0x4a, 0x5a, 0x01, 0x21,
1289        0x02, 0x03, 0x00, 0x0e, 0x00, 0x1f, 0x00, 0x00, 0x42, 0x9c, 0xe6, 0x00, 0x09, 0x00, 0x05,
1290        0x00, 0x00, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x03, 0x61, 0x67, 0x65, 0x04, 0x63, 0x69, 0x74,
1291        0x79, 0xa4, 0x03, 0x03, 0x02, 0x01, 0x00, 0x00, 0x00, 0x11, 0x00, 0x00, 0x00, 0x17, 0x00,
1292        0x00, 0x00, 0x1b, 0x33, 0x04, 0x4a, 0x6f, 0x68, 0x6e, 0x34, 0x02, 0xc1, 0x1f, 0x33, 0x02,
1293        0x4e, 0x59,
1294    ];
1295
1296    #[test]
1297    fn json_enqueue_request_matches_golden() {
1298        let queue = AqQueueDesc::new("TEST_JSON_QUEUE".to_string(), AqPayloadKind::Json, None);
1299        // Insertion-ordered object {name, age, city}.
1300        let value = OsonValue::Object(vec![
1301            ("name".to_string(), OsonValue::String("John".to_string())),
1302            ("age".to_string(), OsonValue::Number("30".to_string())),
1303            ("city".to_string(), OsonValue::String("NY".to_string())),
1304        ]);
1305        let props = AqMsgProps {
1306            payload: Some(AqPayloadValue::Json(value)),
1307            ..AqMsgProps::default()
1308        };
1309        // Container negotiates >= 23ai so supports_oson_long_fnames = true.
1310        let bytes = build_aq_enq_payload(&queue, &props, &AqEnqOptions::default(), 4, FV, true)
1311            .expect("build json enqueue");
1312        assert_eq!(bytes, GOLDEN_JSON_ENQ);
1313    }
1314
1315    // ---- version-gate boundary tests (epic rust-oracledb-xver-parity-so3w) ----
1316    //
1317    // Each test builds/parses the message at exactly (boundary - 1) and
1318    // (boundary), proving the version-gated field is ABSENT below and PRESENT
1319    // at/above the boundary. These cover the AQ rows in docs/reference-gates.tsv;
1320    // the live matrix already crosses 20.1/21.1 (18c ver 11 < both vs 21c ver 16),
1321    // but the offline tests pin the exact bytes on every PR, not just nightly.
1322
1323    /// Assert `hi` is exactly `lo` with one contiguous block inserted: the
1324    /// gated field is absent below the boundary, present at/above it, and
1325    /// nothing else on the wire moved.
1326    fn assert_single_insertion(lo: &[u8], hi: &[u8], label: &str) {
1327        assert!(
1328            hi.len() > lo.len(),
1329            "{label}: gated field must add bytes at/above the boundary"
1330        );
1331        let prefix = lo.iter().zip(hi).take_while(|(a, b)| a == b).count();
1332        let suffix = lo[prefix..]
1333            .iter()
1334            .rev()
1335            .zip(hi[prefix..].iter().rev())
1336            .take_while(|(a, b)| a == b)
1337            .count();
1338        assert_eq!(
1339            prefix + suffix,
1340            lo.len(),
1341            "{label}: below-boundary bytes must equal above-boundary bytes minus one contiguous inserted block"
1342        );
1343    }
1344
1345    fn caps_fv(fv: u8) -> ClientCapabilities {
1346        ClientCapabilities {
1347            ttc_field_version: fv,
1348            max_string_size: 32767,
1349            charset_id: 873,
1350        }
1351    }
1352
1353    // Reference messages/aq_enq.pyx:115 — uint8 JSON-payload pointer, >= 20.1.
1354    #[test]
1355    fn aq_enq_gates_json_payload_pointer_on_20_1() {
1356        let queue = AqQueueDesc::new("Q".to_string(), AqPayloadKind::Raw, None);
1357        let props = AqMsgProps {
1358            payload: Some(AqPayloadValue::Raw(b"x".to_vec())),
1359            ..AqMsgProps::default()
1360        };
1361        let build = |fv| {
1362            build_aq_enq_payload(&queue, &props, &AqEnqOptions::default(), 1, fv, false)
1363                .expect("enq payload")
1364        };
1365        assert_single_insertion(
1366            &build(TNS_CCAP_FIELD_VERSION_20_1 - 1),
1367            &build(TNS_CCAP_FIELD_VERSION_20_1),
1368            "aq enq JSON-payload pointer (20.1)",
1369        );
1370    }
1371
1372    // Reference messages/aq_deq.pyx:130 — uint8 JSON-payload flag, >= 20.1.
1373    #[test]
1374    fn aq_deq_gates_json_payload_flag_on_20_1() {
1375        let queue = AqQueueDesc::new("Q".to_string(), AqPayloadKind::Raw, None);
1376        let build = |fv| {
1377            build_aq_deq_payload(&queue, &AqDeqOptions::default(), 1, fv).expect("deq payload")
1378        };
1379        assert_single_insertion(
1380            &build(TNS_CCAP_FIELD_VERSION_20_1 - 1),
1381            &build(TNS_CCAP_FIELD_VERSION_20_1),
1382            "aq deq JSON-payload flag (20.1)",
1383        );
1384    }
1385
1386    // Reference messages/aq_deq.pyx:132 — ub4 shard id (-1), >= 21.1. The 20.1
1387    // JSON flag is present on both sides (15,16 >= 14), so only the shard moves.
1388    #[test]
1389    fn aq_deq_gates_shard_id_on_21_1() {
1390        let queue = AqQueueDesc::new("Q".to_string(), AqPayloadKind::Raw, None);
1391        let build = |fv| {
1392            build_aq_deq_payload(&queue, &AqDeqOptions::default(), 1, fv).expect("deq payload")
1393        };
1394        assert_single_insertion(
1395            &build(TNS_CCAP_FIELD_VERSION_21_1 - 1),
1396            &build(TNS_CCAP_FIELD_VERSION_21_1),
1397            "aq deq shard id (21.1)",
1398        );
1399    }
1400
1401    // Reference messages/aq_array.pyx:196 — ub4 shard id (array enq/deq), >= 21.1.
1402    // Built with an empty message list so the array-level shard is the only
1403    // 21.1-gated field on the wire (a non-empty array also carries the
1404    // per-message-props shard, which is covered separately above).
1405    #[test]
1406    fn aq_array_gates_shard_id_on_21_1() {
1407        let queue = AqQueueDesc::new("Q".to_string(), AqPayloadKind::Raw, None);
1408        let build = |fv| {
1409            build_aq_array_enq_payload(&queue, &[], &AqEnqOptions::default(), 1, fv, false)
1410                .expect("array enq payload")
1411        };
1412        assert_single_insertion(
1413            &build(TNS_CCAP_FIELD_VERSION_21_1 - 1),
1414            &build(TNS_CCAP_FIELD_VERSION_21_1),
1415            "aq array shard id (21.1)",
1416        );
1417    }
1418
1419    // Reference messages/aq_base.pyx:197 — ub4 shard id (message props), >= 21.1.
1420    #[test]
1421    fn aq_msg_props_gates_shard_id_on_21_1() {
1422        let build = |fv| {
1423            let mut w = TtcWriter::new();
1424            write_msg_props(&mut w, &AqMsgProps::default(), fv).expect("msg props");
1425            w.into_bytes()
1426        };
1427        assert_single_insertion(
1428            &build(TNS_CCAP_FIELD_VERSION_21_1 - 1),
1429            &build(TNS_CCAP_FIELD_VERSION_21_1),
1430            "aq message-props shard id (21.1)",
1431        );
1432    }
1433
1434    // Reference messages/aq_base.pyx:129 — read/skip ub4 shard number (deq
1435    // props), >= 21.1. Round-trip: a response written at one field version and
1436    // parsed at another must misalign, proving the read side branches on the
1437    // same boundary as the write side.
1438    #[test]
1439    fn aq_deq_msg_props_read_gates_shard_on_21_1() {
1440        let response_at = |fv: u8| {
1441            let mut writer = TtcWriter::new();
1442            writer.write_u8(TNS_MSG_TYPE_PARAMETER);
1443            writer.write_ub4(1);
1444            write_msg_props(&mut writer, &AqMsgProps::default(), fv).expect("write message props");
1445            writer.write_ub4(0); // recipients
1446            writer.write_ub4(0); // TOID
1447            writer.write_ub4(0); // OID
1448            writer.write_ub4(0); // snapshot
1449            writer.write_ub2(0); // version
1450            writer.write_ub4(6); // image length (4-byte header + "AB")
1451            writer.write_ub2(0); // flags
1452            writer
1453                .write_bytes_with_length(&[0, 0, 0, 0, b'A', b'B'])
1454                .expect("write raw image field");
1455            writer.write_raw(&[0u8; TNS_AQ_MESSAGE_ID_LENGTH]);
1456            writer.write_u8(TNS_MSG_TYPE_END_OF_RESPONSE);
1457            writer.into_bytes()
1458        };
1459        let lo = TNS_CCAP_FIELD_VERSION_21_1 - 1;
1460        let hi = TNS_CCAP_FIELD_VERSION_21_1;
1461
1462        // Matched field versions round-trip to the same clean RAW payload.
1463        let raw_of = |resp: AqDeqResult| match resp.message.and_then(|m| m.payload) {
1464            Some(AqDeqPayload::Raw(bytes)) => Some(bytes),
1465            _ => None,
1466        };
1467        let matched_hi = raw_of(
1468            parse_aq_deq_response(&response_at(hi), caps_fv(hi), &AqPayloadKind::Raw)
1469                .expect("matched hi parse"),
1470        );
1471        let matched_lo = raw_of(
1472            parse_aq_deq_response(&response_at(lo), caps_fv(lo), &AqPayloadKind::Raw)
1473                .expect("matched lo parse"),
1474        );
1475        assert_eq!(
1476            matched_hi, matched_lo,
1477            "both bands decode the same RAW payload"
1478        );
1479        assert_eq!(matched_hi.as_deref(), Some(&b"AB"[..]));
1480
1481        // A response with the shard present, parsed as if it were absent, must
1482        // consume the 5 shard bytes as later fields — so it either fails closed
1483        // or decodes a different payload. Either way the read side branches on
1484        // the 21.1 boundary.
1485        let mismatched = parse_aq_deq_response(&response_at(hi), caps_fv(lo), &AqPayloadKind::Raw);
1486        let diverged = match mismatched {
1487            Err(_) => true,
1488            Ok(resp) => raw_of(resp) != matched_hi,
1489        };
1490        assert!(
1491            diverged,
1492            "read side must consume the 21.1 shard: skipping it changes the decode"
1493        );
1494    }
1495}
oracledb_protocol/thin/aq.rs

oracledb_protocol/thin/
aq.rs
