TOraQueue, TOraQueueAdmin and TOraQueueTable Components

TOraQueue, TOraQueueAdmin and TOraQueueTable components provide access to Oracle Streams Advanced Queuing. Oracle Streams AQ provides database-integrated message queuing functionality. It is built on top of Oracle Streams and leverages the functions of Oracle Database so that messages can be stored persistently, propagated between queues on different computers and databases, and transmitted using Oracle Net Services and HTTP(S).

Because Oracle Streams AQ is implemented in database tables, all operational benefits of high availability, scalability, and reliability are also applicable to queue data. Standard database features such as recovery, restart, and security are supported by Oracle Streams AQ. Like other database tables, queue tables can be imported and exported.

When applications communicate with each other, producer applications enqueue messages and consumer applications dequeue messages. At the basic level of queuing, one producer enqueues one or more messages into one queue. Each message is dequeued and processed once by one of the consumers. A message stays in the queue until a consumer dequeues it or the message expires. A producer may stipulate the delay before the message is available for the consumption, and the time after which the message expires. Likewise, a consumer may wait when trying to dequeue a message if no message is available. An agent program or application may act both as a producer and a consumer.

TOraQueue component provides access to functionality of DBMS_AQ Oracle package. User must have AQ_USER_ROLE to work with this component. TOraQueue component can be used to enqueue and dequeue messages from the given queue. Queue message includes a payload and message properties. Type of payload is defined for each queue. This can be Oracle object type or 'RAW' type. RAW payload contains any array of bytes. To use TOraQueue component set its Session and QueueName properties. Then one of Enqueue method overloads can be used to enqueue message.

To enqueue message with RAW payload use the Enqueue method overload with string or TBytes Payload parameter. For example:

var
  MsgProp: TQueueMessageProperties;
begin
  MsgProp := TQueueMessageProperties.Create;
  try
    MsgProp.Priority := 2;
    MsgProp.Delay := 30; // delay in sec before message can be dequeued
    MsgProp.Expiration := 180; // message expiration in sec
 
    OraQueue.Enqueue('123', MsgProp);
  finally
    MsgProp.Free;
  end;
end;

To enqueue message to queue with object payload use Enqueue method overload with TOraObject Payload parameter. For example:

var
  MsgProp: TQueueMessageProperties;
  Payload: TOraObject;
begin
  MsgProp := TQueueMessageProperties.Create;
  try
    MsgProp.Priority := 2;
    MsgProp.Delay := 30; // delay in sec before message can be dequeued
    MsgProp.Expiration := 180; // message expiration in sec
 
    Payload := TOraObject.Create;
    try
      Payload.AllocObject(OraSession.OCISvcCtx, 'OBJ_MES');
      Payload.AttrAsInteger['A'] := 3;
      Payload.AttrAsString['B'] := 'Hello';
 
      OraQueue.Enqueue(Payload, MsgProp);
    finally
      Payload.Free;
    end;
  finally
    MsgProp.Free;
  end;
end;

To dequeue message, use one of TOraQueue.Dequeue method overloads. If there are no messages available for dequeuing, Dequeue method will wait until a message will be available. The following example demonstrates the usage of Dequeue method for a queue with object payload:

var
  Payload: TOraObject;
  a: integer;
  b: string;
begin
  Payload := TOraObject.Create;
  try
    OraQueue.Dequeue(Payload);
 
    a := Payload.AttrAsInteger['A'];
    b := Payload.AttrAsString['B'];
  finally
    Payload.Free;
  end;
end;

To get notification when a message available to dequeuing appears in the queue use TOraQueue.Listen method or set AsyncNotification property to True and write OnMessage event handler.

Listen method listens on one or more queues on behalf of a list of agents. This method waits until a message is available in one of the queues and then returns the agent that corresponds this queue. Listen method should be called in separate thread to avoid program blocking. For example:

procedure TListenThread.Execute;
var
  Agents: TQueueAgents;
  Agent: TQueueAgent;
  Message: string;
begin
  Agents := TQueueAgents.Create;
  try
    Agents.Add.Address := 'QUEUE1'; // Address property should contain the queue name
    Agents.Add.Address := 'QUEUE2';
 
    Agent := TQueueAgent.Create;
    try
      while not Terminated do begin
        OraQueue.Listen(Agents, Agent);
        OraQueue.Dequeue(Message);
      end;
    finally
      Agent.Free;
    end;
  finally
    Agents.Free;
  end;
end;

TOraQueueAdmin and TOraQueueTable components are used to administrate queues. User must have AQ_ADMINISTRATOR_ROLE to work with these components. TOraQueueAdmin and TOraQueueTable components can be used to create and drop queues and queue tables, alter queue properties.

Queue can be persistent or non-persistent. To create persistent queue first a queue table must be created. Use TOraQueueTable component to create a queue table. Set properties of the component to required values and then call CreateQueueTable method. Use AlterQueueTable method to alter a queue table properties and DropQueueTable method to drop a queue table.

When a queue table is created, TOraQueueAdmin component can be used to create a queue. Set properties of the component to required values and then call CreateQueue method. Use AlterQueue method of TOraQueueAdmin to alter a queue properties and DropQueue method to drop a queue.

Before messages can be enqueued and dequeued enqueuing and dequeuing must be started on the queue. Use StartQueue method of TOraQueueAdmin to start  enqueuing and dequeuing on the queue.