An OracleAQQueue object represents a queue.
System.Object
OracleAQQueue
// C# public class OracleAQQueue : IDisposable
| Provider | ODP.NET, Unmanaged Driver |
| Assembly | Oracle.DataAccess.dll |
| Namespace | Oracle.DataAccess.Client |
| .NET Framework | 3.5, 4.0, 4.5 |
All public static methods are thread-safe, although instance methods do not guarantee thread safety.
A queue is a repository of messages and may either be a user queue, or an exception queue. A user queue is for normal message processing. A message is moved from a user queue to an exception queue if it cannot be retrieved and processed for some reason.
See Also:
OracleAQQueue members are listed in the following tables.
OracleAQQueue constructors are listed in Table 12-20.
Table 12-20 OracleAQQueue Constructors
| Constructor | Description |
|---|---|
|
Instantiate a new instance of the |
The OracleAQQueue static method is listed in Table 12-21.
Table 12-21 OracleAQQueue Static Methods
| Static Method | Description |
|---|---|
|
Listens for messages on one or more queues for one or more consumers, as specified in the array of |
OracleAQQueue properties are listed in Table 12-22.
Table 12-22 OracleAQQueue Properties
| Property | Description |
|---|---|
|
Specifies the |
|
|
Specifies the dequeueing options to use when dequeuing a message from a queue. |
|
|
Specifies the enqueueing options used to enqueue a message to a queue. |
|
|
Specifies the type of queue table associated with this queue. |
|
|
Returns the name of the queue. |
|
|
Specifies the various notification options for notifications that are registered using the |
|
|
Specifies the array of consumers, for a multiconsumer queue, that are to be notified asynchronously for any incoming messages on the queue. |
|
|
Specifies the type name on which the queue and the corresponding queue table is based if the |
The OracleAQQueue public methods are listed in Table 12-23.
Table 12-23 OracleAQQueue Public Methods
| Public Method | Description |
|---|---|
|
Dequeues messages from queues (Overloaded). |
|
|
Dequeues multiple messages from queues (Overloaded). |
|
|
Releases any resources or memory allocated by the object |
|
|
Enqueues messages to queues (Overloaded). |
|
|
Enqueues multiple messages to a queue (Overloaded). |
|
|
Listens for messages on the queue on behalf of |
The OracleAQQueue event is listed in Table 12-24.
OracleAQQueue constructors create new instances of the OracleAQQueue class.
This constructor takes a queue name to initialize a queue object.
OracleAQQueue(string, OracleConnection)
This constructor takes a queue name and connection to initialize a queue object. The connection does not need be open during the queue object construction.
OracleAQQueue(string, OracleConnection, OracleAQMessageType)
This constructor takes a queue name, connection, and message type enumeration to initialize a queue object.
OracleAQQueue(string, OracleConnection, OracleAQMessageType, string)
This constructor takes a queue name, connection, message type enumeration, and UDT type name to initialize a queue object.
See Also:
This constructor takes a queue name to initialize a queue object.
// C#
public OracleAQQueue(string name);
name
The name of the queue as specified in the database.
ArgumentNullException - The queue name is null.
ArgumentException - The queue name is empty.
The operation of creating an OracleAQQueue object does not involve checking for the existence of the queue in the database.
This constructor takes a queue name and connection to initialize a queue object. The connection does not need to be open during the queue object construction.
// C# public OracleAQQueue(string name, OracleConnection con);
name
Name of the queue as specified in the database.
con
An OracleConnection object that connects to the queue.
ArgumentNullException - Either the connection is null or queue name is null.
ArgumentException - Queue name is empty.
The connection can be accessed using the Connection property, and it must be opened before calling any operational APIs such as Enqueue and Dequeue.
Creating an OracleAQQueue object does not check for the existence of the queue in the database.
This constructor takes a queue name, connection and message type enumeration to initialize a queue object. The connection does not need to be open during the queue object construction.
// C# public OracleAQQueue(string name, OracleConnection con, OracleAQMessageType messageType);
name
The name of the queue as specified in the database.
con
An OracleConnection object that is used to connect to the queue.
messageType
An OracleAQMessageType enumeration specifying the type of the message that is enqueued or dequeued from this queue.
ArgumentNullException - Either the connection is null or queue name is null.
ArgumentException - Queue name is empty or the specified message type is not valid.
Creating an OracleAQQueue object does not check for the existence of the queue in the database.
You need to set the UdtTypeName property before using the queue object if the messageType is a UDT. Another approach is to create a queue using the other constructor overload by supplying the udtTypeName.
This constructor takes a queue name, connection, message type enumeration, and UDT type name to initialize a queue object. The connection does not need to be open during the queue object construction.
// C# public OracleAQQueue(string name, OracleConnection con, OracleAQMessageType messageType, string udtTypeName);
name
The name of the queue as specified in the database.
con
An OracleConnection object that is used to connect to the queue.
messageType
An OracleAQMessageType enumeration specifying the type of the message that is enqueued or dequeued from this queue.
udtTypeName
The name of the database object type used if the messageType is UDT. The udtTypeName parameter represents the type on which the queue is based.
ArgumentNullException - The connection is null or the queue name is null.
ArgumentException - The queue name is empty or the specified messageType is not valid.
Creating an OracleAQQueue object does not check for the existence of the queue in the database.
OracleAQQueue static methods are listed in Table 12-25.
Table 12-25 OracleAQQueue Static Methods
| Static Method | Description |
|---|---|
|
Listens for messages on one or more queues for one or more consumers, as specified in the array of |
See Also:
Listen methods listen for messages on one or more queues for one or more consumers as specified in the array of OracleAQAgent objects.
Listen(OracleConnection, OracleAQAgent[ ])
This static method listens for messages on one or more queues for one or more consumers as specified in the array of OracleAQAgent objects.
Listen(OracleConnection, OracleAQAgent[ ], int)
This static method listens for messages on one or more queues for one or more consumers as specified in the array of OracleAQAgent objects. It also specifies a wait time.
This static method listens for messages on one or more queues for one or more consumers as specified in the array of OracleAQAgent objects.
// C# public static OracleAQAgent Listen(OracleConnection con, OracleAQAgent[] listenConsumers);
con
An OracleConnection instance.
listenConsumers
The array of consumers being listened for. The name of the OracleAQAgent object must be null or empty for single consumer queues.
An OracleAQAgent object.
ArgumentNullException - The con or listenConsumers parameter is null.
InvalidOperationException - The connection is not open.
Listen is useful in situations where one needs to monitor multiple queues until a message is available for a consumer in one of the queues. The Name property of the OracleAQAgent object represents the name of the consumer and the Address property represents the name of the queue.
This call blocks the calling thread until there is a message ready for consumption for a consumer in the list. It returns an OracleAQAgent object which specifies the consumer and queue for which a message is ready to be dequeued.
This static method listens for messages on one or more queues for one or more consumers as specified in the array of OracleAQAgent objects. The Name property of the OracleAQAgent object represents the name of the consumer and the Address property of the OracleAQAgent object represents the name of the queue.
In case of timeout, this method returns null.
// C# public static OracleAQAgent Listen(OracleConnection con, OracleAQAgent[] listenConsumers, int waitTime);
con
An OracleConnection instance.
listenConsumers
The array of consumers being listened for. The name of the OracleAQAgent object must be null or empty for single consumer queues.
waitTime
Wait time in seconds.
An OracleAQAgent object.
ArgumentNullException - The con or listenConsumers parameter is null.
InvalidOperationException - The connection is not open.
ArgumentException - waitTime is less than -1.
Listen is useful in situations where one needs to monitor multiple queues until a message is available for a consumer in one of the queues. The Name property of the OracleAQAgent object represents the name of the consumer and the Address property of the OracleAQAgent object represents the name of the queue.
A waitTime of -1 implies an infinite wait time.
This call blocks the calling thread until there is a message ready for consumption for a consumer in the list. It returns an OracleAQAgent object which specifies the consumer and queue for which a message is ready to be dequeued.
OracleAQQueue properties are listed in Table 12-26.
Table 12-26 OracleAQQueue Properties
| Property | Description |
|---|---|
|
Specifies the |
|
|
Specifies the dequeueing options to use when dequeuing a message from a queue. |
|
|
Specifies the enqueueing options used to enqueue a message to a queue. |
|
|
Specifies the type of queue table associated with this queue. |
|
|
Returns the name of the queue. |
|
|
Specifies the various notification options for notifications that are registered using the |
|
|
Specifies the array of consumers, for a multiconsumer queue, that are to be notified asynchronously for any incoming messages on the queue. |
|
|
Specifies the type name on which the queue and the corresponding queue table is based if the |
See Also:
This property specifies the OracleConnection object associated with the queue.
// C#
public OracleConnection Connection {get; set;}
An OracleConnection object that indicates the connection associated with the queue.
ObjectDisposedException - The object is already disposed.
This connection must be opened before calling methods like Enqueue and Dequeue.
This instance property specifies the dequeueing options to use when dequeuing a message from a queue.
// C#
public OracleAQDequeueOptions DequeueOptions {get; set}
An OracleAQDequeueOptions object.
ObjectDisposedException - The object is already disposed.
The default value is an OracleAQDequeueOptions object with default property values. Setting this property to null resets all dequeue options to their default values.
This instance property specifies the enqueueing options used to enqueue a message to a queue.
// C#
public OracleAQEnqueueOptions EnqueueOptions {get; set}
An OracleAQEnqueueOptions object.
ObjectDisposedException - The object is already disposed.
The default value is an OracleAQEnqueueOptions object with default property values. Setting this property to null resets all enqueue options to their default values.
This instance property specifies the type of queue table associated with this queue.
// C#
public OracleAQMessageType MessageType {get; set;}
An OracleAQMessageType enumerated value.
ArgumentOutOfRangeException - The type value specified is invalid.
ObjectDisposedException - The object is already disposed.
The MessageType property also dictates the type of message payloads that are enqueued or dequeued from the queue. It is possible to enqueue a variety of payloads depending on the MessageType.
Table 12-27 lists the allowed payload types for various message types.
Table 12-27 Message Types and Payloads
| OracleAQQueue.MessageType | Allowed OracleAQMessage.Payload type to Enqueue |
|---|---|
|
|
|
|
|
|
|
|
|
Table 12-28 lists the payload types for dequeued messages.
Table 12-28 Payload Types for Dequeued Messages
| OracleAQQueue.MessageType | DequeueOptions.ProviderSpecificType | OracleAQMessage.Payload of dequeued message |
|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
N.A. |
|
This instance property returns the name of the queue.
// C#
public string Name {get;}
A string that indicates the name of the queue.
ObjectDisposedException - The object is already disposed.
This instance property specifies the various notification options for notifications that are registered using the MessageAvailable event.
// C#
public OracleNotificationRequest Notification {get;}
Specifies an OracleNotificationRequest object whose properties can be changed to alter the notification behavior.
This property can be used to change various notification options. The notification options must be changed before registering with the MessageAvailable event. This property can be modified again only after unregistering from the MessageAvailable event.
This instance property specifies the array of consumers, for a multiconsumer queue, that are to be notified asynchronously for any incoming messages on the queue.
// C#
public string[] NotificationConsumers {get; set;}
Specifies an array of consumer name strings for which the notifications are delivered.
ObjectDisposedException - The object is already disposed.
InvalidOperationException - MessageAvailable registration is active.
The consumer names must be in uppercase. This functionality only supports queues with uppercase names.
The list of consumers is used in the MessageAvailable event. The list must be set before registering for the event. This property cannot be modified after registering for the MessageAvailable event. This property can be modified again only after unregistering from MessageAvailable event.
This instance property specifies the type name on which the queue and the corresponding queue table is based if the MessageType is OracleAQMessageType.UDT.
// C#
public string UdtTypeName {get; set;}
Specifies the Oracle user-defined type name if the MessageType is OracleAQMessageType.UDT.
ObjectDisposedException - The object is already disposed.
The UdtTypeName property corresponds to the user-defined type name of the payload. This property must always be specified if the payload is a user-defined type. This property need not be set for other payload types.
OracleAQQueue public methods are listed in Table 12-29.
Table 12-29 OracleAQQueue Public Methods
| Public Method | Description |
|---|---|
|
Dequeues messages from queues (Overloaded). |
|
|
Dequeues multiple messages from queues (Overloaded). |
|
|
Releases any resources or memory allocated by the object |
|
|
Enqueues messages to queues (Overloaded). |
|
|
Enqueues multiple messages to a queue (Overloaded). |
|
|
Listens for messages on the queue on behalf of |
See Also:
Dequeue methods dequeue messages from queues.
This instance method dequeues messages from a queue using the DequeueOptions for the instance.
Dequeue(OracleAQDequeueOptions)
This instance method dequeues messages from a queue using the supplied dequeue options.
This instance method is used to dequeue a message from a queue using the DequeueOptions for the instance.
// C# public OracleAQMessage Dequeue();
An OracleAQMessage instance representing the dequeued message.
InvalidOperationException - The connection is not open.
ObjectDisposedException - The object is already disposed.
OracleException - In case of timeout, an exception is thrown with the message, ORA-25228: timeout or end-of-fetch during message dequeue from queue_name.Timeout may happen if DequeueOptions.Wait is set to a value other than -1.
The MessageType property must be set appropriately before calling this function. If the MessageType is OracleAQMessageType.UDT, then the UdtTypeName property must also be set.
Dequeued buffered messages always have null MessageId values.
This instance method dequeues messages from a queue using the supplied dequeue options.
// C#
public OracleAQMessage Dequeue(OracleAQDequeueOptions dequeueOptions);
dequeueOptions
An OracleAQDequeueOptions object.
An OracleAQMessage instance representing the dequeued message.
InvalidOperationException - The connection is not open.
ObjectDisposedException - The object is already disposed.
OracleException - In case of timeout, an exception is thrown with the message, ORA-25228: timeout or end-of-fetch during message dequeue from queue_name. Timeout may happen if DequeueOptions.Wait is set to a value other than -1.
If the supplied dequeueOptions object is null, then the dequeue options default values are used. The queue object's DequeueOptions property is ignored for this operation.
Calling this method does not change the DequeueOptions property of the queue.
The MessageType property must be set appropriately before calling this function. If the MessageType is OracleAQMessageType.UDT, then the UdtTypeName property must also be set.
Dequeued buffered messages always have null MessageId values.
DequeueArray methods dequeue multiple messages from queues.
This instance method dequeues multiple messages from a queue using the DequeueOptions of the instance.
DequeueArray(int, OracleAQDequeueOptions)
This instance method dequeues multiple messages from a queue using the supplied dequeue options.
This instance method dequeues multiple messages from a queue using the DequeueOptions of the instance.
// C#
public OracleAQMessage[] DequeueArray(int dequeueCount);
dequeueCount
An integer specifying the numbers of messages to dequeue.
An array of OracleAQMessage instances representing the dequeued messages.
ArgumentOutOfRangeException - dequeueCount is less than or equal to 0.
InvalidOperationException - The connection is not open.
ObjectDisposedException - The object is already disposed.
OracleException - In case of timeout, an exception is thrown with the message, ORA-25228: timeout or end-of-fetch during message dequeue from queue_name. Timeout may happen if DequeueOptions.Wait is set to a value other than -1.
This method is supported for Oracle Database 10g and higher releases.
The MessageType property must be set appropriately before calling this function. If the MessageType is OracleAQMessageType.UDT, then the UdtTypeName property must be set as well.
The size of the returned array may be less than the dequeueCount. It depends on the actual number of messages present in the queue.
For database versions earlier than Oracle Database 12c Release 1 (12.1), the MessageId property of persistent OracleAQMessage objects retrieved using DequeueArray is always null.
Dequeued buffered messages always have null MessageId values irrespective of the database version.
This instance method dequeues multiple messages from a queue using the supplied dequeue options.
// C# public OracleAQMessage[] DequeueArray(int dequeueCount, OracleAQDequeueOptions dequeueOptions);
dequeueCount
An integer specifying the numbers of messages to dequeue.
dequeueOptions
An OracleAQDequeueOptions object.
An array of OracleAQMessage instances representing the dequeued messages.
ArgumentOutOfRangeException - dequeueCount is less than or equal to 0.
InvalidOperationException - The connection is not open.
ObjectDisposedException - The object is already disposed.
OracleException - In case of timeout, an exception is thrown with the message, ORA-25228: timeout or end-of-fetch during message dequeue from queue_name. Timeout may happen if DequeueOptions.Wait is set to a value other than -1.
This method is supported for Oracle Database 10g Release 1 (10.1) and higher releases. Calling this method does not change the DequeueOptions property of the queue.
If the supplied dequeueOptions object is null, then the dequeue options default values are used. The DequeueOptions property of the queue object is ignored in this operation.
The MessageType property must be set appropriately before calling this function. If the MessageType is OracleAQMessageType.UDT, then the UdtTypeName property must be set as well.
The size of the returned array may be less than the dequeueCount. It dependes on the actual number of messages present in the queue.
For database versions earlier than Oracle Database 12c Release 1 (12.1), the MessageId property of persistent OracleAQMessage objects retrieved using DequeueArray is always null.
Dequeued buffered messages always have null MessageId values irrespective of the database version.
This method releases any resources or memory allocated by the object.
// C# public void Dispose();
IDisposable.
Enqueue instance methods enqueue messages to queues.
This instance method enqueues messages to a queue using the EnqueueOptions of the instance.
Enqueue(OracleAQMessage, OracleAQEnqueueOptions)
This instance method enqueues messages to a queue using the supplied enqueue options.
This instance method enqueues messages to a queue using the EnqueueOptions of the instance.
// C#
public void Enqueue(OracleAQMessage message);
message
An OracleAQMessage object.
Exceptions
ObjectDisposedException - The object is already disposed.
InvalidOperationException - The connection is not open.
ArgumentNullException - The message parameter is null.
ArgumentException - The message payload is OracleXmlType and the connection used to create OracleXmlType is different from the queue's connection.
MessageId of the enqueued message is populated after the call to Enqueue completes. Enqueued buffered messages always have null MessageId values.
The MessageType property needs to be set appropriately before calling this function. If the MessageType is OracleAQMessageType.UDT, then the UdtTypeName property must be set as well.
This instance method enqueues messages to a queue using the supplied enqueue options.
// C# public void Enqueue(OracleAQMessage message, OracleAQEnqueueOptions enqueueOptions);
message
An OracleAQMessage object.
enqueueOptions
An OracleAQEnqueueOptions object.
ObjectDisposedException - The object is already disposed.
InvalidOperationException - The connection is not open.
ArgumentNullException - The message parameter is null.
ArgumentException - The message payload is OracleXmlType and the connection used to create OracleXmlType is different from the queue's connection.
If the supplied enqueueOptions object is null, then the enqueue options default values are used. The EnqueueOptions property of the queue object is ignored in this operation.
The MessageId of the enqueued message is populated after the call to Enqueue completes. Enqueued buffered messages always have null MessageId values. Calling this method does not change the EnqueueOptions property of the queue.
The MessageType property must be set appropriately before calling this function. If the MessageType is OracleAQMessageType.UDT, then the UdtTypeName property must also be set.
EnqueueArray instance methods enqueue multiple messages to a queue.
EnqueueArray(OracleAQMessage[ ])
This instance method enqueues multiple messages to a queue using the EnqueueOptions of the instance.
EnqueueArray(OracleAQMessage[ ], OracleAQEnqueueOptions)
This instance method enqueues multiple messages to a queue using the supplied enqueue options.
This instance method enqueues multiple messages to a queue using the EnqueueOptions of the instance.
// C#
public int EnqueueArray(OracleAQMessage[] messages);
messages
An array of OracleAQMessage objects.
An integer representing the number of messages actually enqueued.
ArgumentNullException - The message parameter is null.
ArgumentException - At least one of the OracleAQMessage[] elements is null, or at least one of the OracleAQMessage[] elements has a payload of OracleXmlType, which is created using a connection that is different from the queue's connection.
InvalidOperationException - The OracleAQMessage array is empty or the connection is not open.
ObjectDisposedException - The object is already disposed.
This method is supported by Oracle Database 10g and higher releases. The MessageId properties of the enqueued messages are populated after the call to Enqueue completes. Enqueued buffered messages always have null MessageId values.
The MessageType property must be set appropriately before calling this function. If the MessageType is OracleAQMessageType.UDT, then the UdtTypeName property must also be set.
This instance method enqueues multiple messages to a queue using the supplied enqueue options.
// C# public int EnqueueArray(OracleAQMessage[] messages, OracleAQEnqueueOptions enqueueOptions);
messages
An array of OracleAQMessage objects.
enqueueOptions
An OracleAQEnqueueOptions object.
An integer representing the number of messages actually enqueued.
ArgumentNullException - The message parameter is null.
ArgumentException - At least one of the OracleAQMessage[] elements is null, or at least one of the OracleAQMessage[] elements has a payload of OracleXmlType, which is created using a connection that is different from the queue's connection.
InvalidOperationException - The OracleAQMessage array is empty or the connection is not open.
ObjectDisposedException - The object is already disposed.
This method is supported by Oracle Database 10g and higher releases. MessageId properties of the enqueued messages are populated after the call to Enqueue completes. Enqueued buffered messages always have null MessageId values. Calling this method does not change the EnqueueOptions property of the queue.
If the supplied enqueueOptions object is null, then the enqueue options default values are used. The EnqueueOptions property of the queue object is ignored in this operation.
The MessageType property must be set appropriately before calling this function. If the MessageType is OracleAQMessageType.UDT, then the UdtTypeName property must also be set.
Listen methods listen for messages on the queue on behalf of listenConsumers.
This method listens for messages on the queue on behalf of listenConsumers.
This method listens for messages on behalf of listenConsumers for a specified time.
This method listens for messages on the queue on behalf of listenConsumers.
// C#
public string Listen(string[] listenConsumers);
listenConsumers
An array of consumers to listen for on this queue. This parameter should be null in case of single consumer queues.
A string.
InvalidOperationException - The connection is not open.
ObjectDisposedException - The object is already disposed.
This call blocks the calling thread until there is a message ready for consumption for a consumer in the listenConsumers array. It returns a string representing the consumer name for which the message is ready.
Listen is useful in situations that require waiting until a message is available in the queue for consumers whose names are specified in listenConsumers.
The following example demonstrates using the Listen method. The first part of the example performs the requisite database setup for the database user, SCOTT. The second part of the example demonstrates how a thread can listen and wait until a message is enqueued.
-- Part I: Database setup required for this demo
------------------------------------------------------------------
-- SQL to grant appropriate privilege to database user, SCOTT
------------------------------------------------------------------
SQL> ALTER USER SCOTT ACCOUNT UNLOCK IDENTIFIED BY Pwd4Sct;
User altered.
GRANT ALL ON DBMS_AQADM TO scott;
------------------------------------------------------------------
-- PLSQL to create queue-table and queue and start queue for SCOTT
------------------------------------------------------------------
BEGIN
DBMS_AQADM.CREATE_QUEUE_TABLE(
queue_table=>'scott.test_q_tab',
queue_payload_type=>'RAW',
multiple_consumers=>FALSE);
DBMS_AQADM.CREATE_QUEUE(
queue_name=>'scott.test_q',
queue_table=>'scott.test_q_tab');
DBMS_AQADM.START_QUEUE(queue_name=>'scott.test_q');
END;
/
------------------------------------------------------------------
-- PLSQL to stop queue and drop queue & queue-table from SCOTT
------------------------------------------------------------------
BEGIN
DBMS_AQADM.STOP_QUEUE('scott.test_q');
DBMS_AQADM.DROP_QUEUE(
queue_name => 'scott.test_q',
auto_commit => TRUE);
DBMS_AQADM.DROP_QUEUE_TABLE(
queue_table => 'scott.test_q_tab',
force => FALSE,
auto_commit => TRUE);
END;
/
-- End of Part I, database setup.
//Part II: Demonstrates using the Listen method
//C#
using System;
using System.Text;
using Oracle.DataAccess.Client;
using Oracle.DataAccess.Types;
using System.Threading;
namespace ODPSample
{
/// <summary>
/// Demonstrates how a thread can listen and wait until a message is enqueued.
/// Once a message is enqueued, the listening thread returns from the
/// blocked Listen() method invocation and dequeues the message.
/// </summary>
class EnqueueDequeue
{
static bool s_bListenReturned = false;
static void Main(string[] args)
{
// Create connection
string constr = "user id=scott;password=Pwd4Sct;data source=oracle";
OracleConnection con = new OracleConnection(constr);
// Create queue
OracleAQQueue queue = new OracleAQQueue("scott.test_q", con);
try
{
// Open connection
con.Open();
// Set message type for the queue
queue.MessageType = OracleAQMessageType.Raw;
// Spawning a thread which will listen for a message
ThreadStart ts = new ThreadStart(TestListen);
Thread t = new Thread(ts);
t.Start();
System.Threading.Thread.Sleep(2000);
// Begin transaction for enqueue
OracleTransaction txn = con.BeginTransaction();
// Prepare message and RAW payload
OracleAQMessage enqMsg = new OracleAQMessage();
byte[] bytePayload = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
enqMsg.Payload = bytePayload;
// Prepare to Enqueue
queue.EnqueueOptions.Visibility = OracleAQVisibilityMode.OnCommit;
Console.WriteLine("[Main Thread] Enqueuing a message...");
Console.WriteLine("[Main Thread] Enqueued Message Payload : "
+ ByteArrayToString(enqMsg.Payload as byte[]));
Console.WriteLine();
// Enqueue message
queue.Enqueue(enqMsg);
// Enqueue transaction commit
txn.Commit();
// Loop till Listen returns
while (!s_bListenReturned)
System.Threading.Thread.Sleep(1000);
}
catch (Exception e)
{
Console.WriteLine("Error: {0}", e.Message);
}
finally
{
// Close/Dispose objects
queue.Dispose();
con.Close();
con.Dispose();
}
}
static void TestListen()
{
// Create connection
string constr = "user id=scott;password=Pwd4Sct;data source=oracle";
OracleConnection conListen = new OracleConnection(constr);
// Create queue
OracleAQQueue queueListen = new OracleAQQueue("scott.test_q", conListen);
try
{
// Open the connection for Listen thread.
// Connection blocked on Listen thread can not be used for other DB
// operations
conListen.Open();
// Set message type for the queue
queueListen.MessageType = OracleAQMessageType.Raw;
// Listen
queueListen.Listen(null);
Console.WriteLine("[Listen Thread] Listen returned... Dequeuing...");
// Begin txn for Dequeue
OracleTransaction txn = conListen.BeginTransaction();
// Prepare to Dequeue
queueListen.DequeueOptions.Visibility = OracleAQVisibilityMode.OnCommit;
queueListen.DequeueOptions.Wait = 10;
// Dequeue message
OracleAQMessage deqMsg = queueListen.Dequeue();
Console.WriteLine("[Listen Thread] Dequeued Message Payload : "
+ ByteArrayToString(deqMsg.Payload as byte[]));
// Dequeue txn commit
txn.Commit();
// Allow the main thread to exit
s_bListenReturned = true;
}
catch (Exception e)
{
Console.WriteLine("Error: {0}", e.Message);
}
finally
{
// Close/Dispose objects
queueListen.Dispose();
conListen.Close();
conListen.Dispose();
}
}
// Function to convert byte[] to string
static private string ByteArrayToString(byte[] byteArray)
{
StringBuilder sb = new StringBuilder();
for (int n = 0; n < byteArray.Length; n++)
{
sb.Append((int.Parse(byteArray[n].ToString())).ToString("X"));
}
return sb.ToString();
}
}
}
This method listens for messages on behalf of listenConsumers for a specified time.
// C# public string Listen(string[] listenConsumers, int waitTime);
listenConsumers
Array of consumers for which to listen on this queue.
waitTime
Wait time in seconds.
A string
InvalidOperationException - The connection is not open.
ArgumentException - waitTime is less than -1.
ObjectDisposedException - The object is already disposed.
Listen is useful in situations that require waiting until a message is available in the queue for consumers whose names are specified in listenConsumers.
This call blocks the calling thread until there is a message ready for consumption for a consumer in the listenConsumers array. It returns a string representing the consumer name for which the message is ready.The method returns null if a timeout occurs.
The listenConsumers parameter should be null for single consumer queues. An empty string is returned in such cases.
A waitTime of -1 implies infinite wait time.
The OracleAQQueue event is listed in Table 12-30.
Table 12-30 OracleAQQueue Events
| Event Name | Description |
|---|---|
|
Notifies when a message is available in the queue for |
See Also:
This event is notified when a message is available in the queue for NotificationConsumers.
// C# public event OracleAQMessageAvailableEventHandler MessageAvailable;
The event handler receives an OracleAQMessageAvailableEventArgs object.
InvalidOperationException - The connection is not open.
Asynchronous notification is supported in all queue tables created with a database compatibility level of 8.1 or higher.
In order to receive the notification about message availability, the client must create an OracleAQMessageAvailableEventHandler delegate to listen to this event. The delegate should be added to this event only after setting the NotificationConsumers and Notification properties.
The notification registration takes place after the first delegate is added to the event. The notification is unregistered when the last delegate is removed from the event. Notifications set on an OracleAQQueue object get cancelled automatically when the object gets disposed.
Oracle Data Provider for .NET opens a port to listen for notifications. HA events, load balancing, and continuous query notification features also share the same port. This port can be configured centrally by setting the database notification port in an application or Web configuration file. The following example code specifies a port number of 1200:
<configuration>
<oracle.dataaccess.client>
<settings>
<add name="DbNotificationPort" value="1200"/>
</settings>
</oracle.dataaccess.client>
</configuration>
If the configuration file does not exist or the db notification port is not specified, then ODP.NET uses a valid and random port number. The configuration file may also request for a random port number by specifying a db notification port value of -1.
The notification listener, which runs in the same application domain as ODP.NET, uses the specified port number to listen to notifications from the database. A notification listener gets created when the application registers with OracleAQQueue.MessageAvailable event. One notification listener can listen to all notification types. Only one notification listener is created for each application domain.
The following example demonstrates application notification. The first part of the example performs the requisite database setup for the database user, SCOTT. The second part of the example demonstrates how an application is notified when a message is available in the queue.
-- Part I: Database setup required for this demo
------------------------------------------------------------------
-- SQL to grant appropriate privilege to database user, SCOTT
------------------------------------------------------------------
SQL> ALTER USER SCOTT ACCOUNT UNLOCK IDENTIFIED BY Pwd4Sct;
User altered.
SQL> GRANT ALL ON DBMS_AQADM TO scott;
------------------------------------------------------------------
-- PLSQL to create queue-table and queue and start queue for SCOTT
------------------------------------------------------------------
BEGIN
DBMS_AQADM.CREATE_QUEUE_TABLE(
queue_table=>'scott.test_q_tab',
queue_payload_type=>'RAW',
multiple_consumers=>FALSE);
DBMS_AQADM.CREATE_QUEUE(
queue_name=>'scott.test_q',
queue_table=>'scott.test_q_tab');
DBMS_AQADM.START_QUEUE(queue_name=>'scott.test_q');
END;
/
------------------------------------------------------------------
-- PLSQL to stop queue and drop queue & queue-table from SCOTT
------------------------------------------------------------------
BEGIN
DBMS_AQADM.STOP_QUEUE('scott.test_q');
DBMS_AQADM.DROP_QUEUE(
queue_name => 'scott.test_q',
auto_commit => TRUE);
DBMS_AQADM.DROP_QUEUE_TABLE(
queue_table => 'scott.test_q_tab',
force => FALSE,
auto_commit => TRUE);
END;
/
-- End of Part I, database setup.
//Part II: Demonstrates application notification
//C#
using System;
using System.Text;
using Oracle.DataAccess.Client;
using Oracle.DataAccess.Types;
namespace ODPSample
{
/// <summary>
/// Demonstrates how the application can be notified when a message is
/// available in a queue.
/// </summary>
class Notification
{
static bool isNotified = false;
static void Main(string[] args)
{
// Create connection
string constr = "user id=scott;password=Pwd4Sct;data source=oracle";
OracleConnection con = new OracleConnection(constr);
// Create queue
OracleAQQueue queue = new OracleAQQueue("scott.test_q", con);
try
{
// Open connection
con.Open();
// Set message type for the queue
queue.MessageType = OracleAQMessageType.Raw;
// Add the event handler to handle the notification. The
// MsgReceived method will be invoked when a message is enqueued
queue.MessageAvailable +=
new OracleAQMessageAvailableEventHandler(Notification.MsgReceived);
Console.WriteLine("Notification registered...");
// Begin txn for enqueue
OracleTransaction txn = con.BeginTransaction();
Console.WriteLine("Now enqueuing message...");
// Prepare message and RAW payload
OracleAQMessage enqMsg = new OracleAQMessage();
byte[] bytePayload = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
enqMsg.Payload = bytePayload;
// Prepare to Enqueue
queue.EnqueueOptions.Visibility = OracleAQVisibilityMode.OnCommit;
// Enqueue message
queue.Enqueue(enqMsg);
Console.WriteLine("Enqueued Message Payload : "
+ ByteArrayToString(enqMsg.Payload as byte[]));
Console.WriteLine("MessageId of Enqueued Message : "
+ ByteArrayToString(enqMsg.MessageId));
Console.WriteLine();
// Enqueue txn commit
txn.Commit();
// Loop while waiting for notification
while (isNotified == false)
{
System.Threading.Thread.Sleep(2000);
}
}
catch (Exception e)
{
Console.WriteLine("Error: {0}", e.Message);
}
finally
{
// Close/Dispose objects
queue.Dispose();
con.Close();
con.Dispose();
}
}
static void MsgReceived(object src, OracleAQMessageAvailableEventArgs arg)
{
try
{
Console.WriteLine("Notification Received...");
Console.WriteLine("QueueName : {0}", arg.QueueName);
Console.WriteLine("Notification Type : {0}", arg.NotificationType);
//following type-cast to "byte[]" is required only for .NET 1.x
byte[] notifiedMsgId = (byte[]) arg.MessageId[0];
Console.WriteLine("MessageId of Notified Message : "
+ ByteArrayToString(notifiedMsgId));
isNotified = true;
}
catch (Exception e)
{
Console.WriteLine("Error: {0}", e.Message);
}
}
// Function to convert byte[] to string
static private string ByteArrayToString(byte[] byteArray)
{
StringBuilder sb = new StringBuilder();
for (int n = 0; n < byteArray.Length; n++)
{
sb.Append((int.Parse(byteArray[n].ToString())).ToString("X"));
}
return sb.ToString();
}
}
}