Operational Interface: Basic Operations, 10 of 19

Listening to One (Many) Multi-Consumer Queue(s)

Figure 11-9 Use Case Diagram: Listening to One(Many) Multi-Consumer Queue(s)

Text description of adq11qopa.gif follows

Text description of the illustration adq11qopa.gif

To refer to the table of all basic operations having to do with the Operational Interface see:

"Use Case Model: Operational Interface -- Basic Operations"

Usage Notes

Not applicable.

Syntax

See Chapter 3, "AQ Programmatic Environments" for a list of available functions in each programmatic environment. Use the following syntax references for each programmatic environment:

PL/SQL (DBMS_AQ Package): Oracle9i Supplied PL/SQL Packages and Types Reference DBMS_AQ, LISTEN procedure
C (OCI): Oracle Call Interface Programmer's Guide Relational Functions, OCIAQListen
Visual Basic (OO4O) (Oracle Objects for OLE (OO4O) Online Help): From Help Topics, Contents tab, select OO4O Automation Server > OBJECTS > OraAQ Object > Monitoring Messages
Feature not available through the Java API

Examples

See Chapter 3, "AQ Programmatic Environments" for a list of available functions in each programmatic environment. Examples in the following programmatic environments are provided:

PL/SQL (DBMS_AQ Package): Listen to Queue(s)

/* The listen call allows you to monitor a list of queues for messages for 
   specific agents. You need to have dequeue privileges for all the queues 
   you wish to monitor. */

Listen to Multi-Consumer Queue (Timeout of Zero).

DECLARE
   Agent_w_msg      aq$_agent;
   My_agent_list    dbms_aq.agent_list_t;

BEGIN
   /* NOTE:  MCQ1, MCQ2, MCQ3 are multi-consumer queues  in SCOTT's schema
   *        SCQ1, SCQ2, SCQ3 are single consumer queues in SCOTT's schema
   */
    Qlist(1):= aq$_agent('agent1', 'MCQ1',  NULL);
    Qlist(2):= aq$_agent('agent2', 'scott.MCQ2', NULL);
    Qlist(3):= aq$_agent('agent3', 'scott.MCQ3', NULL);

   /* Listen with a time-out of zero: */
   DBMS_AQ.LISTEN(
      agent_list   =>    My_agent_list, 
      wait         =>    0, 
      agent        =>    agent_w_msg);
   DBMS_OUTPUT.PUT_LINE('Message in Queue :- ' ||  agent_w_msg.address);
   DBMS_OUTPUT.PUT_LINE('');
END;
   /

Listen to Mixture of Multi-Consumer Queues (Timeout 100 Seconds).

DECLARE
   Agent_w_msg      aq$_agent;
   My_agent_list    dbms_aq.agent_list_t;

BEGIN
   /* NOTE:  MCQ1, MCQ2, MCQ3 are multi-consumer queues  in SCOTT's schema
   *        SCQ1, SCQ2, SCQ3 are single consumer queues in SCOTT's schema
   */
   Qlist(1):= aq$_agent('agent1', 'MCQ1',  NULL);
   Qlist(2):= aq$_agent(NULL, 'scott.SQ1', NULL);
   Qlist(3):= aq$_agent('agent3', 'scott.MCQ3', NULL);
   /* Listen with a time-out of 100 seconds */
   DBMS_AQ.LISTEN(
      Agent_list   =>   My_agent_list, 
      Wait         =>   100, 
      Agent        =>    agent_w_msg);
      DBMS_OUTPUT.PUT_LINE('Message in Queue :- ' ||  agent_w_msg.address 
                           || 'for agent' || agent_w_msg.name);
      DBMS_OUTPUT.PUT_LINE('');
   END;
   /

C (OCI): Listen to Multi-Consumer Queue(s)

Listening to Multi-consumer Queues with a Zero Timeout, a Timeout of 120 Seconds, and a Timeout of 100 Seconds

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <oci.h>

static void checkerr(errhp, status)
OCIError *errhp;
sword status;
{
    text errbuf[512];
    ub4 buflen;
    sb4 errcode;

    switch (status)
    {
   case OCI_SUCCESS:
       break;
   case OCI_SUCCESS_WITH_INFO:
       printf("Error - OCI_SUCCESS_WITH_INFO\n");
       break;
   case OCI_NEED_DATA:
       printf("Error - OCI_NEED_DATA\n");
       break;
   case OCI_NO_DATA:
       printf("Error - OCI_NO_DATA\n");
       break;
   case OCI_ERROR:
       OCIErrorGet ((dvoid *) errhp, (ub4) 1, (text *) NULL, &errcode,
       errbuf, (ub4) sizeof(errbuf), (ub4) OCI_HTYPE_ERROR);
       printf("Error - %s\n", errbuf);
       break;
   case OCI_INVALID_HANDLE:
       printf("Error - OCI_INVALID_HANDLE\n");
       break;
   case OCI_STILL_EXECUTING:
       printf("Error - OCI_STILL_EXECUTE\n");
       break;
   case OCI_CONTINUE:
       printf("Error - OCI_CONTINUE\n");
       break;
   default:
   break;
    }
}

void SetAgent(OCIAQAgent *agent, 
         text       *appname, 
         text       *queue,
         OCIError   *errhp,
         OCIEnv     *envhp);

void GetAgent(OCIAQAgent *agent, 
         OCIError   *errhp);

/*----------------------------------------------------------------*/
/* OCI Listen examples for multi-consumers                        */
/*                                                                */
void SetAgent(agent, appname, queue, errhp)
OCIAQAgent    *agent;
text          *appname;
text          *queue;
OCIError      *errhp;
{
  OCIAttrSet(agent, 
        OCI_DTYPE_AQAGENT, 
        appname ? (dvoid *)appname : (dvoid *)"", 
        appname ? strlen((const char *)appname) : 0,
             OCI_ATTR_AGENT_NAME, 
        errhp);

  OCIAttrSet(agent, 
        OCI_DTYPE_AQAGENT, 
        queue ? (dvoid *)queue : (dvoid *)"", 
        queue ? strlen((const char *)queue) : 0,
             OCI_ATTR_AGENT_ADDRESS, 
        errhp);

  printf("Set agent name to %s\n", appname ? (char *)appname : "NULL");
  printf("Set agent address to %s\n", queue ? (char *)queue : "NULL");
}

/* get agent from descriptor */
void GetAgent(agent, errhp)
OCIAQAgent *agent;
OCIError   *errhp;
{
   text      *appname;
   text      *queue;
   ub4       appsz;
   ub4       queuesz;

   if (!agent )
  {
     printf("agent was NULL \n");
     return;
  }
  checkerr(errhp, OCIAttrGet(agent, OCI_DTYPE_AQAGENT, 
     (dvoid *)&appname, &appsz, OCI_ATTR_AGENT_NAME, errhp));
  checkerr(errhp, OCIAttrGet(agent, OCI_DTYPE_AQAGENT, 
     (dvoid *)&queue, &queuesz, OCI_ATTR_AGENT_ADDRESS, errhp));
  if (!appsz)
     printf("agent name: NULL\n");
  else printf("agent name: %.*s\n", appsz, (char *)appname);
  if (!queuesz)
     printf("agent address: NULL\n");
  else printf("agent address: %.*s\n", queuesz, (char *)queue);
}

/* main from AQ Listen to Multi-Consumer Queue(s) */

/*  int main() */
int main(char *argv, int argc)
{
    OCIEnv     *envhp;
    OCIServer  *srvhp;
    OCIError   *errhp;
    OCISvcCtx  *svchp;
    OCISession *usrhp;
    OCIAQAgent *agent_list[3];
    OCIAQAgent *agent;
    int         i;

 /* Standard OCI Initialization */

  OCIInitialize((ub4) OCI_OBJECT, 
      (dvoid *)0,  
      (dvoid * (*)()) 0,
      (dvoid * (*)()) 0,  
      (void (*)()) 0 );
  
  OCIHandleAlloc( (dvoid *) NULL, (dvoid **) &envhp, (ub4) OCI_HTYPE_ENV,
     0, (dvoid **) 0);
  
  OCIEnvInit( &envhp, (ub4) OCI_DEFAULT, 0, (dvoid **)0);
  
  OCIHandleAlloc( (dvoid *) envhp, (dvoid **) &errhp, (ub4) OCI_HTYPE_ERROR,
     0, (dvoid **) 0);

  OCIHandleAlloc( (dvoid *) envhp, (dvoid **) &srvhp, (ub4) OCI_HTYPE_SERVER,
     0, (dvoid **) 0);
  
  OCIServerAttach( srvhp, errhp, (text *) 0, (sb4) 0, (ub4) OCI_DEFAULT);
  
  OCIHandleAlloc( (dvoid *) envhp, (dvoid **) &svchp, (ub4) OCI_HTYPE_SVCCTX,
     0, (dvoid **) 0);
  
  /* set attribute server context in the service context */
  OCIAttrSet( (dvoid *) svchp, (ub4) OCI_HTYPE_SVCCTX, (dvoid *)srvhp, (ub4) 0,
     (ub4) OCI_ATTR_SERVER, (OCIError *) errhp);
  
   /* allocate a user context handle */
  OCIHandleAlloc((dvoid *)envhp, (dvoid **)&usrhp, (ub4) OCI_HTYPE_SESSION,
     (size_t) 0, (dvoid **) 0);
  
   /* allocate a user context handle */
  OCIHandleAlloc((dvoid *)envhp, (dvoid **)&usrhp, (ub4) OCI_HTYPE_SESSION,
     (size_t) 0, (dvoid **) 0);
  
  
  OCIAttrSet((dvoid *)usrhp, (ub4)OCI_HTYPE_SESSION,
     (dvoid *)"scott", (ub4)strlen("scott"), OCI_ATTR_USERNAME, errhp);
   

   OCIAttrSet((dvoid *) usrhp, (ub4) OCI_HTYPE_SESSION,
      (dvoid *) "tiger", (ub4) strlen("tiger"),
      (ub4) OCI_ATTR_PASSWORD, errhp);   
  
  OCISessionBegin (svchp, errhp, usrhp, OCI_CRED_RDBMS, OCI_DEFAULT);
  
  OCIAttrSet((dvoid *)svchp, (ub4)OCI_HTYPE_SVCCTX,
     (dvoid *)usrhp, (ub4)0, OCI_ATTR_SESSION, errhp);

  /* AQ LISTEN Initialization - allocate agent handles */
  for (i = 0; i < 3; i++)
  {
     OCIDescriptorAlloc(envhp, (dvoid **)&agent_list[i], 
         OCI_DTYPE_AQAGENT, 0, (dvoid **)0);
  }

  /* 
   *  MCQ1, MCQ2, MCQ3 are multi-consumer queues in SCOTT's schema 
   */
  /* Listening to Multi-consumer Queues with Zero Timeout */

  SetAgent(agent_list[0], "app1", "MCQ1", errhp);
  SetAgent(agent_list[1], "app2", "MCQ2", errhp);
  SetAgent(agent_list[2], "app3", "MCQ3", errhp);
 
  checkerr(errhp, OCIAQListen(svchp, errhp, agent_list, 3, 0, &agent, 0));

  printf("MESSAGE for :- \n");             
  GetAgent(agent, errhp);
  printf("\n");


  /* Listening to Multi-consumer Queues with Timeout of 120 Seconds */

  SetAgent(agent_list[0], "app1", "SCOTT.MCQ1", errhp);
  SetAgent(agent_list[1], "app2", "SCOTT.MCQ2", errhp);
  SetAgent(agent_list[2], "app3", "SCOTT.MCQ3", errhp);
 
  checkerr(errhp, OCIAQListen(svchp, errhp, agent_list, 3, 120, &agent, 0));

  printf("MESSAGE for :- \n");             
  GetAgent(agent, errhp);
  printf("\n");


  /* Listening to a Mixture of Single and Multi-consumer Queues 
   * with a Timeout of 100 Seconds
   */

  SetAgent(agent_list[0], "app1", "SCOTT.MCQ1", errhp);
  SetAgent(agent_list[1], "app2", "SCOTT.MCQ2", errhp);
  SetAgent(agent_list[2], (text *)0, "SCOTT.SCQ3", errhp);
 
  checkerr(errhp, OCIAQListen(svchp, errhp, agent_list, 3, 100, &agent, 0));

  printf("MESSAGE for :- \n");             
  GetAgent(agent, errhp);
  printf("\n");

}