Added support for deliveryMode parameter in deqOptions

Author: sharadraju

doc/src/release_notes.rst

@@ -29,6 +29,9 @@ Common Changes
 #)  Fixed bug that threw empty error message when NaN values were used
     in JSON binds.
 
+#) Added support for :attr:`deqOptions.deliveryMode` in
+    :ref:`Advanced Queuing <aq>`.
+
 Thin Mode Changes
 +++++++++++++++++
 

lib/aqDeqOptions.js

@@ -74,6 +74,19 @@ class AqDeqOptions {
     this._impl.setCorrelation(value);
   }
 
+  //---------------------------------------------------------------------------
+  // deliveryMode
+  //
+  // Property for the deliveryMode to use for dequeuing messages.
+  //---------------------------------------------------------------------------
+  set deliveryMode(value) {
+    errors.assertPropValue(value === constants.AQ_MSG_DELIV_MODE_PERSISTENT ||
+      value === constants.AQ_MSG_DELIV_MODE_BUFFERED ||
+      value === constants.AQ_MSG_DELIV_MODE_PERSISTENT_OR_BUFFERED,
+    "deliveryMode");
+    this._impl.setDeliveryMode(value);
+  }
+
   //---------------------------------------------------------------------------
   // mode
   //

src/njsAqDeqOptions.c

@@ -45,6 +45,7 @@ NJS_NAPI_METHOD_DECL_SYNC(njsAqDeqOptions_getWait);
 NJS_NAPI_METHOD_DECL_SYNC(njsAqDeqOptions_setCondition);
 NJS_NAPI_METHOD_DECL_SYNC(njsAqDeqOptions_setConsumerName);
 NJS_NAPI_METHOD_DECL_SYNC(njsAqDeqOptions_setCorrelation);
+NJS_NAPI_METHOD_DECL_SYNC(njsAqDeqOptions_setDeliveryMode);
 NJS_NAPI_METHOD_DECL_SYNC(njsAqDeqOptions_setMode);
 NJS_NAPI_METHOD_DECL_SYNC(njsAqDeqOptions_setMsgId);
 NJS_NAPI_METHOD_DECL_SYNC(njsAqDeqOptions_setNavigation);
@@ -81,6 +82,8 @@ static const napi_property_descriptor njsClassProperties[] = {
             NULL, napi_enumerable, NULL },
     { "setCorrelation", NULL, njsAqDeqOptions_setCorrelation, NULL, NULL, NULL,
             napi_enumerable, NULL },
+    { "setDeliveryMode", NULL, njsAqDeqOptions_setDeliveryMode, NULL, NULL, NULL,
+            napi_enumerable, NULL },
     { "setMode", NULL, njsAqDeqOptions_setMode, NULL, NULL, NULL,
             napi_enumerable, NULL },
     { "setMsgId", NULL, njsAqDeqOptions_setMsgId, NULL, NULL, NULL,
@@ -309,6 +312,22 @@ NJS_NAPI_METHOD_IMPL_SYNC(njsAqDeqOptions_setCorrelation, 1, NULL)
 }
 
 
+//-----------------------------------------------------------------------------
+// njsAqDeqOptions_setDeliveryMode()
+//   Set accessor of "deliveryMode" property.
+//-----------------------------------------------------------------------------
+NJS_NAPI_METHOD_IMPL_SYNC(njsAqDeqOptions_setDeliveryMode, 1, NULL)
+{
+    njsAqDeqOptions *options = (njsAqDeqOptions*) callingInstance;
+    uint32_t value;
+
+    NJS_CHECK_NAPI(env, napi_get_value_uint32(env, args[0], &value))
+    if (dpiDeqOptions_setDeliveryMode(options->handle, (uint16_t) value) < 0)
+        return njsUtils_throwErrorDPI(env, globals);
+    return true;
+}
+
+
 //-----------------------------------------------------------------------------
 // njsAqDeqOptions_setMode()
 //   Set accessor of "mode" property.

test/aq1.js

@@ -46,6 +46,7 @@ describe('217. aq1.js', function() {
 
   const rawQueueName = "NODB_RAW_QUEUE";
   const RAW_TABLE = 'NODB_RAW_QUEUE_TAB';
+  let credential;
 
   before(async function() {
     if (!dbConfig.test.DBA_PRIVILEGE || oracledb.thin) {
@@ -56,7 +57,7 @@ describe('217. aq1.js', function() {
       this.skip();
     } else {
       await testsUtil.createAQtestUser(AQ_USER, AQ_USER_PWD);
-      const credential = {
+      credential = {
         user: AQ_USER,
         password: AQ_USER_PWD,
         connectString: dbConfig.connectString
@@ -251,7 +252,7 @@ describe('217. aq1.js', function() {
     const message = {
       payload: messageString // the message itself
     };
-    const myMsg = await queue1.enqOne(message);
+    await queue1.enqOne(message);
 
     // Get the deliveryMode attribute in enqOptions
     assert.strictEqual(queue1.enqOptions.deliveryMode,
@@ -265,7 +266,7 @@ describe('217. aq1.js', function() {
         visibility: oracledb.AQ_VISIBILITY_IMMEDIATE, // Change the visibility so that no explicit commit is required
       }
     );
-    queue2.deqOptions.deliveryMode = myMsg.deliveryMode;
+    queue2.deqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_PERSISTENT;
     const msg = await queue2.deqOne();
     if (msg) {
       assert.strictEqual(msg.payload.toString(), messageString);
@@ -499,4 +500,103 @@ describe('217. aq1.js', function() {
     assert.strictEqual(msg.payload.length, largeMessage.length);
     assert.strictEqual(msg.payload.toString(), largeMessage.toString());
   }); // 217.17
+
+  it('217.18 test enqueue/dequeue with BUFFERED delivery mode', async function() {
+    const queue = await conn.getQueue(rawQueueName);
+    queue.enqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    queue.enqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_BUFFERED;
+
+    const messageString = 'This is a buffered message';
+    await queue.enqOne(messageString);
+
+    const conn2 = await oracledb.getConnection(credential);
+    const queue2 = await conn2.getQueue(rawQueueName);
+    queue2.deqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    queue2.deqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_BUFFERED;
+
+    const msg = await queue2.deqOne();
+    await conn2.close();
+
+    assert(msg);
+    assert.strictEqual(msg.payload.toString(), messageString);
+    assert.strictEqual(msg.deliveryMode, oracledb.AQ_MSG_DELIV_MODE_BUFFERED);
+  }); // 217.18
+
+  it('217.19 test enqueue/dequeue with PERSISTENT delivery mode', async function() {
+    const queue = await conn.getQueue(rawQueueName);
+    queue.enqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    queue.enqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_PERSISTENT;
+
+    const messageString = 'This is a persistent message';
+    await queue.enqOne(messageString);
+
+    const conn2 = await oracledb.getConnection(credential);
+    const queue2 = await conn2.getQueue(rawQueueName);
+    queue2.deqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    queue2.deqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_PERSISTENT;
+
+    const msg = await queue2.deqOne();
+    await conn2.commit();
+    await conn2.close();
+
+    assert(msg);
+    assert.strictEqual(msg.payload.toString(), messageString);
+    assert.strictEqual(msg.deliveryMode, oracledb.AQ_MSG_DELIV_MODE_PERSISTENT);
+  }); // 217.19
+
+  it('217.20 test enqueue/dequeue with PERSISTENT_OR_BUFFERED delivery mode', async function() {
+    const queue = await conn.getQueue(rawQueueName);
+    queue.enqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    queue.enqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_PERSISTENT_OR_BUFFERED;
+
+    const messageString = 'This is a persistent or buffered message';
+    await queue.enqOne(messageString);
+
+    const conn2 = await oracledb.getConnection(credential);
+    const queue2 = await conn2.getQueue(rawQueueName);
+    queue2.deqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    queue2.deqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_PERSISTENT_OR_BUFFERED;
+
+    const msg = await queue2.deqOne();
+    await conn2.commit();
+    await conn2.close();
+
+    assert(msg);
+    assert.strictEqual(msg.payload.toString(), messageString);
+  }); // 217.20
+
+  it('217.21 test enqueue/dequeue with mismatched delivery modes', async function() {
+    const queue = await conn.getQueue(rawQueueName);
+    // queue.deqOptions.wait = oracledb.AQ_DEQ_NO_WAIT;
+    // queue.deqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    // queue.deqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_PERSISTENT_OR_BUFFERED;
+    // let myMsg = await queue.deqOne();
+    // while (myMsg) {
+    //   myMsg = await queue.deqOne();
+    // }
+    queue.enqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    queue.enqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_BUFFERED;
+
+    const messageString = 'This should not be dequeued with persistent mode';
+    await queue.enqOne(messageString);
+
+    const conn2 = await oracledb.getConnection(credential);
+    const queue2 = await conn2.getQueue(rawQueueName);
+    queue2.deqOptions.navigation = oracledb.AQ_DEQ_NAV_FIRST_MSG;
+    queue2.deqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    queue2.deqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_PERSISTENT;
+    queue2.deqOptions.wait = oracledb.AQ_DEQ_NO_WAIT;
+
+    const msg = await queue2.deqOne();
+    await conn2.close();
+
+    assert.strictEqual(msg, undefined);
+
+    // Cleanup: dequeue properly with matching mode
+    const cleanupQueue = await conn.getQueue(rawQueueName);
+    cleanupQueue.deqOptions.visibility = oracledb.AQ_VISIBILITY_IMMEDIATE;
+    cleanupQueue.deqOptions.deliveryMode = oracledb.AQ_MSG_DELIV_MODE_BUFFERED;
+    await cleanupQueue.deqOne();
+  }); // 217.21
+
 });

test/list.txt

@@ -4787,6 +4787,10 @@ oracledb.OUT_FORMAT_OBJECT and resultSet =  true
     217.15 test dequeue with wait timeout
     217.16 test buffer message payload - 32KB
     217.17 test large buffer message payload - 1MB
+    217.18 test enqueue/dequeue with BUFFERED delivery mode 
+    217.19 test enqueue/dequeue with PERSISTENT delivery mode
+    217.20 test enqueue/dequeue with PERSISTENT_OR_BUFFERED delivery mode
+    217.21 test enqueue/dequeue with mismatched delivery modes
 
 218. aq2.js
     218.1 examples/aqobject.js