Вот этот примерчик.
Создание очереди:
| Код |
Rem ------------------------------------------------------
REM create queue table for persistent multiple consumers
Rem ------------------------------------------------------
connect pubsub/pubsub;
Rem Create or replace a queue table
begin
DBMS_AQADM.CREATE_QUEUE_TABLE(
QUEUE_TABLE=>'pubsub.raw_msg_table',
MULTIPLE_CONSUMERS => TRUE,
QUEUE_PAYLOAD_TYPE =>'RAW',
COMPATIBLE => '8.1.5');
end;
/
Rem ------------------------------------------------------
Rem Create a persistent queue for publishing messages
Rem ------------------------------------------------------
Rem Create a queue for logon events
begin
DBMS_AQADM.CREATE_QUEUE(QUEUE_NAME=>'pubsub.logon',
QUEUE_TABLE=>'pubsub.raw_msg_table',
COMMENT=>'Q for error triggers');
end;
/
Rem ------------------------------------------------------
Rem Start the queue
Rem ------------------------------------------------------
begin
DBMS_AQADM.START_QUEUE('pubsub.logon');
end;
/
Rem ------------------------------------------------------
Rem define new_enqueue for convenience
Rem ------------------------------------------------------
create or replace procedure new_enqueue(queue_name in varchar2,
payload in raw ,
correlation in varchar2 := NULL,
exception_queue in varchar2 := NULL)
as
enq_ct dbms_aq.enqueue_options_t;
msg_prop dbms_aq.message_properties_t;
enq_msgid raw(16);
userdata raw(1000);
begin
msg_prop.exception_queue := exception_queue;
msg_prop.correlation := correlation;
userdata := payload;
DBMS_AQ.ENQUEUE(queue_name,enq_ct, msg_prop,userdata,enq_msgid);
end;
/
Rem ------------------------------------------------------
Rem add subscriber with rule based on current user name,
Rem using correlation_id
Rem ------------------------------------------------------
declare
subscriber sys.aq$_agent;
begin
subscriber := sys.aq$_agent('SNOOP', null, null);
dbms_aqadm.add_subscriber(queue_name => 'pubsub.logon',
subscriber => subscriber,
rule => 'CORRID = ''SCOTT'' ');
end;
/
Rem ------------------------------------------------------
Rem create a trigger on logon on database
Rem ------------------------------------------------------
Rem create trigger on after logon
create or replace trigger systrig2
AFTER LOGON
ON DATABASE
begin
new_enqueue('pubsub.logon', hextoraw('9999'), dbms_standard.login_user);
end;
/
|
Код программы:
| Код |
ub4 namespace = OCI_SUBSCR_NAMESPACE_AQ;
/* callback function for notification of logon of user 'scott' on database */
ub4 notifySnoop(ctx, subscrhp, pay, payl, desc, mode)
dvoid *ctx;
OCISubscription *subscrhp;
dvoid *pay;
ub4 payl;
dvoid *desc;
ub4 mode;
{
printf("Notification : User Scott Logged on\n");
}
int main()
{
OCISession *authp = (OCISession *) 0;
OCISubscription *subscrhpSnoop = (OCISubscription *)0;
/*****************************************************
Initialize OCI Process/Environment
Initialize Server Contexts
Connect to Server
Set Service Context
******************************************************/
/* Registration Code Begins */
/* Each call to initSubscriptionHn allocates
and Initialises a Registration Handle */
initSubscriptionHn( &subscrhpSnoop, /* subscription handle */
"PUBSUB.SNOOP:ADMIN", /* subscription name */
/* <queue_name>:<agent_name> */
(dvoid*)notifySnoop); /* callback function */
/*****************************************************
The Client Process does not need a live Session for Callbacks
End Session and Detach from Server
******************************************************/
OCISessionEnd ( svchp, errhp, authp, (ub4) OCI_DEFAULT);
/* detach from server */
OCIServerDetach( srvhp, errhp, OCI_DEFAULT);
while (1) /* wait for callback */
sleep(1);
}
void initSubscriptionHn (subscrhp,
subscriptionName,
func)
OCISubscription **subscrhp;
char* subscriptionName;
dvoid * func;
{
/* allocate subscription handle */
(void) OCIHandleAlloc((dvoid *) envhp, (dvoid **)subscrhp,
(ub4) OCI_HTYPE_SUBSCRIPTION,
(size_t) 0, (dvoid **) 0);
/* set subscription name in handle */
(void) OCIAttrSet((dvoid *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
(dvoid *) subscriptionName,
(ub4) strlen((char *)subscriptionName),
(ub4) OCI_ATTR_SUBSCR_NAME, errhp);
/* set callback function in handle */
(void) OCIAttrSet((dvoid *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
(dvoid *) func, (ub4) 0,
(ub4) OCI_ATTR_SUBSCR_CALLBACK, errhp);
(void) OCIAttrSet((dvoid *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
(dvoid *) 0, (ub4) 0,
(ub4) OCI_ATTR_SUBSCR_CTX, errhp);
/* set namespace in handle */
(void) OCIAttrSet((dvoid *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
(dvoid *) &namespace, (ub4) 0,
(ub4) OCI_ATTR_SUBSCR_NAMESPACE, errhp);
checkerr(errhp, OCISubscriptionRegister(svchp, subscrhp, 1, errhp,
OCI_DEFAULT));
}
|
|
