Added support for recipient lists in Oracle Avanced Queueing in thin

mode.

Author: anthony-tuininga

doc/src/release_notes.rst

@@ -17,6 +17,8 @@ oracledb 3.2.0 (TBD)
 Thin Mode Changes
 +++++++++++++++++
 
+#)  Added support for :ref:`recipient lists <reciplists>` in Oracle Advanced
+    Queuing.
 #)  Emulate support for :meth:`Queue.deqmany()` with JSON payloads when using
     Oracle Database 21c by internally calling :meth:`Queue.deqone()` as many
     times as needed.

doc/src/user_guide/aq.rst

@@ -51,8 +51,8 @@ types which are detailed below.
 **Usage Notes**
 
 For classic queues, the use of :data:`oracledb.ENQ_IMMEDIATE` with bulk
-enqueuing, JMS payloads, and :ref:`Recipient Lists <reciplists>` are only
-supported in python-oracledb :ref:`Thick mode <enablingthick>`.
+enqueuing, and JMS payloads are only supported in python-oracledb
+:ref:`Thick mode <enablingthick>`.
 
 Transactional Event Queues do not support :attr:`EnqOptions.transformation`,
 :attr:`DeqOptions.transformation`, or :ref:`Recipient Lists <reciplists>`.

src/oracledb/aq.py

@@ -68,9 +68,6 @@ def _verify_message(self, message: "MessageProperties") -> None:
         else:
             if not isinstance(message.payload, (str, bytes)):
                 errors._raise_err(errors.ERR_PAYLOAD_CANNOT_BE_ENQUEUED)
-        if self.connection.thin:
-            if message.recipients:
-                errors._raise_not_supported("specifying AQ message recipients")
 
     @property
     def connection(self) -> "connection_module.Connection":

src/oracledb/impl/thin/messages/aq_array.pyx

@@ -159,7 +159,11 @@ cdef class AqArrayMessage(AqBaseMessage):
             buf.write_uint8(TNS_MSG_TYPE_ROW_DATA)
             buf.write_ub4(flags)                    # aqi flags
             self._write_msg_props(buf, props_impl)
-            buf.write_ub4(0)                        # num recipients
+            if props_impl.recipients is None:
+                buf.write_ub4(0)                    # num recipients
+            else:
+                buf.write_ub4(3 * len(props_impl.recipients))
+                self._write_recipients(buf, props_impl)
             buf.write_sb4(self.enq_options_impl.visibility)
             buf.write_ub4(0)                        # relative msg id
             buf.write_sb4(0)                        # seq deviation

src/oracledb/impl/thin/messages/aq_base.pyx

@@ -179,13 +179,13 @@ cdef class AqBaseMessage(Message):
         self._write_value_with_length(buf, props_impl.enq_txn_id)
         buf.write_ub4(4)                            # number of extensions
         buf.write_uint8(0x0e)                       # unknown extra byte
-        buf.write_extension_values(None, None, TNS_AQ_EXT_KEYWORD_AGENT_NAME)
-        buf.write_extension_values(None, None,
-                                   TNS_AQ_EXT_KEYWORD_AGENT_ADDRESS)
-        buf.write_extension_values(None, b'\x00',
-                                   TNS_AQ_EXT_KEYWORD_AGENT_PROTOCOL)
-        buf.write_extension_values(None, None,
-                                   TNS_AQ_EXT_KEYWORD_ORIGINAL_MSGID)
+        buf.write_keyword_value_pair(None, None, TNS_AQ_EXT_KEYWORD_AGENT_NAME)
+        buf.write_keyword_value_pair(None, None,
+                                     TNS_AQ_EXT_KEYWORD_AGENT_ADDRESS)
+        buf.write_keyword_value_pair(None, b'\x00',
+                                     TNS_AQ_EXT_KEYWORD_AGENT_PROTOCOL)
+        buf.write_keyword_value_pair(None, None,
+                                     TNS_AQ_EXT_KEYWORD_ORIGINAL_MSGID)
         buf.write_ub4(0)                            # user property
         buf.write_ub4(0)                            # cscn
         buf.write_ub4(0)                            # dscn
@@ -206,6 +206,22 @@ cdef class AqBaseMessage(Message):
         else:
             buf.write_bytes(props_impl.payload_obj)
 
+
+    cdef int _write_recipients(self, WriteBuffer buf,
+                               ThinMsgPropsImpl props_impl) except -1:
+        """
+        Write the recipient list of the message property object to the
+        buffer.
+        """
+        cdef:
+            uint16_t index = 0
+            str recipient
+        for recipient in props_impl.recipients:
+            buf.write_keyword_value_pair(recipient, None, index)
+            buf.write_keyword_value_pair(None, None, index + 1)
+            buf.write_keyword_value_pair(None, b'\x00', index + 2)
+            index += 3
+
     cdef int _write_value_with_length(self, WriteBuffer buf,
                                       object value) except -1:
         """

src/oracledb/impl/thin/messages/aq_enq.pyx

@@ -64,8 +64,12 @@ cdef class AqEnqMessage(AqBaseMessage):
         buf.write_uint8(1)                      # queue name (pointer)
         buf.write_ub4(len(queue_name_bytes))    # queue name length
         self._write_msg_props(buf, self.props_impl)
-        buf.write_uint8(0)                      # recipients (pointer)
-        buf.write_ub4(0)                        # number of key/value pairs
+        if self.props_impl.recipients is None:
+            buf.write_uint8(0)                  # recipients (pointer)
+            buf.write_ub4(0)                    # number of key/value pairs
+        else:
+            buf.write_uint8(1)                  # recipients (pointer)
+            buf.write_ub4(3 * len(self.props_impl.recipients))
         buf.write_ub4(self.enq_options_impl.visibility)
         buf.write_uint8(0)                      # relative message id
         buf.write_ub4(0)                        # relative message length
@@ -115,5 +119,7 @@ cdef class AqEnqMessage(AqBaseMessage):
                 buf.write_uint8(0)              # JSON payload (pointer)
 
         buf.write_bytes_with_length(queue_name_bytes)
+        if self.props_impl.recipients is not None:
+            self._write_recipients(buf, self.props_impl)
         buf.write_bytes(self.queue_impl.payload_toid)
         self._write_payload(buf, self.props_impl)

src/oracledb/impl/thin/packet.pyx

@@ -895,24 +895,23 @@ cdef class WriteBuffer(Buffer):
         self.write_ub4(obj_impl.flags)      # flags
         self.write_bytes_with_length(packed_data)
 
-    cdef int write_extension_values(self, str txt_value, bytes bytes_value,
-                                    uint16_t keyword) except -1:
+    cdef int write_keyword_value_pair(self, str text_value, bytes binary_value,
+                                      uint16_t keyword) except -1:
         """
-        Writes extension's text value, binary value and keyword entry to the
-        buffer.
+        Writes a keyword/value pair (text and binary values) to the buffer.
         """
-        cdef bytes txt_value_bytes
-        if txt_value is None:
-            self.write_uint8(0)
+        cdef bytes text_value_bytes
+        if text_value is None:
+            self.write_ub4(0)
         else:
-            txt_value_bytes = txt_value.encode()
-            self.write_ub4(len(txt_value_bytes))
-            self.write_bytes_with_length(txt_value_bytes)
-        if bytes_value is None:
-            self.write_uint8(0)
+            text_value_bytes = text_value.encode()
+            self.write_ub4(len(text_value_bytes))
+            self.write_bytes_with_length(text_value_bytes)
+        if binary_value is None:
+            self.write_ub4(0)
         else:
-            self.write_ub4(len(bytes_value))
-            self.write_bytes_with_length(bytes_value)
+            self.write_ub4(len(binary_value))
+            self.write_bytes_with_length(binary_value)
         self.write_ub2(keyword)
 
     cdef int write_lob_with_length(self, BaseThinLobImpl lob_impl) except -1:

tests/test_2700_aq_dbobject.py

@@ -410,9 +410,6 @@ def test_2718(self):
         props = queue.deqone()
         self.assertEqual(props.msgid, actual_msgid)
 
-    @unittest.skipIf(
-        test_env.get_is_thin(), "Thin mode doesn't support recipient list yet"
-    )
     def test_2719(self):
         "2719 - verify use of recipients property"
         books_type = self.conn.gettype(self.book_type_name)