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        return Ok(Some(AqDeqPayload::Object(image)));
1019    }
1020    // RAW / JSON branch.
1021    let _toid = reader.read_bytes_with_length()?;
1022    let _oid = reader.read_bytes_with_length()?;
1023    let _snapshot = reader.read_bytes_with_length()?;
1024    let _version = reader.read_ub2()?;
1025    let image_length = reader.read_ub4()? as usize;
1026    reader.limits().check_response_bytes(image_length)?;
1027    let _flags = reader.read_ub2()?;
1028    if image_length > 0 {
1029        // reference: payload = read_bytes()[4:image_length]
1030        let raw = reader
1031            .read_bytes()?
1032            .ok_or(ProtocolError::TtcDecode("AQ payload missing"))?;
1033        if raw.len() < image_length {
1034            return Err(ProtocolError::TtcDecode("AQ payload shorter than declared"));
1035        }
1036        let end = image_length;
1037        let start = 4.min(end);
1038        let payload = raw.get(start..end).unwrap_or_default().to_vec();
1039        if matches!(kind, AqPayloadKind::Json) {
1040            let value = decode_oson_with_limits(&payload, reader.limits())?;
1041            return Ok(Some(AqDeqPayload::Json(value)));
1042        }
1043        return Ok(Some(AqDeqPayload::Raw(payload)));
1044    }
1045    if matches!(kind, AqPayloadKind::Raw) {
1046        return Ok(Some(AqDeqPayload::Raw(Vec::new())));
1047    }
1048    Ok(None)
1049}
1050
1051#[cfg(test)]
1052mod tests {
1053    use super::*;
1054
1055    // Field version negotiated against the 23ai container (lane-1523). >= EXT_1
1056    // (18) so token_num is emitted; >= 21_1 (16) so shard id is emitted.
1057    const FV: u8 = 24;
1058
1059    fn caps() -> ClientCapabilities {
1060        ClientCapabilities {
1061            ttc_field_version: FV,
1062            max_string_size: 32767,
1063            charset_id: 873,
1064        }
1065    }
1066
1067    // Golden RAW enqueue request (FUNC 121), captured from python-oracledb 4.0.1
1068    // against lane-1523: msgproperties(payload=b"sample raw data 1",
1069    // correlation="CORR1", priority=2); enqone. TTC payload (offset 10 past the
1070    // packet header), seq_num=4. See tests/golden/aq_raw.txt.
1071    const GOLDEN_RAW_ENQ: &[u8] = &[
1072        0x03, 0x79, 0x04, 0x00, 0x01, 0x01, 0x0e, 0x01, 0x02, 0x00, 0x81, 0x01, 0x01, 0x05, 0x05,
1073        0x43, 0x4f, 0x52, 0x52, 0x31, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x04, 0x0e, 0x00, 0x00,
1074        0x01, 0x40, 0x00, 0x00, 0x01, 0x41, 0x00, 0x01, 0x01, 0x01, 0x00, 0x01, 0x42, 0x00, 0x00,
1075        0x01, 0x45, 0x00, 0x00, 0x00, 0x00, 0x04, 0xff, 0xff, 0xff, 0xff, 0x01, 0x00, 0x01, 0x02,
1076        0x00, 0x00, 0x00, 0x01, 0x01, 0x10, 0x01, 0x01, 0x00, 0x01, 0x01, 0x11, 0x01, 0x01, 0x10,
1077        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1078        0x00, 0x00, 0x00, 0x00, 0x0e, 0x54, 0x45, 0x53, 0x54, 0x5f, 0x52, 0x41, 0x57, 0x5f, 0x51,
1079        0x55, 0x45, 0x55, 0x45, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1080        0x00, 0x00, 0x00, 0x00, 0x17, 0x73, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x20, 0x72, 0x61, 0x77,
1081        0x20, 0x64, 0x61, 0x74, 0x61, 0x20, 0x31,
1082    ];
1083
1084    // Golden RAW dequeue request (FUNC 122): deqoptions wait=NO_WAIT,
1085    // navigation=DEQ_FIRST_MSG; deqone. seq_num=6.
1086    const GOLDEN_RAW_DEQ: &[u8] = &[
1087        0x03, 0x7a, 0x06, 0x00, 0x01, 0x01, 0x0e, 0x01, 0x01, 0x01, 0x01, 0x00, 0x00, 0x01, 0x03,
1088        0x01, 0x01, 0x01, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x01, 0x10, 0x01, 0x01, 0x01,
1089        0x01, 0x01, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0xff, 0xff, 0xff, 0xff, 0x0e,
1090        0x54, 0x45, 0x53, 0x54, 0x5f, 0x52, 0x41, 0x57, 0x5f, 0x51, 0x55, 0x45, 0x55, 0x45, 0x00,
1091        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x17,
1092    ];
1093
1094    #[test]
1095    fn raw_enqueue_request_matches_golden() {
1096        let queue = AqQueueDesc::new("TEST_RAW_QUEUE".to_string(), AqPayloadKind::Raw, None);
1097        let props = AqMsgProps {
1098            priority: 2,
1099            correlation: Some("CORR1".to_string()),
1100            payload: Some(AqPayloadValue::Raw(b"sample raw data 1".to_vec())),
1101            ..AqMsgProps::default()
1102        };
1103        let bytes = build_aq_enq_payload(&queue, &props, &AqEnqOptions::default(), 4, FV, false)
1104            .expect("build enqueue");
1105        assert_eq!(bytes, GOLDEN_RAW_ENQ);
1106    }
1107
1108    #[test]
1109    fn raw_dequeue_request_matches_golden() {
1110        let queue = AqQueueDesc::new("TEST_RAW_QUEUE".to_string(), AqPayloadKind::Raw, None);
1111        let deq = AqDeqOptions {
1112            wait: 0,
1113            navigation: 1,
1114            ..AqDeqOptions::default()
1115        };
1116        let bytes = build_aq_deq_payload(&queue, &deq, 6, FV).expect("build dequeue");
1117        assert_eq!(bytes, GOLDEN_RAW_DEQ);
1118    }
1119
1120    #[test]
1121    fn empty_queue_dequeue_yields_no_message() {
1122        // ORA-25228 is cleared and surfaces as no message. We can't synthesize a
1123        // full error packet trivially here, so just confirm an empty response
1124        // (status-only) yields None without error.
1125        let caps = caps();
1126        let res = parse_aq_deq_response(&[], caps, &AqPayloadKind::Raw).expect("parse");
1127        assert!(res.message.is_none());
1128    }
1129
1130    fn deq_response_with_raw_image(image_length: u32, raw_image: &[u8]) -> Vec<u8> {
1131        let mut writer = TtcWriter::new();
1132        writer.write_u8(TNS_MSG_TYPE_PARAMETER);
1133        writer.write_ub4(1);
1134        write_msg_props(&mut writer, &AqMsgProps::default(), FV).expect("write message props");
1135        writer.write_ub4(0); // recipients
1136        writer.write_ub4(0); // TOID
1137        writer.write_ub4(0); // OID
1138        writer.write_ub4(0); // snapshot
1139        writer.write_ub2(0); // version
1140        writer.write_ub4(image_length);
1141        writer.write_ub2(0); // flags
1142        writer
1143            .write_bytes_with_length(raw_image)
1144            .expect("write raw image field");
1145        writer.write_raw(&[0u8; TNS_AQ_MESSAGE_ID_LENGTH]);
1146        writer.write_u8(TNS_MSG_TYPE_END_OF_RESPONSE);
1147        writer.into_bytes()
1148    }
1149
1150    #[test]
1151    fn raw_dequeue_rejects_declared_image_length_shortfall() {
1152        let response = deq_response_with_raw_image(8, &[0, 0, 0, 0, b'A', b'B']);
1153        let err = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Raw)
1154            .expect_err("short RAW image must fail");
1155        assert!(matches!(err, ProtocolError::TtcDecode(_)), "got {err:?}");
1156    }
1157
1158    #[test]
1159    fn json_dequeue_rejects_declared_image_length_shortfall() {
1160        let response = deq_response_with_raw_image(8, &[0, 0, 0, 0, b'A', b'B']);
1161        let err = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Json)
1162            .expect_err("short JSON image must fail");
1163        assert!(matches!(err, ProtocolError::TtcDecode(_)), "got {err:?}");
1164    }
1165
1166    #[test]
1167    fn raw_dequeue_accepts_exact_declared_image_length() {
1168        let response = deq_response_with_raw_image(6, &[0, 0, 0, 0, b'A', b'B']);
1169        let res = parse_aq_deq_response(&response, caps(), &AqPayloadKind::Raw)
1170            .expect("exact RAW image should parse");
1171        let message = res.message.expect("message present");
1172        match message.payload {
1173            Some(AqDeqPayload::Raw(payload)) => assert_eq!(payload, vec![b'A', b'B']),
1174            other => panic!("unexpected payload {other:?}"),
1175        }
1176    }
1177
1178    #[test]
1179    fn aq_array_response_respects_protocol_batch_limit() {
1180        let limits = ProtocolLimits {
1181            max_batch_rows: 1,
1182            ..ProtocolLimits::DEFAULT
1183        };
1184        let err = parse_aq_array_response_with_limits(
1185            &[],
1186            caps(),
1187            TNS_AQ_ARRAY_ENQ,
1188            2,
1189            &AqPayloadKind::Raw,
1190            limits,
1191        )
1192        .expect_err("client-side AQ batch count above policy must fail");
1193        assert!(
1194            matches!(
1195                err,
1196                ProtocolError::ResourceLimit {
1197                    limit: "batch_rows",
1198                    observed: 2,
1199                    maximum: 1,
1200                }
1201            ),
1202            "got {err:?}"
1203        );
1204    }
1205
1206    // Golden JSON enqueue (FUNC 121): msgproperties(payload=dict(name="John",
1207    // age=30, city="NY")); enqone against TEST_JSON_QUEUE, seq_num=4. The OSON
1208    // image (after `ff 4a 5a 01`) also exercises encode_oson byte-parity.
1209    const GOLDEN_JSON_ENQ: &[u8] = &[
1210        0x03, 0x79, 0x04, 0x00, 0x01, 0x01, 0x0f, 0x00, 0x00, 0x81, 0x01, 0x00, 0x00, 0x00, 0x00,
1211        0x00, 0x00, 0x01, 0x04, 0x0e, 0x00, 0x00, 0x01, 0x40, 0x00, 0x00, 0x01, 0x41, 0x00, 0x01,
1212        0x01, 0x01, 0x00, 0x01, 0x42, 0x00, 0x00, 0x01, 0x45, 0x00, 0x00, 0x00, 0x00, 0x04, 0xff,
1213        0xff, 0xff, 0xff, 0x01, 0x00, 0x01, 0x02, 0x00, 0x00, 0x00, 0x01, 0x01, 0x10, 0x01, 0x01,
1214        0x00, 0x00, 0x00, 0x01, 0x01, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1215        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x0f, 0x54, 0x45, 0x53, 0x54,
1216        0x5f, 0x4a, 0x53, 0x4f, 0x4e, 0x5f, 0x51, 0x55, 0x45, 0x55, 0x45, 0x00, 0x00, 0x00, 0x00,
1217        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x47, 0x01, 0x28, 0x00,
1218        0x26, 0x00, 0x04, 0x61, 0x08, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1219        0x00, 0x43, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1220        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x43, 0xff, 0x4a, 0x5a, 0x01, 0x21,
1221        0x02, 0x03, 0x00, 0x0e, 0x00, 0x1f, 0x00, 0x00, 0x42, 0x9c, 0xe6, 0x00, 0x09, 0x00, 0x05,
1222        0x00, 0x00, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x03, 0x61, 0x67, 0x65, 0x04, 0x63, 0x69, 0x74,
1223        0x79, 0xa4, 0x03, 0x03, 0x02, 0x01, 0x00, 0x00, 0x00, 0x11, 0x00, 0x00, 0x00, 0x17, 0x00,
1224        0x00, 0x00, 0x1b, 0x33, 0x04, 0x4a, 0x6f, 0x68, 0x6e, 0x34, 0x02, 0xc1, 0x1f, 0x33, 0x02,
1225        0x4e, 0x59,
1226    ];
1227
1228    #[test]
1229    fn json_enqueue_request_matches_golden() {
1230        let queue = AqQueueDesc::new("TEST_JSON_QUEUE".to_string(), AqPayloadKind::Json, None);
1231        // Insertion-ordered object {name, age, city}.
1232        let value = OsonValue::Object(vec![
1233            ("name".to_string(), OsonValue::String("John".to_string())),
1234            ("age".to_string(), OsonValue::Number("30".to_string())),
1235            ("city".to_string(), OsonValue::String("NY".to_string())),
1236        ]);
1237        let props = AqMsgProps {
1238            payload: Some(AqPayloadValue::Json(value)),
1239            ..AqMsgProps::default()
1240        };
1241        // Container negotiates >= 23ai so supports_oson_long_fnames = true.
1242        let bytes = build_aq_enq_payload(&queue, &props, &AqEnqOptions::default(), 4, FV, true)
1243            .expect("build json enqueue");
1244        assert_eq!(bytes, GOLDEN_JSON_ENQ);
1245    }
1246
1247    // ---- version-gate boundary tests (epic rust-oracledb-xver-parity-so3w) ----
1248    //
1249    // Each test builds/parses the message at exactly (boundary - 1) and
1250    // (boundary), proving the version-gated field is ABSENT below and PRESENT
1251    // at/above the boundary. These cover the AQ rows in docs/reference-gates.tsv;
1252    // the live matrix already crosses 20.1/21.1 (18c ver 11 < both vs 21c ver 16),
1253    // but the offline tests pin the exact bytes on every PR, not just nightly.
1254
1255    /// Assert `hi` is exactly `lo` with one contiguous block inserted: the
1256    /// gated field is absent below the boundary, present at/above it, and
1257    /// nothing else on the wire moved.
1258    fn assert_single_insertion(lo: &[u8], hi: &[u8], label: &str) {
1259        assert!(
1260            hi.len() > lo.len(),
1261            "{label}: gated field must add bytes at/above the boundary"
1262        );
1263        let prefix = lo.iter().zip(hi).take_while(|(a, b)| a == b).count();
1264        let suffix = lo[prefix..]
1265            .iter()
1266            .rev()
1267            .zip(hi[prefix..].iter().rev())
1268            .take_while(|(a, b)| a == b)
1269            .count();
1270        assert_eq!(
1271            prefix + suffix,
1272            lo.len(),
1273            "{label}: below-boundary bytes must equal above-boundary bytes minus one contiguous inserted block"
1274        );
1275    }
1276
1277    fn caps_fv(fv: u8) -> ClientCapabilities {
1278        ClientCapabilities {
1279            ttc_field_version: fv,
1280            max_string_size: 32767,
1281            charset_id: 873,
1282        }
1283    }
1284
1285    // Reference messages/aq_enq.pyx:115 — uint8 JSON-payload pointer, >= 20.1.
1286    #[test]
1287    fn aq_enq_gates_json_payload_pointer_on_20_1() {
1288        let queue = AqQueueDesc::new("Q".to_string(), AqPayloadKind::Raw, None);
1289        let props = AqMsgProps {
1290            payload: Some(AqPayloadValue::Raw(b"x".to_vec())),
1291            ..AqMsgProps::default()
1292        };
1293        let build = |fv| {
1294            build_aq_enq_payload(&queue, &props, &AqEnqOptions::default(), 1, fv, false)
1295                .expect("enq payload")
1296        };
1297        assert_single_insertion(
1298            &build(TNS_CCAP_FIELD_VERSION_20_1 - 1),
1299            &build(TNS_CCAP_FIELD_VERSION_20_1),
1300            "aq enq JSON-payload pointer (20.1)",
1301        );
1302    }
1303
1304    // Reference messages/aq_deq.pyx:130 — uint8 JSON-payload flag, >= 20.1.
1305    #[test]
1306    fn aq_deq_gates_json_payload_flag_on_20_1() {
1307        let queue = AqQueueDesc::new("Q".to_string(), AqPayloadKind::Raw, None);
1308        let build = |fv| {
1309            build_aq_deq_payload(&queue, &AqDeqOptions::default(), 1, fv).expect("deq payload")
1310        };
1311        assert_single_insertion(
1312            &build(TNS_CCAP_FIELD_VERSION_20_1 - 1),
1313            &build(TNS_CCAP_FIELD_VERSION_20_1),
1314            "aq deq JSON-payload flag (20.1)",
1315        );
1316    }
1317
1318    // Reference messages/aq_deq.pyx:132 — ub4 shard id (-1), >= 21.1. The 20.1
1319    // JSON flag is present on both sides (15,16 >= 14), so only the shard moves.
1320    #[test]
1321    fn aq_deq_gates_shard_id_on_21_1() {
1322        let queue = AqQueueDesc::new("Q".to_string(), AqPayloadKind::Raw, None);
1323        let build = |fv| {
1324            build_aq_deq_payload(&queue, &AqDeqOptions::default(), 1, fv).expect("deq payload")
1325        };
1326        assert_single_insertion(
1327            &build(TNS_CCAP_FIELD_VERSION_21_1 - 1),
1328            &build(TNS_CCAP_FIELD_VERSION_21_1),
1329            "aq deq shard id (21.1)",
1330        );
1331    }
1332
1333    // Reference messages/aq_array.pyx:196 — ub4 shard id (array enq/deq), >= 21.1.
1334    // Built with an empty message list so the array-level shard is the only
1335    // 21.1-gated field on the wire (a non-empty array also carries the
1336    // per-message-props shard, which is covered separately above).
1337    #[test]
1338    fn aq_array_gates_shard_id_on_21_1() {
1339        let queue = AqQueueDesc::new("Q".to_string(), AqPayloadKind::Raw, None);
1340        let build = |fv| {
1341            build_aq_array_enq_payload(&queue, &[], &AqEnqOptions::default(), 1, fv, false)
1342                .expect("array enq payload")
1343        };
1344        assert_single_insertion(
1345            &build(TNS_CCAP_FIELD_VERSION_21_1 - 1),
1346            &build(TNS_CCAP_FIELD_VERSION_21_1),
1347            "aq array shard id (21.1)",
1348        );
1349    }
1350
1351    // Reference messages/aq_base.pyx:197 — ub4 shard id (message props), >= 21.1.
1352    #[test]
1353    fn aq_msg_props_gates_shard_id_on_21_1() {
1354        let build = |fv| {
1355            let mut w = TtcWriter::new();
1356            write_msg_props(&mut w, &AqMsgProps::default(), fv).expect("msg props");
1357            w.into_bytes()
1358        };
1359        assert_single_insertion(
1360            &build(TNS_CCAP_FIELD_VERSION_21_1 - 1),
1361            &build(TNS_CCAP_FIELD_VERSION_21_1),
1362            "aq message-props shard id (21.1)",
1363        );
1364    }
1365
1366    // Reference messages/aq_base.pyx:129 — read/skip ub4 shard number (deq
1367    // props), >= 21.1. Round-trip: a response written at one field version and
1368    // parsed at another must misalign, proving the read side branches on the
1369    // same boundary as the write side.
1370    #[test]
1371    fn aq_deq_msg_props_read_gates_shard_on_21_1() {
1372        let response_at = |fv: u8| {
1373            let mut writer = TtcWriter::new();
1374            writer.write_u8(TNS_MSG_TYPE_PARAMETER);
1375            writer.write_ub4(1);
1376            write_msg_props(&mut writer, &AqMsgProps::default(), fv).expect("write message props");
1377            writer.write_ub4(0); // recipients
1378            writer.write_ub4(0); // TOID
1379            writer.write_ub4(0); // OID
1380            writer.write_ub4(0); // snapshot
1381            writer.write_ub2(0); // version
1382            writer.write_ub4(6); // image length (4-byte header + "AB")
1383            writer.write_ub2(0); // flags
1384            writer
1385                .write_bytes_with_length(&[0, 0, 0, 0, b'A', b'B'])
1386                .expect("write raw image field");
1387            writer.write_raw(&[0u8; TNS_AQ_MESSAGE_ID_LENGTH]);
1388            writer.write_u8(TNS_MSG_TYPE_END_OF_RESPONSE);
1389            writer.into_bytes()
1390        };
1391        let lo = TNS_CCAP_FIELD_VERSION_21_1 - 1;
1392        let hi = TNS_CCAP_FIELD_VERSION_21_1;
1393
1394        // Matched field versions round-trip to the same clean RAW payload.
1395        let raw_of = |resp: AqDeqResult| match resp.message.and_then(|m| m.payload) {
1396            Some(AqDeqPayload::Raw(bytes)) => Some(bytes),
1397            _ => None,
1398        };
1399        let matched_hi = raw_of(
1400            parse_aq_deq_response(&response_at(hi), caps_fv(hi), &AqPayloadKind::Raw)
1401                .expect("matched hi parse"),
1402        );
1403        let matched_lo = raw_of(
1404            parse_aq_deq_response(&response_at(lo), caps_fv(lo), &AqPayloadKind::Raw)
1405                .expect("matched lo parse"),
1406        );
1407        assert_eq!(
1408            matched_hi, matched_lo,
1409            "both bands decode the same RAW payload"
1410        );
1411        assert_eq!(matched_hi.as_deref(), Some(&b"AB"[..]));
1412
1413        // A response with the shard present, parsed as if it were absent, must
1414        // consume the 5 shard bytes as later fields — so it either fails closed
1415        // or decodes a different payload. Either way the read side branches on
1416        // the 21.1 boundary.
1417        let mismatched = parse_aq_deq_response(&response_at(hi), caps_fv(lo), &AqPayloadKind::Raw);
1418        let diverged = match mismatched {
1419            Err(_) => true,
1420            Ok(resp) => raw_of(resp) != matched_hi,
1421        };
1422        assert!(
1423            diverged,
1424            "read side must consume the 21.1 shard: skipping it changes the decode"
1425        );
1426    }
1427}
oracledb_protocol/thin/aq.rs

oracledb_protocol/thin/
aq.rs
