Added support for enqueuing and dequeuing JSON payloads using Advanced

Queuing (AQ).

Author: anthony-tuininga

doc/src/functions/dpiConn.rst

@@ -607,6 +607,50 @@ handles.
     is created by this function.
 
 
+.. function:: int dpiConn_newJson(dpiConn* conn, dpiJson** json)
+
+    Returns a reference to a new JSON object. This object can be used as the
+    payload in a message enqueued in a JSON queue, or as the value for a
+    variable. The reference should be released by calling
+    :func:`dpiJson_release()` as soon as it is no longer needed.
+
+    The function returns DPI_SUCCESS for success and DPI_FAILURE for failure.
+
+    **conn** [IN] -- a reference to the connection in which the JSON object is
+    going to be used. If the reference is NULL or invalid, an error is
+    returned.
+
+    **json** [OUT] -- a pointer to a reference to the JSON object that
+    is created by this function.
+
+
+.. function:: int dpiConn_newJsonQueue(dpiConn* conn, const char* name, \
+        uint32_t nameLength, dpiQueue** queue)
+
+    Returns a reference to a new queue which enqueues and dequeues messages
+    from Advanced Queueing (AQ) with a JSON payload. The reference should be
+    released by calling :func:`dpiQueue_release()` as soon as it is no longer
+    needed. For queues with RAW or Database Object payloads, use the method
+    :func:`dpiConn_newQueue()` instead.
+
+    The function returns DPI_SUCCESS for success and DPI_FAILURE for failure.
+
+    **conn** [IN] -- a reference to the connection in which messages are to be
+    dequeued or enqueued. If the reference is NULL or invalid, an error is
+    returned.
+
+    **name** [IN] -- the name of the queue, as a byte string in the encoding
+    used for CHAR data. Note that UTF-16 encodings are not currently supported
+    by AQ.
+
+    **nameLength** [IN] -- the length of the name parameter, in bytes.
+
+    **queue** [OUT] -- a reference to the newly created queue which will be
+    populated upon successful completion of this function. The reference should
+    be released by calling :func:`dpiQueue_release()` as soon as it is no
+    longer needed.
+
+
 .. function:: int dpiConn_newMsgProps(dpiConn* conn, dpiMsgProps** props)
 
     Returns a reference to a new set of message properties, used in enqueuing

doc/src/functions/dpiMsgProps.rst

@@ -195,7 +195,7 @@ releasing the last reference by calling the function
 
     Returns the payload associated with the message properties. The payload can
     either be an object or an arbitrary series of bytes and is available after
-    the a call to :func:`dpiQueue_deqOne()` or :func:`dpiQueue_deqMany()`.
+    a call to :func:`dpiQueue_deqOne()` or :func:`dpiQueue_deqMany()`.
 
     The function returns DPI_SUCCESS for success and DPI_FAILURE for failure.
 
@@ -218,6 +218,25 @@ releasing the last reference by calling the function
     also be NULL.
 
 
+.. function:: int dpiMsgProps_getPayloadJson(dpiMsgProps* props, \
+        dpiJson** json)
+
+    Returns the payload associated with the message properties, The payload
+    must be a JSON object and is available after the call to
+    :func:`dpiQueue_deqOne()` or :func:`dpiQueue_deqMany()`; otherwise, the
+    value NULL will be returned.
+
+    The function returns DPI_SUCCESS for success and DPI_FAILURE for failure.
+
+    **props** [IN] -- a reference to the message properties from which the
+    payload is to be retrieved. If the reference is NULL or invalid, an error
+    is returned.
+
+    **json** [OUT] -- a reference to a JSON object which will be populated upon
+    successful completion of this function. If the payload is not a JSON
+    object, this value will be NULL.
+
+
 .. function:: int dpiMsgProps_getPriority(dpiMsgProps* props, int32_t* value)
 
     Returns the priority assigned to the message. See function
@@ -376,6 +395,22 @@ releasing the last reference by calling the function
     **valueLength** [IN] -- the length of the value parameter in bytes.
 
 
+.. function:: int dpiMsgProps_setPayloadJson(dpiMsgProps* props, dpiJson* json)
+
+    Sets the payload for the message as a JSON object. This value will be used
+    when the message is enqueued using :func:`dpiQueue_enqOne()` or
+    :func:`dpiQueue_enqMany()`.
+
+    The function returns DPI_SUCCESS for success and DPI_FAILURE for failure.
+
+    **props** [IN] -- a reference to the message properties on which the
+    payload is to be set. If the reference is NULL or invalid, an error is
+    returned.
+
+    **json** [IN] -- a reference to the JSON object that will be used as the
+    message payload. If the reference is NULL or invalid, an error is returned.
+
+
 .. function:: int dpiMsgProps_setPayloadObject(dpiMsgProps* props, \
         dpiObject* obj)
 

doc/src/releasenotes.rst

@@ -4,8 +4,13 @@ ODPI-C Release notes
 Version 4.5 (TBD)
 -----------------
 
-#) Added support for OAuth token authentication when creating standalone
-   connections and connection pools.
+#)  Added support for OAuth token authentication when creating standalone
+    connections and connection pools.
+#)  Added support for enqueuing and dequeuing JSON payloads using Advanced
+    Queuing (AQ). Functions :func:`dpiConn_newJson()`,
+    :func:`dpiConn_newJsonQueue()`, :func:`dpiMsgProps_getPayloadJson()` and
+    :func:`dpiMsgProps_setPayloadJson()` were added to support this
+    functionality.
 
 
 Version 4.4.1 (June 14, 2022)

doc/src/user_guide/round_trips.rst

@@ -109,6 +109,12 @@ back.
     * - :func:`dpiConn_newEnqOptions()`
       - No
       -
+    * - :func:`dpiConn_newJson()`
+      - No
+      -
+    * - :func:`dpiConn_newJsonQueue()`
+      - No
+      -
     * - :func:`dpiConn_newMsgProps()`
       - No
       -
@@ -522,6 +528,9 @@ back.
     * - :func:`dpiMsgProps_getPayload()`
       - No
       -
+    * - :func:`dpiMsgProps_getPayloadJson()`
+      - No
+      -
     * - :func:`dpiMsgProps_getPriority()`
       - No
       -
@@ -552,6 +561,9 @@ back.
     * - :func:`dpiMsgProps_setPayloadBytes()`
       - No
       -
+    * - :func:`dpiMsgProps_setPayloadJson()`
+      - No
+      -
     * - :func:`dpiMsgProps_setPayloadObject()`
       - No
       -

include/dpi.h

@@ -1005,6 +1005,13 @@ DPI_EXPORT int dpiConn_newDeqOptions(dpiConn *conn, dpiDeqOptions **options);
 // create a new enqueue options object and return it
 DPI_EXPORT int dpiConn_newEnqOptions(dpiConn *conn, dpiEnqOptions **options);
 
+// create a new, empty JSON
+DPI_EXPORT int dpiConn_newJson(dpiConn *conn, dpiJson **json);
+
+// create a new AQ queue with JSON payload
+DPI_EXPORT int dpiConn_newJsonQueue(dpiConn *conn, const char *name,
+        uint32_t nameLength, dpiQueue **queue);
+
 // create a new message properties object and return it
 DPI_EXPORT int dpiConn_newMsgProps(dpiConn *conn, dpiMsgProps **props);
 
@@ -1470,6 +1477,9 @@ DPI_EXPORT int dpiMsgProps_getOriginalMsgId(dpiMsgProps *props,
 DPI_EXPORT int dpiMsgProps_getPayload(dpiMsgProps *props, dpiObject **obj,
         const char **value, uint32_t *valueLength);
 
+// return the payload of the message (JSON)
+DPI_EXPORT int dpiMsgProps_getPayloadJson(dpiMsgProps *props, dpiJson **json);
+
 // return the priority of the message
 DPI_EXPORT int dpiMsgProps_getPriority(dpiMsgProps *props, int32_t *value);
 
@@ -1502,6 +1512,9 @@ DPI_EXPORT int dpiMsgProps_setOriginalMsgId(dpiMsgProps *props,
 DPI_EXPORT int dpiMsgProps_setPayloadBytes(dpiMsgProps *props,
         const char *value, uint32_t valueLength);
 
+// set the payload of the message (as JSON)
+DPI_EXPORT int dpiMsgProps_setPayloadJson(dpiMsgProps *props, dpiJson *json);
+
 // set the payload of the message (as an object)
 DPI_EXPORT int dpiMsgProps_setPayloadObject(dpiMsgProps *props,
         dpiObject *obj);

src/dpiConn.c

@@ -722,10 +722,24 @@ static int dpiConn__getHandles(dpiConn *conn, dpiError *error)
 }
 
 
+//-----------------------------------------------------------------------------
+// dpiConn__getJsonTDO() [INTERNAL]
+//   Internal method used for ensuring that the JSON TDO has been cached on the
+// connection.
+//-----------------------------------------------------------------------------
+int dpiConn__getJsonTDO(dpiConn *conn, dpiError *error)
+{
+    if (conn->jsonTDO)
+        return DPI_SUCCESS;
+    return dpiOci__typeByName(conn, "SYS", 3, "JSON", 4, &conn->jsonTDO,
+            error);
+}
+
+
 //-----------------------------------------------------------------------------
 // dpiConn__getRawTDO() [INTERNAL]
 //   Internal method used for ensuring that the RAW TDO has been cached on the
-//connection.
+// connection.
 //-----------------------------------------------------------------------------
 int dpiConn__getRawTDO(dpiConn *conn, dpiError *error)
 {
@@ -2126,6 +2140,43 @@ int dpiConn_newEnqOptions(dpiConn *conn, dpiEnqOptions **options)
 }
 
 
+//-----------------------------------------------------------------------------
+// dpiConn_newJson() [PUBLIC]
+//   Create a new JSON object and return it.
+//-----------------------------------------------------------------------------
+int dpiConn_newJson(dpiConn *conn, dpiJson **json)
+{
+    dpiError error;
+    int status;
+
+    if (dpiConn__check(conn, __func__, &error) < 0)
+        return dpiGen__endPublicFn(conn, DPI_FAILURE, &error);
+    DPI_CHECK_PTR_NOT_NULL(conn, json);
+    status = dpiJson__allocate(conn, json, &error);
+    return dpiGen__endPublicFn(conn, status, &error);
+}
+
+
+//-----------------------------------------------------------------------------
+// dpiConn_newJsonQueue() [PUBLIC]
+//   Create a new AQ queue object with JSON payload and return it.
+//-----------------------------------------------------------------------------
+int dpiConn_newJsonQueue(dpiConn *conn, const char *name, uint32_t nameLength,
+        dpiQueue **queue)
+{
+    dpiError error;
+    int status;
+
+    if (dpiConn__check(conn, __func__, &error) < 0)
+        return dpiGen__endPublicFn(conn, DPI_FAILURE, &error);
+    DPI_CHECK_PTR_AND_LENGTH(conn, name)
+    DPI_CHECK_PTR_NOT_NULL(conn, queue)
+    status = dpiQueue__allocate(conn, name, nameLength, NULL, queue, 1,
+            &error);
+    return dpiGen__endPublicFn(conn, status, &error);
+}
+
+
 //-----------------------------------------------------------------------------
 // dpiConn_newTempLob() [PUBLIC]
 //   Create a new temporary LOB and return it.
@@ -2194,7 +2245,7 @@ int dpiConn_newQueue(dpiConn *conn, const char *name, uint32_t nameLength,
     DPI_CHECK_PTR_AND_LENGTH(conn, name)
     DPI_CHECK_PTR_NOT_NULL(conn, queue)
     status = dpiQueue__allocate(conn, name, nameLength, payloadType, queue,
-            &error);
+            0, &error);
     return dpiGen__endPublicFn(conn, status, &error);
 }
 

src/dpiImpl.h

@@ -1245,8 +1245,8 @@ typedef struct {
     dpiMsgProps **props;                // array of dpiMsgProps handles
     void **handles;                     // array of OCI msg prop handles
     void **instances;                   // array of instances
-    void **indicators;                  // array of indicators
-    int16_t *rawIndicators;             // array of indicators (RAW queues)
+    void **indicators;                  // array of indicator pointers
+    int16_t *scalarIndicators;          // array of scalar indicator buffers
     void **msgIds;                      // array of OCI message ids
 } dpiQueueBuffer;
 
@@ -1290,6 +1290,7 @@ struct dpiConn {
     const char *releaseString;          // cached release string or NULL
     uint32_t releaseStringLength;       // cached release string length or 0
     void *rawTDO;                       // cached RAW TDO
+    void *jsonTDO;                      // cached JSON TDO
     dpiVersionInfo versionInfo;         // Oracle database version info
     uint32_t commitMode;                // commit mode (for two-phase commits)
     uint16_t charsetId;                 // database character set ID
@@ -1495,6 +1496,7 @@ struct dpiMsgProps {
     void *handle;                       // OCI message properties handle
     dpiObject *payloadObj;              // payload (object)
     void *payloadRaw;                   // payload (RAW)
+    dpiJson *payloadJson;               // payload (JSON)
     void *msgIdRaw;                     // message ID (RAW)
 };
 
@@ -1555,6 +1557,7 @@ struct dpiQueue {
     dpiDeqOptions *deqOptions;          // dequeue options
     dpiEnqOptions *enqOptions;          // enqueue options
     dpiQueueBuffer buffer;              // buffer area
+    int isJson;                         // is JSON payload?
 };
 
 
@@ -1685,6 +1688,7 @@ int dpiConn__create(dpiConn *conn, const dpiContext *context,
         const dpiCommonCreateParams *commonParams,
         dpiConnCreateParams *createParams, dpiError *error);
 void dpiConn__free(dpiConn *conn, dpiError *error);
+int dpiConn__getJsonTDO(dpiConn *conn, dpiError *error);
 int dpiConn__getRawTDO(dpiConn *conn, dpiError *error);
 int dpiConn__getServerVersion(dpiConn *conn, int wantReleaseString,
         dpiError *error);
@@ -1861,7 +1865,8 @@ void dpiSodaDocCursor__free(dpiSodaDocCursor *cursor, dpiError *error);
 // definition of internal dpiQueue methods
 //-----------------------------------------------------------------------------
 int dpiQueue__allocate(dpiConn *conn, const char *name, uint32_t nameLength,
-        dpiObjectType *payloadType, dpiQueue **queue, dpiError *error);
+        dpiObjectType *payloadType, dpiQueue **queue, int isJson,
+        dpiError *error);
 void dpiQueue__free(dpiQueue *queue, dpiError *error);
 
 

src/dpiMsgProps.c

@@ -360,6 +360,23 @@ int dpiMsgProps_getPayload(dpiMsgProps *props, dpiObject **obj,
 }
 
 
+//-----------------------------------------------------------------------------
+// dpiMsgProps_getPayloadJson() [PUBLIC]
+//   Get the JSON payload for the message (as a dpiJson).
+//-----------------------------------------------------------------------------
+int dpiMsgProps_getPayloadJson(dpiMsgProps *props, dpiJson **json)
+{
+    dpiError error;
+
+    if (dpiGen__startPublicFn(props, DPI_HTYPE_MSG_PROPS, __func__,
+            &error) < 0)
+        return dpiGen__endPublicFn(props, DPI_FAILURE, &error);
+    DPI_CHECK_PTR_NOT_NULL(props, json)
+    *json = props->payloadJson;
+    return dpiGen__endPublicFn(props, DPI_SUCCESS, &error);
+}
+
+
 //-----------------------------------------------------------------------------
 // dpiMsgProps_getPriority() [PUBLIC]
 //   Return the priority of the message.
@@ -493,6 +510,27 @@ int dpiMsgProps_setPayloadBytes(dpiMsgProps *props, const char *value,
 }
 
 
+//-----------------------------------------------------------------------------
+// dpiMsgProps_setPayloadJson() [PUBLIC]
+//   Set the payload for the message (as a JSON object).
+//-----------------------------------------------------------------------------
+int dpiMsgProps_setPayloadJson(dpiMsgProps *props, dpiJson *json)
+{
+    dpiError error;
+    if (dpiGen__startPublicFn(props, DPI_HTYPE_MSG_PROPS, __func__,
+            &error) < 0)
+        return dpiGen__endPublicFn(props, DPI_FAILURE, &error);
+    if (dpiGen__checkHandle(json, DPI_HTYPE_JSON, "check json object",
+            &error) < 0)
+        return dpiGen__endPublicFn(props, DPI_FAILURE, &error);
+    if (props->payloadJson)
+        dpiGen__setRefCount(props->payloadJson, &error, -1);
+    dpiGen__setRefCount(json, &error, 1);
+    props->payloadJson = json;
+    return dpiGen__endPublicFn(props, DPI_SUCCESS, &error);
+}
+
+
 //-----------------------------------------------------------------------------
 // dpiMsgProps_setPayloadObject() [PUBLIC]
 //   Set the payload for the message (as an object).