Streams Advanced Queuing and Publish-Subscribe Functions

This section describes the Streams Advanced Queuing and publish-subscribe functions.

See Also:

"OCI Demonstration Programs" for Streams Advanced Queueing programs

Table 16-7 Advanced Queuing and Publish-Subscribe Functions

Function	Purpose
OCIAQDeq()	Advanced Queuing dequeue
OCIAQDeqArray()	Dequeue an array of messages
OCIAQEnq()	Advanced Queuing enqueue
OCIAQEnqArray()	Enqueue an array of messages
OCIAQListen()	Listen on one or more queues on behalf of a list of agents
OCIAQListen2()	Listen on one or more queues on behalf of a list of agents. Supports buffered messaging and persistent queues.
OCISubscriptionEnable()	Enable notifications on a subscription
OCISubscriptionPost()	Post to a subscription to receive notifications
OCISubscriptionRegister()	Register a subscription
OCISubscriptionUnRegister()	Unregister a subscription

OCIAQDeq()

Purpose

This call is used for a Streams Advanced Queuing dequeue operation using the OCI.

Syntax

sword OCIAQDeq ( OCISvcCtx           *svch,
                 OCIError            *errh,
                 text                *queue_name,
                 OCIAQDeqOptions     *dequeue_options,
                 OCIAQMsgProperties  *message_properties,
                 OCIType             *payload_tdo,
                 dvoid               **payload,
                 dvoid               **payload_ind,
                 OCIRaw              **msgid,
                 ub4                 flags );

Parameters

svch (IN)

OCI service context.

errh (IN)

An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.

queue_name (IN)

The target queue for the dequeue operation.

dequeue_options (IN)

The options for the dequeue operation; stored in an OCIAQDeqOptions descriptor, with OCI type constant OCI_DTYPE_AQDEQ_OPTIONS.

OCI_DTYPE_AQDEQ_OPTIONS has the additional attribute OCI_ATTR_MSG_DELIVERY_MODE (introduced in Oracle Database 10g Release 2) with values:

OCI_MSG_PERSISTENT (default)
OCI_MSG_BUFFERED
OCI_MSG_PERSISTENT_OR_BUFFERED

message_properties (OUT)

The message properties for the message; stored in an OCIAQMsgProperties descriptor, with OCI type constant OCI_DTYPE_AQMSG_PROPERTIES which can have values:

OCI_AQ_PERSISTENT (default)
OCI_AQ_BUFFERED

payload_tdo (IN)

The TDO (type descriptor object) of an object type. For a raw queue, this parameter should point to the TDO of SYS.RAW.

payload (IN/OUT)

A pointer to a pointer to a program variable buffer that is an instance of an object type. For a raw queue, this parameter should point to an instance of OCIRaw.

Memory for the payload is dynamically allocated in the object cache. The application can optionally call OCIObjectFree() to deallocate the payload instance when it is no longer needed. If the pointer to the program variable buffer (*payload) is passed as NULL, the buffer is implicitly allocated in the cache.

The application may choose to pass NULL for payload the first time OCIAQDeq() is called, and let the OCI allocate the memory for the payload. It can then use a pointer to that previously allocated memory in subsequent calls to OCIAQDeq().

To obtain a TDO for the payload, use OCITypeByName(), or OCITypeByRef().

The OCI provides functions which allow the user to set attributes of the payload, such as its text. For information about setting these attributes, refer to "Manipulating Object Attributes".

payload_ind (IN/OUT)

A pointer to a pointer to the program variable buffer containing the parallel indicator structure for the object type.

The memory allocation rules for payload_ind are the same as those for payload,.

msgid (OUT)

The message ID.

flags (IN)

Not currently used; pass as OCI_DEFAULT.

Comments

Users must have the AQ_USER_ROLE or privileges to execute the DBMS_AQ package in order to use this call. The OCI environment must be initialized in object mode (using OCIInitialize()) to use this call.

See Also:

For more information about OCI and Advanced Queuing, refer to "OCI and Streams Advanced Queuing"
For additional information about Advanced Queuing, refer to Oracle Streams Advanced Queuing User's Guide and Reference

Examples

For code examples, refer to the description of OCIAQEnq().

Related Functions

OCIAQEnq(), OCIAQListen2(), OCIInitialize()

OCIAQDeqArray()

Purpose

This call dequeues an array of messages from a queue. The array of messages is all dequeued with the same option and has the same queue table payload column TDO.

Syntax

sword OCIAQDeqArray ( OCISvcCtx           *svchp,
                      OCIError            *errhp,
                      OraText             *queue_name,
                      OCIAQDeqOptions     *deqopt,
                      ub4                 *iters,
                      OCIAQMsgProperties  **msgprop,
                      OCIType             *payload_tdo,
                      dvoid               **payload,
                      dvoid               **payload_ind,
                      OCIRaw              **msgid,
                      dvoid               *ctxp,
                      OCICallbackAQDeq    (cbfp) 
                                          (
                                          dvoid        *ctxp,
                                          dvoid        **payload,
                                          dvoid        **payload_ind
                                          ),
                      ub4                 flags );

Parameters

svchp (IN)

OCI service context (unchanged from OCIAQDeq()).

errhp (IN)

An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error (unchanged from OCIAQDeq()).

queue_name (IN)

The name of the queue from which messages are dequeued (unchanged from OCIAQDeq()).

deqopt (IN)

A pointer to an OCIAQDeqOptions descriptor (unchanged from OCIAQDeq()).

OCI_DTYPE_AQDEQ_OPTIONS OCI type constant has the additional attribute: OCI_ATTR_MSG_DELIVERY_MODE (introduced in Oracle Database 10g Release 2) with values:

OCI_MSG_PERSISTENT (default)
OCI_MSG_BUFFERED
OCI_MSG_PERSISTENT_OR_BUFFERED

iters (IN/OUT)

On input, the number of messages to dequeue. On output, the number of messages successfully dequeued.

msgprop (OUT)

An array of pointers to OCIAQMsgProperties descriptors, of OCI type constant OCI_DTYPE_AQMSG_PROPERTIES, which can have values:

OCI_AQ_PERSISTENT (default)
OCI_AQ_BUFFERED

payload_tdo (OUT)

A pointer to the TDO of the queue table's payload column.

payload (OUT)

An array of pointers to dequeued messages.

payload_ind (OUT)

An array of pointers to indicators.

msgid (OUT)

An array of pointers to the message ID of the dequeued messages.

ctxp (IN)

The context that will be passed to the callback function.

cbfp (IN)

The callback that may be registered to provide a buffer pointer into which the dequeued message will be placed. If NULL, then messages will be dequeued into buffers pointed to by payload.

flags (IN)

Not currently used; pass as OCI_DEFAULT.

Comments

A nonzero wait time, as specified in the OCIAQDeqOptions, is recognized only when there are no messages in the queue. If the queue contains messages that are eligible for dequeue, then the OCIAQDeqArray() function will dequeue up to iters messages and return immediately.

This function is not supported in nonblocking mode.

See Also:

For more information about OCI and Advanced Queuing, refer to "OCI and Streams Advanced Queuing"
For additional information about Advanced Queuing, refer to Oracle Streams Advanced Queuing User's Guide and Reference

Related Functions

OCIAQDeq(), OCIAQEnqArray(), OCIAQListen2(), OCIInitialize()

OCIAQEnq()

Purpose

This call is used for a Streams Advanced Queuing enqueue.

Syntax

sword OCIAQEnq ( OCISvcCtx           *svch,
                 OCIError            *errh,
                 text                *queue_name,
                 OCIAQEnqOptions     *enqueue_options,
                 OCIAQMsgProperties  *message_properties,
                 OCIType             *payload_tdo,
                 dvoid               **payload,
                 dvoid               **payload_ind,
                 OCIRaw              **msgid,
                 ub4                 flags );

Parameters

svch (IN)

OCI service context.

errh (IN)

An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.

queue_name (IN)

The target queue for the enqueue operation.

enqueue_options (IN)

The options for the enqueue operation; stored in an OCIAQEnqOptions descriptor.

message_properties (IN)

The message properties for the message; stored in an OCIAQMsgProperties descriptor, of OCI type constant OCI_DTYPE_AQMSG_PROPERTIES, introduced in Oracle Database 10g Release 2.

Descriptor OCI_DTYPE_AQMSG_PROPERTIES has the attribute, OCI_ATTR_MSG_DELIVERY_MODE, which has values:

OCI_MSG_PERSISTENT (default)
OCI_MSG_BUFFERED

payload_tdo (IN)

The TDO (type descriptor object) of an object type. For a raw queue, this parameter should point to the TDO of SYS.RAW.

payload (IN)

A pointer to a pointer to an instance of an object type. For a raw queue, this parameter should point to an instance of OCIRaw.

The OCI provides functions which allow the user to set attributes of the payload, such as its text.

See Also:

For information about setting these attributes, refer to "Manipulating Object Attributes"

payload_ind (IN)

A pointer to a pointer to the program variable buffer containing the parallel indicator structure for the object type.

msgid (OUT)

The message ID.

flags (IN)

Not currently used; pass as OCI_DEFAULT.

Comments

Users must have the AQ_USER_ROLE or privileges to execute the DBMS_AQ package in order to use this call.

The OCI environment must be initialized in object mode (using OCIInitialize()) to use this call.

See Also:

For more information about OCI and Advanced Queuing, refer to "OCI and Streams Advanced Queuing"
For more information about Advanced Queuing, refer to Oracle Streams Advanced Queuing User's Guide and Reference

To obtain a TDO for the payload, use OCITypeByName(), or OCITypeByRef().

Related Functions

OCIAQDeq(), OCIAQListen2(), OCIInitialize()

OCIAQEnqArray()

Purpose

This call enqueues an array of messages to a queue. The array of messages is enqueued with the same options and has the same payload column TDO.

Syntax

sword OCIAQEnqArray ( OCISvcCtx              *svchp,
                      OCIError               *errhp,
                      OraText                *queue_name,
                      OCIAQEnqOptions        *enqopt,
                      ub4                    *iters,
                      OCIAQMsgProperties     **msgprop,
                      OCIType                *payload_tdo,
                      dvoid                  **payload,
                      dvoid                  **payload_ind,
                      OCIRaw                 **msgid,
                      dvoid                  *ctxp,
                      OCICallbackAQEnq       (cbfp)
                                             (
                                             dvoid       *ctxp,
                                             dvoid       **payload,
                                             dvoid       **payload_ind
                                             ),
                      ub4                    flags );

Parameters

svchp (IN)

The service context (unchanged from OCIAQEnq()).

errhp (IN/OUT)

The error handle (unchanged from OCIAQEnq()).

queue_name (IN)

The name of the queue in which messages are enqueued (unchanged from OCIAQEnq()).

enqopt (IN)

A pointer to an OCIAQEnqOptions descriptor (unchanged from OCIAQEnq()).

iters (IN/OUT)

On input, the number of messages to enqueue. On output, the number of messages successfully enqueued.

msgprop (IN)

An array of pointers to OCIAQMsgProperties descriptors, of OCI type constant OCI_DTYPE_AQMSG_PROPERTIES, introduced in Oracle Database 10g Release 2.

OCI_DTYPE_AQMSG_PROPERTIES has the attribute, OCI_ATTR_MSG_DELIVERY_MODE, which has values:

OCI_MSG_PERSISTENT (default)
OCI_MSG_BUFFERED

payload_tdo (IN)

A pointer to the TDO of the queue table's payload column.

payload (IN)

An array of pointers to messages to be enqueued.

payload_ind (IN)

An array of pointers to indicators, or a NULL pointer if indicator variables are not used.

msgid (OUT)

An array of pointers to the message ID of the enqueued messages or a NULL pointer if no message IDs are returned.

ctxp (IN)

The context that will be passed to the registered callback function.

cbfp (IN)

A callback that may be registered to provide messages dynamically. If NULL, then all messages must be materialized prior to calling OCIAQEnqArray().

flags (IN)

Not currently used; pass as OCI_DEFAULT.

Comments

This function is not supported in nonblocking mode.

Related Functions

OCIAQEnq(), OCIAQDeqArray(), OCIAQListen2(), OCIInitialize()

OCIAQListen()

Purpose

Listens on one or more queues on behalf of a list of agents.

Syntax

sword OCIAQListen (OCISvcCtx      *svchp, 
                   OCIError       *errhp,
                   OCIAQAgent     **agent_list, 
                   ub4            num_agents,
                   sb4            wait, 
                   OCIAQAgent     **agent,
                   ub4            flags);

Parameters

svchpp (IN/OUT): The service context handle.
errhp (IN/OUT): An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.
agent_list (IN): List of agents for which to monitor messages.
num_agents (IN): Number of agents in the agent list.
wait (IN): Timeout interval for the listen call.
agent (OUT): Agent for which there is a message. OCIAgent is an OCI descriptor.
flags (IN): Not currently used; pass as OCI_DEFAULT.

Comments

This is a blocking call that returns when there is a message ready for consumption for an agent in the list. If there are no messages found when the wait time expires, an error is returned.

Related Functions

OCIAQEnq(), OCIAQDeq(), OCIAQListen2(), OCISvcCtxToLda(), OCISubscriptionEnable(), OCISubscriptionPost(), OCISubscriptionRegister(),OCISubscriptionUnRegister()

OCIAQListen2()

Purpose

Listens on one or more queues on behalf of a list of agents. Supports buffered messaging as well as persistent queues. Introduced in Oracle Database 10g Release 2.

Syntax

sword OCIAQListen2 (OCISvcCtx         *svchp, 
                    OCIError          *errhp,
                    OCIAQAgent        **agent_list, 
                    ub4               num_agents,
                    OCIAQListenOpts   *lopts, 
                    OCIAQAgent        **agent,
                    OCIAQLisMsgProps  *lmops,
                    ub4               flags);

Parameters

svchpp (IN/OUT)

The service context handle.

errhp (IN/OUT)

An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.

agent_list (IN)

List of agents for which to monitor messages.

num_agents (IN)

Number of agents in the agent list.

lopts (IN)

Type constant OCI_DTYPE_AQLIS_OPTIONS has the following attributes:

OCI_AQ_WAIT - maximum wait time, in seconds, for the listen call
OCI_MSG_DELIVERY_MODE - has one of the following values:
- OCI_MSG_PERSISTENT
- OCI_MSG_BUFFERED
- OCI_MSG_PERSISTENT_OR_BUFFERED

agent (OUT)

Agent for which there is a message. OCIAgent is an OCI descriptor.

lmops (OUT)

OCI_DTYPE_AQLIS_MSG_PROPERTIES (listen message properties) has one attribute, OCI_MSG_DELIVERY_MODE, which has the following values:

OCI_MSG_PERSISTENT
OCI_MSG_BUFFERED

flags (IN)

Not currently used; pass as OCI_DEFAULT.

Comments

This is a blocking call that returns when there is a message ready for consumption for an agent in the list. If there are no messages found when the wait time expires, an error is returned.

Related Functions

OCIAQEnq(), OCIAQDeq(), OCISvcCtxToLda(), OCISubscriptionEnable(), OCISubscriptionPost(), OCISubscriptionRegister(),OCISubscriptionUnRegister()

OCISubscriptionDisable()

Purpose

Disables a subscription registration which turns off all notifications.

Syntax

ub4 OCISubscriptionDisable ( OCISubscription   *subscrhp,
                             OCIError          *errhp
                             ub4               mode );

Parameters

subscrhp (IN): A subscription handle with the OCI_ATTR_SUBSCR_NAME and OCI_ATTR_SUBSCR_NAMESPACE attributes set.

See Also:
Subscription Handle Attributes
errhp (OUT): An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.
mode (IN): Call-specific mode. Valid values:

OCI_DEFAULT - executes the default call which discards all notifications on this subscription until the subscription is enabled

Comments

This call is used to temporarily turn off notifications. This is useful when the application is running a critical section of the code and should not be interrupted.

The user need not be connected or authenticated to perform this operation. A registration must have been performed to the subscription specified by the subscription handle before this call is made.

All notifications subsequent to an OCISubscriptionDisable() are discarded by the system until an OCISubscriptionEnable() is performed.

Related Functions

OCIAQListen2(), OCISubscriptionEnable(), OCISubscriptionPost(), OCISubscriptionRegister(), OCISubscriptionUnRegister()

OCISubscriptionEnable()

Purpose

Enables a subscription registration that has been disabled. This turns on all notifications.

Syntax

ub4 OCISubscriptionEnable ( OCISubscription   *subscrhp,
                            OCIError          *errhp
                            ub4               mode );

Parameters

subscrhp (IN): A subscription handle with the OCI_ATTR_SUBSCR_NAME and OCI_ATTR_SUBSCR_NAMESPACE attributes set.

See Also:
For information, see Subscription Handle Attributes
errhp (OUT): An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.
mode (IN): Call-specific mode. Valid value:

OCI_DEFAULT - executes the default call which buffers all notifications on this subscription until a subsequent enable is performed

Comments

This call is used to turn on notifications after a subscription registration has been disabled.

The user need not be connected or authenticated to perform this operation. A registration must have been done for the specified subscription before this call is made.

Related Functions

OCIAQListen2(), OCISvcCtxToLda(), OCISubscriptionPost(), OCISubscriptionRegister(), OCISubscriptionUnRegister()

OCISubscriptionPost()

Purpose

Posts to a subscription which allows all clients who are registered for the subscription to get notifications.

Syntax

ub4 OCISubscriptionPost ( OCISvcCtx         *svchp,
                          OCISubscription   **subscrhpp,
                          ub2               count,
                          OCIError          *errhp
                          ub4               mode );

Parameters

svchp (IN)

An OCI service context (after release 7). This service context should have a valid authenticated user handle.

subscrhpp (IN)

An array of subscription handles. Each element of this array should be a subscription handle with the OCI_ATTR_SUBSCR_NAME and OCI_ATTR_SUBSCR_NAMESPACE attributes set.

See Also:

For information, see Subscription Handle Attributes

The OCI_ATTR_SUBSCR_PAYLOAD attribute has to be set for each subscription handle prior to this call. If it is not set, the payload is assumed to be NULL and no payload is delivered when the notification is received by the clients that have registered interest. Note that the caller will have to preserve the payload until the post is done as the OCIAttrSet() call keeps track of the reference to the payload but does not copy the contents.

count (IN)

The number of elements in the subscription handle array.

errhp (OUT)

An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.

mode (IN)

Call-specific mode. Valid value:

OCI_DEFAULT - executes the default call

Comments

Posting to a subscription involves identifying the subscription name and the payload if desired. If no payload is associated, the payload length can be set to 0.

This call provides a best-effort guarantee. A notification goes to registered clients at most once.

This call is primarily used for nonpersistent notification and is useful in the case of several system events. If the application needs more rigid guarantees, it can use the Advanced Queuing functionality by enqueuing to queue.

Related Functions

OCIAQListen2(), OCISvcCtxToLda(), OCISubscriptionEnable(), OCISubscriptionRegister(), OCISubscriptionUnRegister()

OCISubscriptionRegister()

Purpose

Registers a callback for message notification.

Syntax

ub4 OCISubscriptionRegister ( OCISvcCtx         *svchp,
                              OCISubscription   **subscrhpp,
                              ub2               count,
                              OCIError          *errhp
                              ub4               mode );

Parameters

svchp (IN): An OCI service context (after release 7). This service context should have a valid authenticated user handle.
subscrhpp (IN): An array of subscription handles. Each element of this array should be a subscription handle with all of the following attributes set:

OCI_ATTR_SUBSCR_NAME,
OCI_ATTR_SUBSCR_NAMESPACE,
OCI_ATTR_SUBSCR_RECPTPROTO.

Otherwise, an error will be returned.

One of attributes

OCI_ATTR_SUBSCR_CBACK,
OCI_ATTR_SUBSCR_CTX,
OCI_ATTR_SUBSCR_RECPT,

must also be set.

See Also:

For information about the handle attributes, see Subscription Handle Attributes

When a notification is received for the registration denoted by subscrhpp[i], either the user-defined callback function (OCI_ATTR_SUBSCR_CBACK) set for subscrhpp[i] will be invoked with the context (OCI_ATTR_SUBSCR_CTX) set for subscrhpp[i], or an e-mail will be sent to (OCI_ATTR_SUBSCR_RECPT) set for subscrhpp[i], or the PL/SQL procedure (OCI_ATTR_SUBSCR_RECPT) set for subscrhpp[i], will be invoked in the database, provided the subscriber of subscrhpp[i] has the appropriate permissions on the procedure.

count (IN): The number of elements in the subscription handle array.
errhp (OUT): An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.
mode (IN): Call-specific mode. Valid values:

OCI_DEFAULT - executes the default call which specifies that the registration is treated as disconnected
OCI_NOTIFY_CONNECTED - notifications are received only if the client is connected (not supported in this release)

Whenever a new client process comes up, or an old one goes down and comes back up, it needs to register for all subscriptions of interest. If the client stays up and the server first goes down and then comes back up, the client will continue to receive notifications for registrations that are DISCONNECTED. However, the client will not receive notifications for CONNECTED registrations as they will be lost once the server goes down and comes back up.

Comments

This call is invoked for registration to a subscription which identifies the subscription name of interest and the associated callback to be invoked. Interest in several subscriptions can be registered at one time.

This interface is only valid for the asynchronous mode of message delivery. In this mode, a subscriber issues a registration call which specifies a callback. When messages are received that match the subscription criteria, the callback is invoked. The callback may then issue an explicit message_receive (dequeue) to retrieve the message.

The user must specify a subscription handle at registration time with the namespace attribute set to OCI_SUBSCR_NAMESPACE_AQ.

The subscription name is the string SCHEMA.QUEUE if the registration is for a single consumer queue and SCHEMA.QUEUE:CONSUMER_NAME if the registration is for a multi-consumer queue. The string should be in upper case.

Each namespace will have its own privilege model. If the user performing the register is not entitled to register in the namespace for the specified subscription, an error is returned.

Related Functions

OCIAQListen2(), OCISvcCtxToLda(), OCISubscriptionEnable(), OCISubscriptionPost(), OCISubscriptionUnRegister()

OCISubscriptionUnRegister()

Purpose

Unregisters a subscription which turns off notifications.

Syntax

ub4 OCISubscriptionUnRegister ( OCISvcCtx         *svchp,
                                OCISubscription   *subscrhp,
                                OCIError          *errhp
                                ub4               mode );

Parameters

svchp (IN): An OCI service context (after release 7). This service context should have a valid authenticated user handle.
subscrhp (IN): A subscription handle with the OCI_ATTR_SUBSCR_NAME and OCI_ATTR_SUBSCR_NAMESPACE attributes set.

See Also:
For information, see Subscription Handle Attributes
errhp (OUT): An error handle you can pass to OCIErrorGet() for diagnostic information in the event of an error.
mode (IN): Call-specific mode. Valid value:

OCI_DEFAULT - executes the default call

Comments

Unregistering to a subscription is going to ensure that the user will not receive notifications regarding the specified subscription in future. If the user wishes to resume notification, then the only option is to re-register to the subscription.

All notifications that would otherwise have been delivered are not delivered after a subsequent register is performed because the user is no longer in the list of interested clients.

Related Functions

OCIAQListen2(), OCISvcCtxToLda(), OCISubscriptionEnable(), OCISubscriptionPost(), OCISubscriptionRegister().