Listen(string[])
This method listens for messages on the queue on behalf of listenConsumers
.
Declaration
// C#
public string Listen(string[] listenConsumers);
Parameters
-
listenConsumers
An array of consumers to listen for on this queue. This parameter should be
null
in case of single consumer queues.
Return Value
A string
.
Exceptions
InvalidOperationException
- The connection is not open.
ObjectDisposedException
- The object is already disposed.
Remarks
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
.
Example
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(); } } }