9 OCI Programming Advanced Topics

OCI Connection Pooling Concepts

Oracle Database has several transaction monitoring capabilities such as the fine-grained management of database sessions and connections. Fine-grained management of database sessions is done by separating the notion of database sessions (user handles) from connections (server handles). By using OCI calls for session switching and session migration, an application server or transaction monitor can multiplex several sessions over fewer physical connections, thus achieving a high degree of scalability by pooling connections and back-end Oracle server processes.

The connection pool itself is normally configured with a shared pool of physical connections, translating to a back-end server pool containing an identical number of dedicated server processes.

The number of physical connections is less than the number of database sessions in use by the application. The number of physical connections and back-end server processes are also reduced by using connection pooling. Thus many more database sessions can be multiplexed.

Multiple Connection Pools

You can use the advanced concept of multiple connection pools for different database connections. Multiple connection pools can also be used when different priorities are assigned to users. Different service-level guarantees can be implemented using connection pooling.

Figure 9-1 illustrates OCI connection pooling.

Figure 9-1 OCI Connection Pooling

Description of "Figure 9-1 OCI Connection Pooling"

OCI Calls for Connection Pooling

To use connection pooling in your application, you must:

1. Allocate the Pool Handle

2. Create the Connection Pool

3. Log On to the Database

4. Deal with SGA Limitations in Connection Pooling

5. Log Off from the Database

6. Destroy the Connection Pool

7. Free the Pool Handle

OCICPool *poolhp;
OCIHandleAlloc((void *) envhp, (void **) &poolhp, OCI_HTYPE_CPOOL, 
                      (size_t) 0, (void **) 0));

OCIConnectionPoolCreate((OCIEnv *)envhp,
                   (OCIError *)errhp, (OCICPool *)poolhp,
                   &poolName, &poolNameLen,
                   (text *)database,strlen(database),
                   (ub4) connMin, (ub4) connMax, (ub4) connIncr,
                   (text *)poolUsername,strlen(poolUserLen),
                   (text *)poolPassword,strlen(poolPassLen),
                   OCI_DEFAULT));

 for (i = 0; i < MAXTHREADS; ++i)
  {
    checkerr(errhp, OCILogoff((void *) svchp[i], errhp));
  }
  checkerr(errhp, OCIConnectionPoolDestroy(poolhp, errhp, OCI_DEFAULT));
  checkerr(errhp, OCIHandleFree((void *)poolhp, OCI_HTYPE_CPOOL));

Examples of OCI Connection Pooling

Examples of connection pooling in tested complete programs can be found in cdemocp.c and cdemocpproxy.c in directory demo.

Functionality of OCI Session Pooling

Session pooling can perform the following tasks:

• Create, maintain, and manage a pool of stateless sessions transparently.

• Provide an interface for the application to create a pool and specify the minimum, increment, and maximum number of sessions in the pool.

• Provide an interface for the user to obtain and release a default or tagged session to the pool. A tagged session is one with certain client-defined properties.

• Allow the application to dynamically change the number of minimum and maximum number of sessions.

• Provide a mechanism to always maintain an optimum number of open sessions, by closing sessions that have been idle for a very long time, and creating sessions when required.

• Allow for session pooling with authentication.

Homogeneous and Heterogeneous Session Pools

A session pool can be either homogeneous or heterogeneous. Homogeneous session pooling means that sessions in the pool are alike for authentication (they have the same user name, password, and privileges). Heterogeneous session pooling means that you must provide authentication information because the sessions can have different security attributes and privileges.

Using Tags in Session Pools

The tags provide a way for users to customize sessions in the pool. A client can get a default or untagged session from a pool, set certain attributes on the session (such as NLS settings), and return the session to the pool, labeling it with an appropriate tag in the OCISessionRelease() call.

The user, or some other user, can request a session with the same tags to have a session with the same attributes, and can do so by providing the same tag in the OCISessionGet() call.

OCI Handles for Session Pooling

The following handle types are for session pooling.

OCISPool *spoolhp;
OCIHandleAlloc((void *) envhp, (void **) &spoolhp, OCI_HTYPE_SPOOL, 
                        (size_t) 0, (void **) 0));

OCIAuthInfo *authp;
OCIHandleAlloc((void *) envhp, (void **) &authp, OCI_HTYPE_AUTHINFO, 
                      (size_t) 0, (void **) 0));

Using OCI Session Pooling

The steps in writing a simple session pooling application that uses a user name and password are as follows:

1. Allocate the session pool handle using OCIHandleAlloc() for an OCISPool handle. Multiple session pools can be created for an environment handle.

2. Create the session pool using OCISessionPoolCreate() with mode set to OCI_DEFAULT (for a new session pool). See the function for a discussion of the other modes.

3. Loop for each thread. Create the thread with a function that does the following:

   1. Allocates an authentication information handle of type OCIAuthInfo using OCIHandleAlloc()

   2. Sets the user name and password in the authentication information handle using OCIAttrSet()

   3. Gets a pooled session using OCISessionGet() with mode set to OCI_SESSGET_SPOOL

   4. Performs the transaction

   5. Allocates the handle

   6. Prepares the statement

      Note:

      When using service contexts obtained from OCI session pool, you are required to use the service context returned by OCISessionGet() (or OCILogon2()), and not create other service contexts outside of these calls.

      Any statement handle obtained using OCIStmtPrepare2() with the service context should be subsequently used only in conjunction with the same service context, and never with a different service context.

   7. Executes the statement

   8. Commits or rolls back the transactions

   9. Releases the session (log off) with OCISessionRelease()

   10. Frees the authentication information handle with OCIHandleFree()

   11. Ends the loop for each thread

4. Destroy the session pool using OCISessionPoolDestroy().

OCI Calls for Session Pooling

Here are the usages for OCI calls for session pooling. OCI provides calls for session pooling to perform the following tasks:

• Allocate the Pool Handle

• Create the Connection Pool

• Log On to the Database

• Log Off from the Database

• Destroy the Connection Pool

• Free the Pool Handle

OCISPool *poolhp; 
OCIHandleAlloc((void *) envhp, (void **) &poolhp, OCI_HTYPE_SPOOL, (size_t) 0,
               (void **) 0));

OCISessionPoolCreate(envhp, errhp, poolhp, (OraText **)&poolName, 
              (ub4 *)&poolNameLen, database, 
              (ub4)strlen((const signed char *)database),
              sessMin, sessMax, sessIncr,
              (OraText *)appusername,
              (ub4)strlen((const signed char *)appusername),
              (OraText *)apppassword,
              (ub4)strlen((const signed char *)apppassword),
              OCI_DEFAULT);

OCISessionPoolDestroy(poolhp, errhp, OCI_DEFAULT);

OCIHandleFree((void *)poolhp, OCI_HTYPE_SPOOL);

Example of OCI Session Pooling

For an example of session pooling in a tested complete program, see cdemosp.c in directory demo.

Functions for Session Creation

OCI offers the following functions for session creation:

• OCILogon()

  OCILogon() is the simplest way to get an OCI session. The advantage is ease of obtaining an OCI service context. The disadvantage is that you cannot perform any advance OCI operations, such as session migration, proxy authentication, or using a connection pool or a session pool.

  See Also:

  • "OCILogon()"

  • "Application Initialization, Connection, and Session Creation"

• OCILogon2()

  OCILogon2() includes the functionality of OCILogon() to get a session. This session may be a new one with a new underlying connection, or one that is started over a virtual connection from an existing connection pool, or one from an existing session pool. The mode parameter value that the function is called with determines its behavior.

  The user cannot modify the attributes (except OCI_ATTR_STMTCACHESIZE) of the service context returned by OCI.

  See Also:

  "OCILogon2()"

• OCISessionBegin()

  OCISessionBegin() supports all the various options of an OCI session, such as proxy authentication, getting a session from a connection pool or a session pool, external credentials, and migratable sessions. This is the lowest level call, where all handles must be explicitly allocated and all attributes set. OCIServerAttach() must be called before this call.

  See Also:

  "OCISessionBegin()"

• OCISessionGet()

  OCISessionGet() is now the recommended method to get a session. This session may be a new one with a new underlying connection, or one that is started over a virtual connection from an existing connection pool, or one from an existing session pool. The mode parameter value that the function is called with determines its behavior. This works like OCILogon2() but additionally enables you to specify tags for obtaining specific sessions from the pool.

  See Also:

  "OCISessionGet()"

Choosing Between Different Types of OCI Sessions

OCI includes the following types of sessions:

• Basic OCI sessions

  The basic OCI session works by using user name and password over a dedicated OCI server handle. This is the no-pool mechanism. See When to Use Connection Pooling, Session Pooling, or Neither for information of when to use it.

  If authentication is obtained through external credentials, then a user name or password is not required.

• Session pool sessions

  Session pool sessions are from the session pool cache. Some sessions may be tagged. These are stateless sessions. Each OCISessionGet() and OCISessionRelease() call gets and releases a session from the session cache. This saves the server from creating and destroying sessions.

  See When to Use Connection Pooling, Session Pooling, or Neither on connection pool sessions versus session pooling sessions versus no-pooling sessions.

• Connection pool sessions

  Connection pool sessions are created using OCISessionGet() and OCISessionBegin() calls from an OCI connection pool. There is no session cache as these are stateful sessions. Each call creates a new session, and the user is responsible for terminating these sessions.

  The sessions are automatically migratable between the server handles of the connection pool. Each session can have user name and password or be a proxy session. See When to Use Connection Pooling, Session Pooling, or Neither on connection pool sessions versus session pooling sessions versus no-pooling sessions.

• Sessions sharing a server handle

  You can multiplex several OCI sessions over a few physical connections. The application does this manually by having the same server handle for these multiple sessions. It is preferred to have the session multiplexing details be left to OCI by using the OCI connection pool APIs.

• Proxy sessions

  Proxy sessions are useful if the password of the client must be protected from the middle tier. Proxy sessions can also be part of an OCI connection pool or an OCI session pool.

  See Also:

  "Middle-Tier Applications in OCI"

• Migratable Sessions

With transaction handles being migratable, there should be no need for applications to use migratable sessions, instead use OCI connection pooling.

Statement Caching Without Session Pooling in OCI

To perform statement caching without session pooling, users perform the usual OCI steps to log on. The call to obtain a session has a mode that specifies whether statement caching is enabled for the session. Initially the statement cache is empty. Developers try to find a statement in the cache using the statement text. If the statement exists, the API returns a previously prepared statement handle; otherwise, it returns a newly prepared statement handle.

The application developer can perform binds and defines and then simply execute and fetch the statement before returning the statement to the cache. If the statement handle is not found in the cache, the developer must set different attributes on the handle in addition to the other steps.

OCIStmtPrepare2() takes a mode that determines if the developer wants a prepared statement handle or a null statement handle if the statement is not found in the cache.

The pseudocode looks like this:

OCISessionBegin( userhp, ... OCI_STMT_CACHE)  ;
OCIAttrset(svchp, userhp, ...);  /* Set the user handle in the service context */
OCIStmtPrepare2(svchp, &stmthp, stmttext, key, ...);
OCIBindByPos(stmthp, ...);
OCIDefineByPos(stmthp, ...);
OCIStmtExecute(svchp, stmthp, ...);
OCIStmtFetch2(svchp, ...);
OCIStmtRelease(stmthp, ...);
...

Statement Caching with Session Pooling in OCI

For statement caching with session pooling, the concepts remain the same, except that the statement cache is enabled at the session pool layer rather than at the session layer.

The attribute OCI_ATTR_SPOOL_STMTCACHESIZE sets the default statement cache size for each of the sessions in the session pool. It is set on the OCI_HTYPE_SPOOL handle. The statement cache size for a particular session in the pool can be overridden at any time by using OCI_ATTR_STMTCACHESIZE on that session. The value of OCI_ATTR_SPOOL_STMTCACHESIZE can be changed at any time. You can use this attribute to enable or disable statement caching at the pool level, after creation, just as attribute OCI_ATTR_STMTCACHESIZE (on the service context) is used to enable or disable statement caching at the session level. This change is reflected on individual sessions in the pool, when they are provided to a user. Tagged sessions are an exception to this behavior. This is explained later in this section.

Enabling or disabling of statement caching is allowed on individual pooled sessions as it is on nonpooled sessions.

A user can enable statement caching on a session retrieved from a non-statement cached pool in an OCISessionGet() or OCILogon2() call by specifying OCI_SESSGET_STMTCACHE or OCI_LOGON2_STMTCACHE, respectively, in the mode argument.

When a user asks for a session from a session pool, the statement cache size for that session defaults to that of the pool. This may also mean enabling or disabling statement caching in that session. For example, if a pooled session (Session A) has statement caching enabled, and statement caching is turned off in the pool, and a user asks for a session, and Session A is returned, then statement caching is turned off in Session A. As another example, if Session A in a pool does not have statement caching enabled, and statement caching at the pool level is turned on, then before returning Session A to a user, statement caching on Session A with size equal to that of the pool is turned on.

This does not hold true if a tagged session is asked for and retrieved. In this case, the size of the statement cache is not changed. Consequently, it is not turned on or off. Moreover, if the user specifies mode OCI_SESSGET_STMTCACHE in the OCISessionGet() call, this is ignored if the session is tagged. In our earlier example, if Session A was tagged, then it is returned as is to the user.

Rules for Statement Caching in OCI

Here are some rules to follow for statement caching in OCI:

• Use the function OCIStmtPrepare2() instead of OCIStmtPrepare(). If you are using OCIStmtPrepare(), you are strongly urged not to use a statement handle across different service contexts. Doing so raises an error if the statement has been obtained by OCIStmtPrepare2(). Migration of a statement handle to a new service context actually closes the cursor associated with the old session and therefore no sharing is achieved. Client-side sharing is also not obtained, because OCI frees all buffers associated with the old session when the statement handle is migrated.

• You are required to keep one service context per session. Any statement handle obtained using OCIStmtPrepare2() with a certain service context should be subsequently used only in conjunction with the same service context, and never with a different service context.

• A call to OCIStmtPrepare2(), even if the session does not have a statement cache, also allocates the statement handle. Therefore, applications using only OCIStmtPrepare2() must not call OCIHandleAlloc() for the statement handle.

• A call to OCIStmtPrepare2() must be followed by a call to OCIStmtRelease() after the user is done with the statement handle. If statement caching is used, this releases the statement to the cache. If statement caching is not used, the statement is deallocated. Do not call OCIHandleFree() to free the memory.

• If the call to OCIStmtPrepare2() is made with the OCI_PREP2_CACHE_SEARCHONLY mode and a NULL statement was returned (statement was not found), the subsequent call to OCIStmtRelease() is not required and must not be performed.

• Do not call OCIStmtRelease() for a statement that was prepared using OCIStmtPrepare().

• The statement cache has a maximum size (number of statements) that can be modified by an attribute on the service context, OCI_ATTR_STMTCACHESIZE. The default value is 20. This attribute can also be used to enable or disable statement caching for the session, pooled or nonpooled. If OCISessionBegin() is called without the mode set as OCI_STMT_CACHE, then OCI_ATTR_STMTCACHESIZE can be set on the service context to a nonzero attribute to turn on statement caching. If statement caching is not turned on at the session pool level, OCISessionGet() returns a non-statement cache-enabled session. You can use OCI_ATTR_STMTCACHESIZE to turn the caching on. Similarly, you can use the same attribute to turn off statement caching by setting the cache size to zero.

• You can tag a statement at the release time so that the next time you can request a statement of the same tag. The tag is used to search the cache. An untagged statement (tag is NULL) is a special case of a tagged statement. Two statements are considered different if they differ in their tags, or if one is untagged and the other is not.

  See Also:

  • "Statement Functions" on page 17-2

  • "Service Context Handle Attributes" on page A-10

  • "Session Pool Handle Attributes" on page A-29

Bind and Define Optimization in Statement Caching

To avoid repeated bind and define operations on statements in the cache by the application, the application can register an opaque context with a statement taken from the statement cache and register a callback function with the service context. The application data such as bind and define buffers can be enclosed in the opaque context. This context is registered with the statement the first time it is taken from the cache. When a statement is taken from the cache the second time and onwards, the application can reuse the bind and define buffers, that it had registered with that statement. It is still the application's responsibility to manage the bind and defines. It can reuse both the bind and define data and the buffers, or it can change only the data and reuse the buffers, or it can free and reallocate the buffers if the current size is not enough. In the last case, it must rebind and redefine. To clean up the memory allocated by the application toward these bind and define buffers, the callback function is called during aging out of the statement or purging of the whole cache as part of session closure. The callback is called for every statement being purged. The application frees the memory and does any other cleanup required, inside the callback function. Example 9-1 shows the pseudocode.

Get the statement using OCIStmtPrepare2(...)
 
Get the opaque context from the statement if it exists
 
If opaque context does not exist
 
{
 
  Allocate fetch buffers, do the OCIBindByPos, OCIDefineByPos, and so forth
 
  Enclose the buffer addresses inside a context and set the context and
  callback function on the statement
 
}
Execute/Fetch using the statement, and process the data in the fetch buffers. 
 
OCIStmtRelease() that statement
 
Next OCIStmtPrepare2()
 
OCIAttrGet() opaque application context from statement handle
 
Execute/Fetch using the statement and process the data in the fetch buffers.
 
OCIStmtRelease()
 
. . .
 
void callback_fn (context, statement, mode)
 
{
 
   /* mode= OCI_CBK_STMTCACHE_STMTPURGE means this was called when statement was
      aging out of the statement cache or if the session is ended */
 
  <free the buffers in the context.>
 
}

OCI Statement Caching Code Example

See cdemostc.c in directory demo for a working example of statement caching.

Registering User Callbacks in OCI

An application can register user callback libraries with the OCIUserCallbackRegister() function. Callbacks are registered in the context of the environment handle. An application can retrieve information about callbacks registered with a handle with the OCIUserCallbackGet() function.

A user-defined callback is a subroutine that is registered against an OCI call and an environment handle. It can be specified to be either an entry callback, a replacement callback, or an exit callback.

• If it is an entry callback, it is called when the program enters the OCI function.

• Replacement callbacks are executed after entry callbacks. If the replacement callback returns a value of OCI_CONTINUE, then a subsequent replacement callback or the normal OCI-specific code is executed. If a replacement callback returns anything other than OCI_CONTINUE, then subsequent replacement callbacks and the OCI code do not execute.

• After a replacement callback returns something other than OCI_CONTINUE, or an OCI function successfully executes, program control transfers to the exit callback (if one is registered).

If a replacement or exit callback returns anything other than OCI_CONTINUE, then the return code from the callback is returned from the associated OCI call.

A user callback can return OCI_INVALID_HANDLE when either an invalid handle or an invalid context is passed to it.

OCIUserCallbackRegister( hndlp, 
                         OCI_HTYPE_ENV, 
                         errh, 
                         stmtprep_entry_dyncbk_fn, 
                         dynamic_context, 
                         OCI_FNCODE_STMTPREPARE,
                         OCI_UCBTYPE_ENTRY
                         (OCIUcb*) NULL);

typedef sword (*OCIUserCallback)
     (void *ctxp,      /* context for the user callback*/
      void *hndlp,     /* handle for the callback, env handle for now */
      ub4 type,         /* type of handlp, OCI_HTYPE_ENV for this release */
      ub4 fcode,        /* function code of the OCI call */
      ub1 when,         /* type of the callback, entry or exit */
      sword returnCode, /* OCI return code */
      ub4 *errnop,      /* Oracle error number */
      va_list arglist); /* parameters of the oci call */

OCIXyzCall()
{
 Acquire mutexes on handles;
 retCode = OCI_SUCCESS;
 errno = 0;
 for all ENTRY callbacks do
  {
     
     EntryretCode = (*entryCallback)(..., retcode, &errno, ...);
     if (retCode != OCI_CONTINUE)
      {
         set errno in error handle or environment handle;
         retCode = EntryretCode;
       }
   }
  for all REPLACEMENT callbacks do
  {
   retCode = (*replacementCallback) (..., retcode, &errno, ...);
   if (retCode != OCI_CONTINUE)
      {
       set errno in error handle or environment handle
       goto executeEXITCallback;
       }
   }

   retCode = return code for XyzCall; /* normal processing of OCI call */

   errno = error number from error handle or env handle;

 executeExitCallback:
   for all EXIT callbacks do
   {
       exitRetCode = (*exitCallback)(..., retCode, &errno,...);
       if (exitRetCode != OCI_CONTINUE)
       {
           set errno in error handle or environment handle;
           retCode = exitRetCode;
       }
   }
    release mutexes;
    return retCode
}

setenv ORA_OCI_UCBPKG mypkg

setenv ORA_OCI_UCBPKG "mypkg;yourpkg;oraclepkg;sunpkg;msoftpkg"

sword fooInit(metaCtx, libCtx, argfmt, argc, argv)
      void *     metaCtx;         /* The metacontext */
      void *     libCtx;          /* The context for this package. */
      ub4        argfmt;          /* package argument format */
      sword      argc;            /* package arg count*/
      void *     argv[];          /* package arguments */
{
  return  (OCISharedLibInit(metaCtx, libCtx, argfmt, argc, argv,
                            fooEnvCallback));
}

typedef sword (*OCIEnvCallbackType)(OCIEnv *env, ub4 mode,
                                    size_t xtramem_sz, void *usrmemp,
                                    OCIUcb *ucbDesc);

setenv ORA_OCI_UCBPKG "pkg1;pkg2;pkg3;pkg4;pkg5" 

pkgNInit(void *metaCtx, void *libCtx, ub4 argfmt, sword argc, void **argv)
{
  return OCISharedLibInit(metaCtx, libCtx, argfmt, argc, argv, pkgNEnvCallback);
}

pkgNEnvCallback(OCIEnv *env, ub4 mode, size_t xtramemsz,
                                void *usrmemp, OCIUcb *ucbDesc)
{
  OCIHandleAlloc((void *)env, (void **)&errh, OCI_HTYPE_ERROR, (size_t) 0,
        (void **)NULL);

  OCIUserCallbackRegister(env, OCI_HTYPE_ENV, errh, pkgN_entry_callback_fn,
        pkgNctx, OCI_FNCODE_STMTPREPARE, OCI_UCBTYPE_ENTRY, ucbDesc);

  OCIUserCallbackRegister(env, OCI_HTYPE_ENV, errh, pkgN_replace_callback_fn,
        pkgNctx, OCI_FNCODE_STMTPREPARE, OCI_UCBTYPE_REPLACE, ucbDesc);

  OCIUserCallbackRegister(env, OCI_HTYPE_ENV, errh, pkgN_exit_callback_fn,
        pkgNctx, OCI_FNCODE_STMTPREPARE, OCI_UCBTYPE_EXIT, ucbDesc);

  return OCI_CONTINUE;
}
 

  OCIUserCallbackRegister(env, OCI_HTYPE_ENV, errh, static_entry_callback_fn,
        pkgNctx, OCI_FNCODE_STMTPREPARE, OCI_UCBTYPE_ENTRY, (OCIUcb *)NULL);

  OCIUserCallbackRegister(env, OCI_HTYPE_ENV, errh, static_replace_callback_fn,
        pkgNctx, OCI_FNCODE_STMTPREPARE, OCI_UCBTYPE_REPLACE, (OCIUcb *)NULL);

  OCIUserCallbackRegister(env, OCI_HTYPE_ENV, errh, static_exit_callback_fn,
        pkgNctx, OCI_FNCODE_STMTPREPARE, OCI_UCBTYPE_EXIT, (OCIUcb *)NULL);
 

static_entry_callback_fn() 
pkg1_entry_callback_fn() 
pkg2_entry_callback_fn() 
pkg3_entry_callback_fn() 
pkg4_entry_callback_fn() 
pkg5_entry_callback_fn() 
 
static_replace_callback_fn() 
 pkg1_replace_callback_fn() 
  pkg2_replace_callback_fn() 
   pkg3_replace_callback_fn() 
    pkg4_replace_callback_fn() 
     pkg5_replace_callback_fn() 
 
      OCI code for OCIStmtPrepare call 
 
pkg5_exit_callback_fn() 
pkg4_exit_callback_fn() 
pkg3_exit_callback_fn() 
pkg2_exit_callback_fn() 
pkg1_exit_callback_fn()

static_exit_callback_fn()

OCI Callbacks from External Procedures

There are several OCI functions that you can use as callbacks from external procedures. These functions are listed in Chapter 20. For information about writing C subroutines that can be called from PL/SQL code, including a list of which OCI calls you can use and some example code, see Oracle Database Development Guide.

Configuring Transparent Application Failover

TAF can be configured on both the client side and the server side. If both are configured, server-side settings take precedence.

Configure TAF on the client side by including the FAILOVER_MODE parameter in the CONNECT_DATA portion of a connect descriptor.

Configure TAF on the server side by modifying the target service with the DBMS_SERVICE.MODIFY_SERVICE packaged procedure.

An initial attempt at failover may not always succeed. OCI provides a mechanism for retrying failover after an unsuccessful attempt.

Transparent Application Failover Callbacks in OCI

Because of the delay that can occur during failover, the application developer may want to inform the user that failover is in progress, and request that the user wait for notification that failover is complete. Additionally, the session on the initial instance may have received some ALTER SESSION commands. These ALTER SESSION commands are not automatically replayed on the second instance. Consequently, the developer may want to replay them on the second instance. OCIAttrSet() calls that affect the session must also be reexecuted.

To accommodate these requirements, the application developer can register a failover callback function. If failover occurs, the callback function is invoked several times while reestablishing the user's session.

The first call to the callback function occurs when the database first detects an instance connection loss. This callback is intended to allow the application to inform the user of an upcoming delay. If failover is successful, a second call to the callback function occurs when the connection is reestablished and usable.

Once the connection has been reestablished, the client may want to replay ALTER SESSION commands and inform the user that failover has happened. If failover is unsuccessful, then the callback is called to inform the application that failover cannot occur. Additionally, the callback is called each time a user handle besides the primary handle is reauthenticated on the new connection. Because each user handle represents a server-side session, the client may want to replay ALTER SESSION commands for that session.

Failover Callback Structure and Parameters

The basic structure of a user-defined application failover callback function is as follows:

sb4 appfocallback_fn ( void       * svchp, 
                       void       * envhp, 
                       void       * fo_ctx, 
                       ub4        fo_type, 
                       ub4        fo_event );

An example is provided in "Failover Callback Example" on page 9-31 for the following parameters:

svchp

The first parameter, svchp, is the service context handle. It is of type void *.

envhp

The second parameter, envhp, is the OCI environment handle. It is of type void *.

fo_ctx

The third parameter, fo_ctx, is a client context. It is a pointer to memory specified by the client. In this area the client can keep any necessary state or context. It is passed as a void *.

fo_type

The fourth parameter, fo_type, is the failover type. This lets the callback know what type of failover the client has requested. The usual values are as follows:

• OCI_FO_SESSION indicates that the user has requested only session failover.

• OCI_FO_SELECT indicates that the user has requested select failover as well.

fo_event

The last parameter is the failover event. This indicates to the callback why it is being called. It has several possible values:

• OCI_FO_BEGIN indicates that failover has detected a lost connection and failover is starting.

• OCI_FO_END indicates successful completion of failover.

• OCI_FO_ABORT indicates that failover was unsuccessful, and there is no option of retrying.

• OCI_FO_ERROR also indicates that failover was unsuccessful, but it gives the application the opportunity to handle the error and retry failover.

• OCI_FO_REAUTH indicates that you have multiple authentication handles and failover has occurred after the original authentication. It indicates that a user handle has been reauthenticated. To determine which one, the application checks the OCI_ATTR_SESSION attribute of the service context handle (which is the first parameter).

Failover Callback Registration

For the failover callback to be used, it must be registered on the server context handle. This registration is done by creating a callback definition structure and setting the OCI_ATTR_FOCBK attribute of the server handle to this structure.

The callback definition structure must be of type OCIFocbkStruct. It has two fields: callback_function, which contains the address of the function to call, and fo_ctx, which contains the address of the client context.

An example of callback registration is included as part of Example 9-9.

Failover Callback Example

This section shows an example of a simple user-defined callback function definition (see Example 9-8), failover callback registration (see Example 9-9), and failover callback unregistration (see Example 9-10).

sb4  callback_fn(svchp, envhp, fo_ctx, fo_type, fo_event)
void * svchp;
void * envhp;
void *fo_ctx;
ub4 fo_type;
ub4 fo_event;
{
switch (fo_event) 
   {
   case OCI_FO_BEGIN:
   {
     printf(" Failing Over ... Please stand by \n");
     printf(" Failover type was found to be %s \n",
                     ((fo_type==OCI_FO_SESSION) ? "SESSION" 
                     :(fo_type==OCI_FO_SELECT) ? "SELECT"
                     : "UNKNOWN!")); 
     printf(" Failover Context is :%s\n", 
                    (fo_ctx?(char *)fo_ctx:"NULL POINTER!"));
     break;
   }
   case OCI_FO_ABORT:
   {
     printf(" Failover stopped. Failover will not occur.\n");
     break;
   }
   case    OCI_FO_END:
   {
       printf(" Failover ended ...resuming services\n");
     break;
   }
   case OCI_FO_REAUTH:
   {
       printf(" Failed over user. Resuming services\n");
     break;
   }
   default:
   {
     printf("Bad Failover Event: %d.\n",  fo_event);
     break;
   }
   }
   return 0;
}

int register_callback(srvh, errh)
void *srvh; /* the server handle */
OCIError *errh; /* the error handle */
{
  OCIFocbkStruct failover;                 /*  failover callback structure */
  /* allocate memory for context */
  if (!(failover.fo_ctx = (void *)malloc(strlen("my context.")+1)))
     return(1);
  /* initialize the context. */
  strcpy((char *)failover.fo_ctx, "my context.");
  failover.callback_function = &callback_fn;
  /* do the registration */
  if (OCIAttrSet(srvh, (ub4) OCI_HTYPE_SERVER,
                (void *) &failover, (ub4) 0,
                (ub4) OCI_ATTR_FOCBK, errh)  != OCI_SUCCESS)
     return(2);
  /* successful conclusion */
  return (0);
}

OCIFocbkStruct failover;   /*  failover callback structure */
sword status;
 
  /* set the failover context to null */
  failover.fo_ctx = NULL; 
  /* set the failover callback to null */ 
  failover.callback_function = NULL; 
  /* unregister the callback */
  status = OCIAttrSet(srvhp, (ub4) OCI_HTYPE_SERVER,
                      (void *) &failover, (ub4) 0,
                      (ub4) OCI_ATTR_FOCBK, errhp);

Handling OCI_FO_ERROR

A failover attempt is not always successful. If the attempt fails, the callback function receives a value of OCI_FO_ABORT or OCI_FO_ERROR in the fo_event parameter. A value of OCI_FO_ABORT indicates that failover was unsuccessful, and no further failover attempts are possible. OCI_FO_ERROR, however, provides the callback function with the opportunity to handle the error. For example, the callback may choose to wait a specified period of time and then indicate to the OCI library that it must reattempt failover.

Consider the timeline of events presented in Table 9-1.

The callback function triggers the new failover attempt by returning a value of OCI_FO_RETRY from the function.

Example 9-11 shows a callback function that you can use to implement the failover strategy similar to the scenario described earlier. In this case, the failover callback enters a loop in which it sleeps and then reattempts failover until it is successful:

/*--------------------------------------------------------------------*/
/* the user-defined failover callback  */
/*--------------------------------------------------------------------*/
sb4  callback_fn(svchp, envhp, fo_ctx, fo_type, fo_event )
void * svchp;
void * envhp;
void *fo_ctx;
ub4 fo_type;
ub4 fo_event;
{
   OCIError *errhp;
   OCIHandleAlloc(envhp, (void **)&errhp, (ub4) OCI_HTYPE_ERROR,
              (size_t) 0, (void **) 0);
   switch (fo_event) 
   {
   case OCI_FO_BEGIN:
   {
     printf(" Failing Over ... Please stand by \n");
     printf(" Failover type was found to be %s \n",
            ((fo_type==OCI_FO_NONE) ? "NONE"
             :(fo_type==OCI_FO_SESSION) ? "SESSION" 
             :(fo_type==OCI_FO_SELECT) ? "SELECT"
             :(fo_type==OCI_FO_TXNAL) ? "TRANSACTION"
             : "UNKNOWN!")); 
     printf(" Failover Context is :%s\n", 
            (fo_ctx?(char *)fo_ctx:"NULL POINTER!"));
     break;
   }
   case OCI_FO_ABORT:
   {
     printf(" Failover aborted. Failover will not occur.\n");
     break;
   }
   case    OCI_FO_END:
   { 
       printf("\n Failover ended ...resuming services\n");
     break;
   }
   case OCI_FO_REAUTH:
   { 
       printf(" Failed over user. Resuming services\n");
     break;
   }
   case OCI_FO_ERROR:
   {
     /* all invocations of this can only generate one line. The newline
      * will be put at fo_end time.
      */
     printf(" Failover error gotten. Sleeping...");
     sleep(3);
     printf("Retrying. ");
     return (OCI_FO_RETRY);
     break;
   }
   default:
   {
     printf("Bad Failover Event: %d.\n",  fo_event);
     break;
   }
   }
   return 0;
}

OCIEvent Handle

The OCIEvent handle encapsulates the attributes from the event payload. OCI implicitly allocates this handle before calling the event callback, which can obtain the read-only attributes of the event by calling OCIAttrGet(). Memory associated with these attributes is only valid for the duration of the event callback.

OCI Failover for Connection and Session Pools

A connection pool in an instance of Oracle RAC consists of a pool of connections connected to different instances of Oracle RAC. Upon receiving the node failure notification, all the connections connected to that particular instance should be cleaned up. For the connections that are in use, OCI must close the connections: transparent application failover (TAF) occurs immediately, and those connections are reestablished. The connections that are idle and in the free list of the pool must be purged, so that a bad connection is never returned to the user from the pool.

To accommodate custom connection pools, OCI provides a callback function that can be registered on the environment handle. If registered, this callback is invoked when an HA event occurs. Session pools are treated the same way as connection pools. Note that server handles from OCI connection pools or session pools are not passed to the callback. Hence in some cases, the callback could be called with an empty list of connections.

OCI Failover for Independent Connections

No special handling is required for independent connections; all such connections that are connected to failed instances are immediately disconnected. For idle connections, TAF is engaged to reestablish the connection when the connection is used on a subsequent OCI call. Connections that are in use at the time of the failure event are broken out immediately, so that TAF can begin. Note that this applies for the "in-use" connections of connection and session pools also.

Event Callback

The event callback, of type OCIEventCallback, has the following signature:

void evtcallback_fn (void      *evtctx,
                     OCIEvent  *eventhp );

In this signature evtctx is the client context, and OCIEvent is an event handle that is opaque to the OCI library. The other input argument is eventhp, the event handle (the attributes associated with an event).

If registered, this function is called once for each event. For Oracle RAC HA events, this callback is invoked after the affected connections have been disconnected. The following environment handle attributes are used to register an event callback and context, respectively:

• OCI_ATTR_EVTCBK is of data type OCIEventCallback *. It is read-only.

• OCI_ATTR_EVTCTX is of data type void *. It is also read-only.

text *myctx = "dummy context"; /* dummy context passed to callback fn */
...
/* OCI_ATTR_EVTCBK and OCI_ATTR_EVTCTX are read-only. */
OCIAttrSet(envhp, (ub4) OCI_HTYPE_ENV, (void *) evtcallback_fn,
           (ub4) 0, (ub4) OCI_ATTR_EVTCBK, errhp);
OCIAttrSet(envhp, (ub4) OCI_HTYPE_ENV, (void *) myctx,
           (ub4) 0, (ub4) OCI_ATTR_EVTCTX, errhp);
...

Within the OCI event callback, the list of affected server handles is encapsulated in the OCIEvent handle. For Oracle RAC HA DOWN events, client applications can iterate over a list of server handles that are affected by the event by using OCIAttrGet() with attribute types OCI_ATTR_HA_SRVFIRST and OCI_ATTR_HA_SRVNEXT:

OCIAttrGet(eventhp, OCI_HTYPE_EVENT, (void *)&srvhp, (ub4 *)0,
           OCI_ATTR_HA_SRVFIRST, errhp); 
/* or, */
OCIAttrGet(eventhp, OCI_HTYPE_EVENT, (void *)&srvhp, (ub4 *)0,
           OCI_ATTR_HA_SRVNEXT, errhp);

When called with attribute OCI_ATTR_HA_SRVFIRST, this function retrieves the first server handle in the list of server handles affected. When called with attribute OCI_ATTR_HA_SRVNEXT, this function retrieves the next server handle in the list. This function returns OCI_NO_DATA and srvhp is a NULL pointer, when there are no more server handles to return.

srvhp is an output pointer to a server handle whose connection has been closed because of an HA event. errhp is an error handle to populate. The application returns an OCI_NO_DATA error when there are no more affected server handles to retrieve.

When retrieving the list of server handles that have been affected by an HA event, be aware that the connection has already been closed and many server handle attributes are no longer valid. Instead, use the user memory segment of the server handle to store any per-connection attributes required by the event notification callback. This memory remains valid until the server handle is freed.

Custom Pooling: Tagged Server Handles

The following features apply to custom pools:

• You can tag a server handle with its parent connection object if it is created on behalf of a custom pool. Use the "user memory" parameters of OCIHandleAlloc() to request that the server handle be allocated with a user memory segment. A pointer to the "user memory" segment is returned by OCIHandleAlloc().

• When an HA event occurs and an affected server handle has been retrieved, there is a means to retrieve the server handle's tag information so appropriate cleanup can be performed. The attribute OCI_ATTR_USER_MEMORY is used to retrieve a pointer to a handle's user memory segment. OCI_ATTR_USER_MEMORY is valid for all user-allocated handles. If the handle was allocated with extra memory, this attribute returns a pointer to the user memory. A NULL pointer is returned for those handles not allocated with extra memory. This attribute is read-only and is of data type void*.

Example 9-12 shows an example of event notification.

sword retval;
OCIServer *srvhp;
struct myctx {
   void *parentConn_myctx;
   uword numval_myctx;
};
typedef struct myctx myctx; 
myctx  *myctxp;
/* Allocate a server handle with user memory - pre 10.2 functionality */
if (retval = OCIHandleAlloc(envhp, (void **)&srvhp, OCI_HTYPE_SERVER,
                            (size_t)sizeof(myctx), (void **)&myctxp)
/* handle error */
myctxp->parentConn_myctx = <parent connection reference>;
 
/* In an event callback function, retrieve the pointer to the user memory */
evtcallback_fn(void *evtctx, OCIEvent *eventhp)
{ 
  myctx *ctxp = (myctx *)evtctx;
  OCIServer *srvhp;
  OCIError *errhp;
  sb4       retcode;
  retcode = OCIAttrGet(eventhp, OCI_HTYPE_SERVER, &srvhp, (ub4 *)0,
                       OCI_ATTR_HA_SRVFIRST, errhp); 
  while (!retcode) /* OCIAttrGet will return OCI_NO_DATA if no more srvhp */ 
  {  
     OCIAttrGet((void *)srvhp, OCI_HTYPE_SERVER, (void *)&ctxp,
                (ub4)0, (ub4)OCI_ATTR_USER_MEMORY, errhp);
           /* Remove the server handle from the parent connection object */
     retcode = OCIAttrGet(eventhp, OCI_HTYPE_SERVER, &srvhp, (ub4 *)0,
                          OCI_ATTR_HA_SRVNEXT, errhp);
...
  }
...
}

Determining Transparent Application Failover (TAF) Capabilities

You can have the application adjust its behavior if a connection is or is not TAF-enabled. Use OCIAttrGet() as follows to determine if a server handle is TAF-enabled:

boolean taf_capable;
...
OCIAttrGet(srvhp, (ub4) OCI_HTYPE_SERVER, (void *) &taf_capable, 
           (ub4) sizeof(taf_capable), (ub4)OCI_ATTR_TAF_ENABLED, errhp);
...

In this example, taf_capable is a Boolean variable, which this call sets to TRUE if the server handle is TAF-enabled, and FALSE if not; srvhp is an input target server handle; OCI_ATTR_TAF_ENABLED is an attribute that is a pointer to a Boolean variable and is read-only; errhp is an input error handle.

Developing Applications that Use Transaction Guard

This section describes developing OCI user applications that use Transaction Guard.

See the chapter about using Transaction Guard in Oracle Database Development Guide for more detailed information about developing applications using Transaction Guard.

For the third-party or user application to use Transaction Guard in order to be able to failover a session for OCI, it must include several major steps:

1. On receipt of an error, determine whether the error is a recoverable error - OCI_ATTR_IS_RECOVERABLE on OCI_ERROR handle. If the error is recoverable, then continue to Step 2.

   Note:

   Do not attempt to use the LTXID to check transaction outcome if the connection has not suffered a recoverable error.

2. Retrieve the LTXID associated with the failed session by using OCI_ATTR_GET() to get the OCI_ATTR_LTXID from the user session handle.

3. Reconnect to the database.

   Note:

   The new session will have a new LTXID, but you will not need it when checking the status of the original session.

4. Invoke the DBMS_APP_CONT.GET_LTXID_OUTCOME PL/SQL procedure with the LTXID obtained from the OCI_ATTR_GET() call. The original LTXID of the failed-over session is marked as forced if that LTXID has not been used. The return state tells the driver if the last transaction was COMMITTED (TRUE/FALSE) and USER_CALL_COMPLETED (TRUE/FALSE).

5. The application can replay an uncommitted transaction or return the result to the user. If the replay itself incurs an outage, then the LTXID for the replaying session is used for the DBMS_APP_CONT.GET_LTXID_OUTCOME procedure.

Topics:

• Typical Transaction Guard Usage

• Transaction Guard Examples

*/
 
#ifndef OCISP_ORACLE
# include <cdemosp.h> 
#endif
 
/* Maximum Number of threads  */ 
#define MAXTHREAD 1
static ub4 sessMin = 1;
static ub4 sessMax = 9;
static ub4 sessIncr = 2;
 
static OCIError   *errhp;
static OCIEnv     *envhp;
static OCISPool   *poolhp=(OCISPool *) 0;
static int employeeNum[MAXTHREAD];
 
static OraText *poolName;
static ub4 poolNameLen;
static CONST OraText *database = (text *)"ltxid_service";
static CONST OraText *appusername =(text *)"scott";
static CONST OraText *apppassword =(text *)"tiger";
 
static CONST char getLtxid[]= 
  ("BEGIN DBMS_APP_CONT.GET_LTXID_OUTCOME ("
   ":ltxid,:committed,:callComplete); END;");
 
static CONST char insertst1[] = 
  ("INSERT INTO EMP(ENAME, EMPNO) values ('NAME1', 1000)");
 
static void checkerr (OCIError *errhp, sword status);
static void threadFunction (dvoid *arg);
 
int main(void)
{
  int i = 0;
  sword lstat;
  int timeout =1;
  OCIEnvCreate (&envhp, OCI_THREADED, (dvoid *)0,  NULL,
                NULL, NULL, 0, (dvoid *)0);
 
  (void) OCIHandleAlloc((dvoid *) envhp, (dvoid **) &errhp, OCI_HTYPE_ERROR,
                        (size_t) 0, (dvoid **) 0);
 
  (void) OCIHandleAlloc((dvoid *) envhp, (dvoid **) &poolhp, OCI_HTYPE_SPOOL,
                        (size_t) 0, (dvoid **) 0);
 
  /* Create the session pool */
  checkerr(errhp, OCIAttrSet((dvoid *) poolhp,
           (ub4) OCI_HTYPE_SPOOL, (dvoid *) &timeout, (ub4)0, 
           OCI_ATTR_SPOOL_TIMEOUT, errhp));
 
  if (lstat = OCISessionPoolCreate(envhp, errhp,poolhp, (OraText **)&poolName, 
              (ub4 *)&poolNameLen, database, 
              (ub4)strlen((const char *)database),
              sessMin, sessMax, sessIncr,
              (OraText *)appusername,
              (ub4)strlen((const char *)appusername),
              (OraText *)apppassword,
              (ub4)strlen((const char *)apppassword),
              OCI_SPC_STMTCACHE|OCI_SPC_HOMOGENEOUS))
  {
    checkerr(errhp,lstat);
  }
 
  printf("Session Pool Created \n");
 
  /* Multiple threads using the session pool */
  {
    OCIThreadId *thrid[MAXTHREAD];
    OCIThreadHandle *thrhp[MAXTHREAD];
 
    OCIThreadProcessInit ();
    checkerr (errhp, OCIThreadInit (envhp, errhp));
    for (i = 0; i < MAXTHREAD; ++i)
    {
      checkerr (errhp, OCIThreadIdInit (envhp, errhp, &thrid[i]));
      checkerr (errhp, OCIThreadHndInit (envhp, errhp, &thrhp[i]));
    }
    for (i = 0; i < MAXTHREAD; ++i)
    {
      employeeNum[i]=i;
      /* Inserting into EMP table */
      checkerr (errhp, OCIThreadCreate (envhp, errhp, threadFunction,
                (dvoid *) &employeeNum[i], thrid[i], thrhp[i]));
    }
    for (i = 0; i < MAXTHREAD; ++i)
    {
      checkerr (errhp, OCIThreadJoin (envhp, errhp, thrhp[i]));
      checkerr (errhp, OCIThreadClose (envhp, errhp, thrhp[i]));
      checkerr (errhp, OCIThreadIdDestroy (envhp, errhp, &(thrid[i])));
      checkerr (errhp, OCIThreadHndDestroy (envhp, errhp, &(thrhp[i])));
    }
    checkerr (errhp, OCIThreadTerm (envhp, errhp));
  } /* ALL THE THREADS ARE COMPLETE */
  lstat =  OCISessionPoolDestroy(poolhp, errhp, OCI_DEFAULT);
 
  printf("Session Pool Destroyed \n");
  
  if (lstat != OCI_SUCCESS)
    checkerr(errhp, lstat);
    
  checkerr(errhp, OCIHandleFree((dvoid *)poolhp, OCI_HTYPE_SPOOL));
    
  checkerr(errhp, OCIHandleFree((dvoid *)errhp, OCI_HTYPE_ERROR));
  return 0;
 
} /* end of main () */
 
/* Inserts records into EMP table */ 
static void threadFunction (dvoid *arg)
{
  int empno = *(int *)arg;
  OCISvcCtx *svchp = (OCISvcCtx *) 0;
  OCISvcCtx *svchp2 = (OCISvcCtx *) 0;
  OCISession *embUsrhp = (OCISession *)0;
  OCIBind *bnd1p, *bnd2p, *bnd3p;
 
  OCIStmt *stmthp = (OCIStmt *)0;
  OCIStmt *getLtxidStm = (OCIStmt *)0;
  OCIError  *errhp2 = (OCIError *) 0;
  OCIAuthInfo *authp = (OCIAuthInfo *)0;
  sword lstat;
  text name[10];
 
  boolean callCompl, committed, isRecoverable;
  ub1 *myLtxid;
  ub4  myLtxidLen;
 
  ub4 numAttempts = 0;
 
  (void) OCIHandleAlloc((dvoid *) envhp, (dvoid **) &errhp2, OCI_HTYPE_ERROR,
                     (size_t) 0, (dvoid **) 0);
 
  lstat =  OCIHandleAlloc((dvoid *) envhp,
                          (dvoid **)&authp, (ub4) OCI_HTYPE_AUTHINFO,
                          (size_t) 0, (dvoid **) 0);
  if (lstat)
    checkerr(errhp2, lstat);
 
  checkerr(errhp2, OCIAttrSet((dvoid *) authp,(ub4) OCI_HTYPE_AUTHINFO, 
           (dvoid *) appusername, (ub4) strlen((char *)appusername),
           (ub4) OCI_ATTR_USERNAME, errhp2));
 
  checkerr(errhp2,OCIAttrSet((dvoid *) authp,(ub4) OCI_HTYPE_AUTHINFO, 
           (dvoid *) apppassword, (ub4) strlen((char *)apppassword),
           (ub4) OCI_ATTR_PASSWORD, errhp2));
 
restart:
  if  (lstat = OCISessionGet(envhp, errhp2, &svchp, authp,
               (OraText *)poolName, (ub4)strlen((char *)poolName), NULL, 
               0, NULL, NULL, NULL, OCI_SESSGET_SPOOL))
  {
    checkerr(errhp2,lstat);
  } 
  
  /* save the ltxid from the session in case we need to call
   *  get_ltxid_outcome to determine the transaction status.
   */
  checkerr(errhp2, OCIAttrGet(svchp, OCI_HTYPE_SVCCTX,
                              (dvoid *)&embUsrhp, (ub4 *)0,
                              (ub4)OCI_ATTR_SESSION, errhp2));
  checkerr(errhp2, OCIAttrGet(embUsrhp, OCI_HTYPE_SESSION,
                              (dvoid *)&myLtxid, (ub4 *)&myLtxidLen,
                              (ub4)OCI_ATTR_LTXID, errhp2));
  
 
  /* */
  checkerr(errhp2, OCIStmtPrepare2(svchp, &stmthp, errhp2, 
                                   (CONST OraText *)insertst1, 
                                   (ub4)sizeof(insertst1),
                                   (const oratext *)0, (ub4)0,
                                   OCI_NTV_SYNTAX, OCI_DEFAULT));
 
  if (!numAttempts)
  {
    char input[1];
 
    printf("Kill SCOTT's session now. Press ENTER when complete\n");
    gets(input);
  }
  lstat = OCIStmtExecute (svchp, stmthp, errhp2, (ub4)1, (ub4)0,
                          (OCISnapshot *)0, (OCISnapshot *)0, 
                          OCI_DEFAULT );
  if (lstat == OCI_ERROR)
  {
    checkerr(errhp2, OCIAttrGet(errhp2, OCI_HTYPE_ERROR,
                                (dvoid *)&isRecoverable, (ub4 *)0,
                                (ub4)OCI_ATTR_ERROR_IS_RECOVERABLE, errhp2));
    if (isRecoverable)
    {
 
      printf("Recoverable error occurred; checking transaction status.\n");
      /* get another session to use for the get_ltxid_outcome call */
      if  (lstat = OCISessionGet(envhp, errhp2, &svchp2, authp,
                                 (OraText *)poolName, 
                                 (ub4)strlen((char *)poolName), NULL, 
                                 0, NULL, NULL, NULL, OCI_SESSGET_SPOOL))
      {
        checkerr(errhp2,lstat);
      } 
      
      checkerr(errhp2,OCIStmtPrepare2(svchp2,&getLtxidStm, errhp2, 
                                      (CONST OraText *)getLtxid,
                                      (ub4)sizeof(getLtxid),
                                      (const oratext *)0, (ub4)0,
                                      OCI_NTV_SYNTAX, OCI_DEFAULT));
      checkerr(errhp, OCIBindByPos(getLtxidStm, &bnd1p, errhp, 1,
                                   (dvoid *) myLtxid, (sword)myLtxidLen,
                                   SQLT_BIN, (dvoid *)0,
                                   (ub2 *) 0, (ub2 *) 0, (ub4) 0, (ub4 *) 0,
                                   OCI_DEFAULT));
      checkerr(errhp, OCIBindByPos(getLtxidStm, &bnd2p, errhp, 2,
                                   (dvoid *) &committed, 
                                   (sword)sizeof(committed),
                                   SQLT_BOL, (dvoid *)0,
                                   (ub2 *) 0, (ub2 *) 0, (ub4) 0, (ub4 *) 0,
                                   OCI_DEFAULT));
      checkerr(errhp, OCIBindByPos(getLtxidStm, &bnd3p, errhp, 3,
                                   (dvoid *) &callCompl, 
                                   (sword)sizeof(callCompl),
                                   SQLT_BOL, (dvoid *)0,
                                   (ub2 *) 0, (ub2 *) 0, (ub4) 0, (ub4 *) 0,
                                   OCI_DEFAULT));
      
      checkerr(errhp2,OCIStmtExecute(svchp2, getLtxidStm, errhp2, 
                                     (ub4)1, (ub4)0,
                                     (OCISnapshot *)0, (OCISnapshot *)0, 
                                     OCI_DEFAULT ));
      checkerr(errhp2, OCISessionRelease(svchp2, errhp2, 
                                         NULL, 0, OCI_DEFAULT));
      if (committed && callCompl)
        printf("Insert successfully commited \n");
      else if (!committed)
      {
        printf("Transaction did not commit; re-executing last transaction\n");
        numAttempts++;
 
        /* As there was an outage, do not return this session to the pool */
        checkerr(errhp2, 
                 OCISessionRelease(svchp, errhp2, 
                                   NULL, 0, OCI_SESSRLS_DROPSESS));
        svchp = (OCISvcCtx *)0;
        goto restart;
      }
    }
  }
  else
  {
    checkerr(errhp2, OCITransCommit(svchp,errhp2,(ub4)0));
    printf("Transaction committed successfully\n");
  }
  if (stmthp)
    checkerr(errhp2, OCIStmtRelease((dvoid *) stmthp, errhp2,
                                    (void *)0, 0, OCI_DEFAULT));
  if (getLtxidStm)
    checkerr(errhp2, OCIStmtRelease((dvoid *) getLtxidStm, errhp2,
                                    (void *)0, 0, OCI_DEFAULT));
 
  if (svchp)
    checkerr(errhp2, OCISessionRelease(svchp, errhp2, NULL, 0, OCI_DEFAULT));
  OCIHandleFree((dvoid *)authp, OCI_HTYPE_AUTHINFO);
  OCIHandleFree((dvoid *)errhp2, OCI_HTYPE_ERROR);
 
} /* end of threadFunction (dvoid *) */
 
/* This function prints the error */
void checkerr(errhp, status)
OCIError *errhp;
sword status;
{
  text errbuf[512];
  sb4 errcode = 0;
 
  switch (status)
  {
  case OCI_SUCCESS:
    break;
  case OCI_SUCCESS_WITH_INFO:
    (void) printf("Error - OCI_SUCCESS_WITH_INFO\n");
    break;
  case OCI_NEED_DATA:
    (void) printf("Error - OCI_NEED_DATA\n");
    break;
  case OCI_NO_DATA:
    (void) printf("Error - OCI_NODATA\n");
    break;
  case OCI_ERROR:
    (void) OCIErrorGet((dvoid *)errhp, (ub4) 1, (text *) NULL, &errcode,
                       errbuf, (ub4) sizeof(errbuf), OCI_HTYPE_ERROR);
    (void) printf("Error - %.*s\n", 512, errbuf);
    break;
  case OCI_INVALID_HANDLE:
    (void) printf("Error - OCI_INVALID_HANDLE\n");
    break;
  case OCI_STILL_EXECUTING:
    (void) printf("Error - OCI_STILL_EXECUTE\n");
    break;
  case OCI_CONTINUE:
    (void) printf("Error - OCI_CONTINUE\n");
    break;
  default:
    break;
  }
}

OCI Streams Advanced Queuing Functions

The OCI library includes several functions related to Streams Advanced Queuing:

• OCIAQEnq()

• OCIAQDeq()

• OCIAQListen() (Deprecated)

• OCIAQListen2()

• OCIAQEnqArray()

• OCIAQDeqArray()

You can enqueue an array of messages to a single queue. The messages all share the same enqueue options, but each message in the array can have different message properties. You can also dequeue an array of messages from a single queue. For transaction group queues, you can dequeue all messages for a single transaction group using one call.

OCI Streams Advanced Queuing Descriptors

The following descriptors are used by OCI Streams AQ operations:

• OCIAQEnqOptions

• OCIAQDeqOptions

• OCIAQMsgProperties

• OCIAQAgent

You can allocate these descriptors with the service handle using the standard OCIDescriptorAlloc() call. The following code shows examples of this:

OCIDescriptorAlloc(svch, &enqueue_options, OCI_DTYPE_AQENQ_OPTIONS, 0, 0 ); 
OCIDescriptorAlloc(svch, &dequeue_options, OCI_DTYPE_AQDEQ_OPTIONS, 0, 0 ); 
OCIDescriptorAlloc(svch, &message_properties, OCI_DTYPE_AQMSG_PROPERTIES, 0, 0);
OCIDescriptorAlloc(svch, &agent, OCI_DTYPE_AQAGENT, 0, 0 ); 

Each descriptor has a variety of attributes that can be set or read.

Streams Advanced Queuing in OCI Versus PL/SQL

The following tables compare functions, parameters, and options for OCI Streams AQ functions and descriptors, and PL/SQL AQ functions in the DBMS_AQ package. Table 9-2 compares AQ functions.

Table 9-3 compares the parameters for the enqueue functions.

queue_name

queue_name

enqueue_options

enqueue_options

message_properties

message_properties

payload

payload

msgid

msgid

Table 9-4 compares the parameters for the dequeue functions.

queue_name

queue_name

dequeue_options

dequeue_options

message_properties

message_properties

payload

payload

msgid

msgid

Table 9-5 compares parameters for the listen functions.

agent_list

agent_list

wait

wait

agent

agent

listen_delivery_mode

Table 9-6 compares parameters for the array enqueue functions.

queue_name

queue_name

enqueue_options

enqopt

array_size

iters

message_properties_array

msgprop

payload_array

payload

msgid_array

msgid

Table 9-7 compares parameters for the array dequeue functions.

queue_name

queue_name

dequeue_options

deqopt

array_size

iters

message_properties_array

msgprop

payload_array

payload

msgid_array

msgid

Table 9-8 compares parameters for the agent attributes.

name

address

protocol

Table 9-9 compares parameters for the message properties.

priority

delay

expiration

correlation

attempts

recipient_list

exception_queue

enqueue_time

state

sender_id

transaction_group

original_msgid

delivery_mode

Table 9-10 compares enqueue option attributes.

visibility

relative_msgid

sequence_deviation

delivery_mode

Table 9-11 compares dequeue option attributes.

consumer_name

dequeue_mode

navigation

visibility

wait

msgid

correlation

delivery_mode

Buffered Messaging

Buffered messaging is a nonpersistent messaging capability within Streams AQ that was first available in Oracle Database 10g Release 2. Buffered messages reside in shared memory and can be lost if there is an instance failure. Unlike persistent messages, redo does not get written to disk. Buffered message enqueue and dequeue is much faster than persistent message operations. Because shared memory is limited, buffered messages may have to be spilled to disk. Flow control can be enabled to prevent applications from flooding the shared memory when the message consumers are slow or have stopped for some reason. The following functions are used for buffered messaging:

• "OCIAQEnq()"

• "OCIAQDeq()"

• "OCIAQListen2()"

Example 9-14 shows an example of enqueue buffered messaging.

...
OCIAQMsgProperties  *msgprop;
OCIAQEnqueueOptions *enqopt;
message              msg;    /* message is an object type */
null_message         nmsg;   /* message indicator */
...
/* Allocate descriptors */
  OCIDescriptorAlloc(envhp, (void **)&enqopt, OCI_DTYPE_AQENQ_OPTIONS, 0,
                     (void **)0));
 
 OCIDescriptorAlloc(envhp, (void **)&msgprop,OCI_DTYPE_AQMSG_PROPERTIES, 0,
                    (void **)0));
 
/* Set delivery mode to buffered */
 dlvm = OCI_MSG_BUFFERED;
 OCIAttrSet(enqopt,  OCI_DTYPE_AQENQ_OPTIONS, (void *)&dlvm, sizeof(ub2),
            OCI_ATTR_MSG_DELIVERY_MODE, errhp);
/* Set visibility to Immediate (visibility must always be immediate for buffered
   messages) */
vis = OCI_ENQ_ON_COMMIT;
 
OCIAttrSet(enqopt, OCI_DTYPE_AQENQ_OPTIONS,(void *)&vis, sizeof(ub4),
           OCI_ATTR_VISIBILITY, errhp)
 
/* Message was an object type created earlier, msg_tdo is its type
   descriptor object */
OCIAQEnq(svchp, errhp, "Test_Queue", enqopt, msgprop, msg_tdo, (void **)&mesg,
         (void **)&nmesg, (OCIRaw **)0, 0));
...

Example 9-15 shows an example of dequeue buffered messaging.

...
OCIAQMsgProperties  *msgprop;
OCIAQDequeueOptions *deqopt;
...
OCIDescriptorAlloc(envhp, (void **)&mprop, OCI_DTYPE_AQMSG_PROPERTIES, 0,
                   (void **)0));
OCIDescriptorAlloc(envhp, (void **)&deqopt, OCI_DTYPE_AQDEQ_OPTIONS, 0,
                   (void **)0);

/* Set visibility to Immediate (visibility must always be immediate for buffered
   message operations) */
vis = OCI_ENQ_ON_COMMIT;
OCIAttrSet(deqopt, OCI_DTYPE_AQDEQ_OPTIONS,(void *)&vis, sizeof(ub4),
           OCI_ATTR_VISIBILITY, errhp)
/* delivery mode is buffered */
dlvm  = OCI_MSG_BUFFERED;
OCIAttrSet(deqopt, OCI_DTYPE_AQDEQ_OPTIONS, (void *)&dlvm,  sizeof(ub2),
           OCI_ATTR_MSG_DELIVERY_MODE, errhp);
/* Set the consumer for which to dequeue the message (this must be specified
   regardless of the type of message being dequeued).
*/
consumer = "FIRST_SUBSCRIBER";
OCIAttrSet(deqopt, OCI_DTYPE_AQDEQ_OPTIONS, (void *)consumer,
           (ub4)strlen((char*)consumer), OCI_ATTR_CONSUMER_NAME, errhp);
/* Dequeue the message but do not return the payload (to simplify the code
   fragment)
*/
OCIAQDeq(svchp, errhp,  "test_queue", deqopt, msgprop, msg_tdo, (void **)0,
         (void **)0, (OCIRaw**)0, 0);
...

Publish-Subscribe Registration Functions in OCI

Registration can be done in two ways:

• Direct registration. You register directly to the database. This way is simple and the registration takes effect immediately. See "Publish-Subscribe Register Directly to the Database".

• Open registration. You register using Lightweight Directory Access Protocol (LDAP), from which the database receives the registration request. This is useful when the client cannot have a database connection (the client wants to register for a database open event while the database is down), or if the client wants to register for the same event or events in multiple databases simultaneously. See "Open Registration for Publish-Subscribe".

/* Set QOS levels */
ub4 qosflags = OCI_SUBSCR_QOS_PAYLOAD;
 
/* Set QOS flags in subscription handle */
(void) OCIAttrSet((dvoid *) subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
                 (dvoid *) &qosflags, (ub4) 0,
                 (ub4) OCI_ATTR_SUBSCR_QOSFLAGS, errhp);
 
/* Set notification grouping class */
ub4 ntfn_grouping_class = OCI_SUBSCR_NTFN_GROUPING_CLASS_TIME;
(void) OCIAttrSet((dvoid *) subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
                 (dvoid *) &ntfn_grouping_class, (ub4) 0,
                 (ub4) OCI_ATTR_SUBSCR_NTFN_GROUPING_CLASS, errhp);
 
/* Set notification grouping value of 10 minutes */
ub4 ntfn_grouping_value = 600;
(void) OCIAttrSet((dvoid *) subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
                 (dvoid *) &ntfn_grouping_value, (ub4) 0,
                 (ub4) OCI_ATTR_SUBSCR_NTFN_GROUPING_VALUE, errhp);
 
/* Set notification grouping type */
ub4 ntfn_grouping_type = OCI_SUBSCR_NTFN_GROUPING_TYPE_SUMMARY;
 
/* Set notification grouping type in subscription handle */
(void) OCIAttrSet((dvoid *) subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
                 (dvoid *) &ntfn_grouping_type, (ub4) 0,
                 (ub4) OCI_ATTR_SUBSCR_NTFN_GROUPING_TYPE, errhp);
 
/* Set namespace specific context */
(void) OCIAttrSet((dvoid *) subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
                 (dvoid *) NULL, (ub4) 0,
                 (ub4) OCI_ATTR_SUBSCR_NAMESPACE_CTX, errhp);

/* Set QOS levels */
ub4 qosflags = OCI_SUBSCR_QOS_RELIABLE | OCI_SUBSCR_QOS_PURGE_ON_NTFN;

/* Set flags in subscription handle */
(void)OCIAttrSet((void *)subscrhp, (ub4)OCI_HTYPE_SUBSCRIPTION,
               (void *)&qosflags, (ub4)0, (ub4)OCI_ATTR_SUBSCR_QOSFLAGS, errhp);

/* Set auto-expiration after 30 seconds */
ub4 timeout = 30;
(void)OCIAttrSet((void *)subscrhp, (ub4)OCI_HTYPE_SUBSCRIPTION,
                 (void *)&timeout, (ub4)0, (ub4)OCI_ATTR_SUBSCR_TIMEOUT, errhp);

ub4 port = 1581;
(void)OCIAttrSet((void *)envhp, (ub4)OCI_HTYPE_ENV, (void *)&port, (ub4)0,
                 (ub4)OCI_ATTR_SUBSCR_PORTNO, errhp);

(void)OCIAttrGet((void *)subhp, (ub4)OCI_HTYPE_ENV, (void *)&port, (ub4)0, 
                 (ub4)OCI_ATTR_SUBSCR_PORTNO, errhp);

text ipaddr[16] = "10.177.246.40";
(void)(OCIAttrSet((dvoid *) envhp, (ub4) OCI_HTYPE_ENV,
       (dvoid *) ipaddr, (ub4) strlen((const char *)ipaddr),
       (ub4) OCI_ATTR_SUBSCR_IPADDR, errhp));

Notification Callback in OCI

The client must register a notification callback that gets invoked when there is some activity on the subscription for which interest has been registered. In the AQ namespace, for instance, this occurs when a message of interest is enqueued.

This callback is typically set through the OCI_ATTR_SUBSCR_CALLBACK attribute of the subscription handle.

The callback must return a value of OCI_CONTINUE and adhere to the following specification:

typedef ub4 (*OCISubscriptionNotify) ( void            *pCtx,
                                       OCISubscription *pSubscrHp,
                                       void            *pPayload,
                                       ub4             iPayloadLen,
                                       void            *pDescriptor,
                                       ub4             iMode);

The parameters are described as follows:

pCtx (IN)

A user-defined context specified when the callback was registered.

pSubscrHp (IN)

The subscription handle specified when the callback was registered.

pPayload (IN)

The payload for this notification. Currently, only ub1 * (a sequence of bytes) for the payload is supported.

iPayloadLen (IN)

The length of the payload for this notification.

pDescriptor (IN)

The namespace-specific descriptor. Namespace-specific parameters can be extracted from this descriptor. The structure of this descriptor is opaque to the user and its type is dependent on the namespace.

The attributes of the descriptor are namespace-specific. For Advanced Queuing (AQ), the descriptor is OCI_DTYPE_AQNFY. For the AQ namespace, the count of notifications received in the group is provided in the notification descriptor. The attributes of pDescriptor are:

• Notification flag (regular = 0, timeout = 1, or grouping notification = 2) - OCI_ATTR_NFY_FLAGS

• Queue name - OCI_ATTR_QUEUE_NAME

• Consumer name - OCI_ATTR_CONSUMER_NAME

• Message ID - OCI_ATTR_NFY_MSGID

• Message properties - OCI_ATTR_MSG_PROP

• Count of notifications received in the group - OCI_ATTR_AQ_NTFN_GROUPING_COUNT

• The group, an OCI collection - OCI_ATTR_AQ_NTFN_GROUPING_MSGID_ARRAY

iMode (IN)

Call-specific mode. The only valid value is OCI_DEFAULT. This value executes the default call.

Example 9-17 shows how to use AQ grouping notification attributes in a notification callback.

ub4 notifyCB1(void *ctx, OCISubscription *subscrhp, void *pay, ub4 payl,
              void *desc, ub4 mode)
{
 oratext            *subname;
 ub4                 size;
 OCIColl            *msgid_array = (OCIColl *)0;
 ub4                 msgid_cnt = 0;
 OCIRaw             *msgid;
 void              **msgid_ptr;
 sb4                 num_msgid = 0;
 void               *elemind = (void *)0;
 boolean             exist;
 ub2                 flags;
 oratext            *hexit = (oratext *)"0123456789ABCDEF";
 ub4                 i, j;
 
 /* get subscription name */
 OCIAttrGet(subscrhp, OCI_HTYPE_SUBSCRIPTION, (void *)&subname, &size,
            OCI_ATTR_SUBSCR_NAME,ctxptr->errhp);
 
 /* print subscripton name */
 printf("Got notification for %.*s\n", size, subname);
 fflush((FILE *)stdout);
 
 /* get the #ntfns received in this group */
 OCIAttrGet(desc, OCI_DTYPE_AQNFY, (void *)&msgid_cnt, &size,
            OCI_ATTR_AQ_NTFN_GROUPING_COUNT, ctxptr->errhp);
 
 /* get the group - collection of msgids */
 OCIAttrGet(desc, OCI_DTYPE_AQNFY, (void *)&msgid_array, &size,
            OCI_ATTR_AQ_NTFN_GROUPING_MSGID_ARRAY, ctxptr->errhp);
 
 /* get notification flag - regular, timeout, or grouping notification? */
 OCIAttrGet(desc, OCI_DTYPE_AQNFY, (void *)&flags, &size,
            OCI_ATTR_NFY_FLAGS, ctxptr->errhp);
 
 /* print notification flag */
 printf("Flag: %d\n", (int)flags);
 
 /* get group (collection) size */
 if (msgid_array)
   checkerr(ctxptr->errhp,
        OCICollSize(ctxptr->envhp, ctxptr->errhp,
        CONST OCIColl *) msgid_array, &num_msgid),
        "Inside notifyCB1-OCICollSize");
 else
   num_msgid =0;
 
 /* print group size */
 printf("Collection size: %d\n", num_msgid);
 
 /* print all msgids in the group */
 for(i = 0; i < num_msgid; i++)
 {
   ub4  rawSize;                                             /* raw size    */
   ub1 *rawPtr;                                              /* raw pointer */
     /* get msgid from group */
   checkerr(ctxptr->errhp,
        OCICollGetElem(ctxptr->envhp, ctxptr->errhp,
               (OCIColl *) msgid_array, i, &exist,
               (void **)(&msgid_ptr), &elemind),
        "Inside notifyCB1-OCICollGetElem");
   msgid = *msgid_ptr;
   rawSize = OCIRawSize(ctxptr->envhp, msgid);
   rawPtr = OCIRawPtr(ctxptr->envhp, msgid);
 
   /* print msgid size */
   printf("Msgid size: %d\n", rawSize);
 
   /* print msgid in hexadecimal format */
   for (j = 0; j < rawSize; j++)
   {                                           /* for each byte in the raw */
     printf("%c", hexit[(rawPtr[j] & 0xf0) >> 4]);
     printf("%c", hexit[(rawPtr[j] & 0x0f)]);
   }
   printf("\n");
 }
 
 /* print #ntfns received in group */
 printf("Notification Count: %d\n", msgid_cnt);
 printf("\n");
 printf("***********************************************************\n");
 fflush((FILE *)stdout);
 return 1;
}

Notification Procedure

The PL/SQL notification procedure that is invoked when there is some activity on the subscription for which interest has been registered, must be created in the database.

This procedure is typically set through the OCI_ATTR_SUBSCR_RECPT attribute of the subscription handle.

Publish-Subscribe Direct Registration Example

Example 9-18 shows how system events, client notification, and Advanced Queuing work together to implement publish subscription notification.

The PL/SQL code in Example 9-18 creates all objects necessary to support a publish-subscribe mechanism under the user schema pubsub. In this code, the Agent snoop subscribes to messages that are published at logon events. Note that the user pubsub needs AQ_ADMINISTRATOR_ROLE and AQ_USER_ROLE privileges to use Advance Queuing functionality. The initialization parameter _SYSTEM_TRIG_ENABLED must be set to TRUE (the default) to enable triggers for system events. Connect as pubsub before running Example 9-18.

----------------------------------------------------------
----create queue table for persistent multiple consumers
----------------------------------------------------------
---- 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;
/
----------------------------------------------------------
---- Create a persistent queue for publishing messages
----------------------------------------------------------
---- 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;
/
----------------------------------------------------------
---- Start the queue
----------------------------------------------------------
begin
  DBMS_AQADM.START_QUEUE('pubsub.logon');
end;
/
----------------------------------------------------------
---- define new_enqueue for convenience
----------------------------------------------------------
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;
/
----------------------------------------------------------
---- add subscriber with rule based on current user name, 
---- using correlation_id
----------------------------------------------------------
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 = ''ix'' ');
end;
/
----------------------------------------------------------
---- create a trigger on logon on database
----------------------------------------------------------
---- 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;
/

----------------------------------------------------------
---- create a PL/SQL callback for notification of logon 
---- of user 'ix' on database
----------------------------------------------------------
---- 
create or replace procedure plsqlnotifySnoop(
  context raw, reginfo sys.aq$_reg_info, descr sys.aq$_descriptor,
  payload raw, payloadl number)
as
begin
 dbms_output.put_line('Notification : User ix Logged on\n');
end;
/

After the subscriptions are created, the client must register for notification using callback functions. Example 9-19 shows sample code that performs the necessary steps for registration. The initial steps of allocating and initializing session handles are omitted here for clarity.

...
static ub4 namespace = OCI_SUBSCR_NAMESPACE_AQ;

static OCISubscription *subscrhpSnoop = (OCISubscription *)0;
static OCISubscription *subscrhpSnoopMail = (OCISubscription *)0;
static OCISubscription *subscrhpSnoopServer = (OCISubscription *)0;

/* callback function for notification of logon of user 'ix' on database */

static ub4 notifySnoop(ctx, subscrhp, pay, payl, desc, mode)
    void *ctx;
    OCISubscription *subscrhp;
    void *pay;
    ub4 payl;
    void *desc;
    ub4 mode;
{
    printf("Notification : User ix Logged on\n");
  (void)OCIHandleFree((void *)subscrhpSnoop,
            (ub4) OCI_HTYPE_SUBSCRIPTION);
    return 1;
}

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

  if (status == OCI_SUCCESS) return;

  switch (status)
  {
  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 ((void *) 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_EXECUTING\n");
    break;
  case OCI_CONTINUE:
    printf("Error - OCI_CONTINUE\n");
    break;
  default:
    printf("Error - %d\n", status);
    break;
  }
}

static void initSubscriptionHn (subscrhp,
                         subscriptionName,
                         func,
                         recpproto,
                         recpaddr,
                         recppres)
OCISubscription **subscrhp;
  char * subscriptionName;
  void * func;
  ub4 recpproto;
  char * recpaddr;
  ub4 recppres;
{
    /* allocate subscription handle */
    (void) OCIHandleAlloc((void *) envhp, (void **)subscrhp,
        (ub4) OCI_HTYPE_SUBSCRIPTION,
        (size_t) 0, (void **) 0);

    /* set subscription name in handle */
    (void) OCIAttrSet((void *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
        (void *) subscriptionName,
        (ub4) strlen((char *)subscriptionName),
        (ub4) OCI_ATTR_SUBSCR_NAME, errhp);

    /* set callback function in handle */
    if (func)
      (void) OCIAttrSet((void *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
          (void *) func, (ub4) 0,
          (ub4) OCI_ATTR_SUBSCR_CALLBACK, errhp);

    /* set context in handle */
    (void) OCIAttrSet((void *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
        (void *) 0, (ub4) 0,
       (ub4) OCI_ATTR_SUBSCR_CTX, errhp);

    /* set namespace in handle */
    (void) OCIAttrSet((void *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
        (void *) &namespace, (ub4) 0,
        (ub4) OCI_ATTR_SUBSCR_NAMESPACE, errhp);

    /* set receive with protocol in handle */
    (void) OCIAttrSet((void *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
        (void *) &recpproto, (ub4) 0,
        (ub4) OCI_ATTR_SUBSCR_RECPTPROTO, errhp);

    /* set recipient address in handle */
    if (recpaddr)
      (void) OCIAttrSet((void *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
          (void *) recpaddr, (ub4) strlen(recpaddr),
          (ub4) OCI_ATTR_SUBSCR_RECPT, errhp);

    /* set receive with presentation in handle */
    (void) OCIAttrSet((void *) *subscrhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
        (void *) &recppres, (ub4) 0,
        (ub4) OCI_ATTR_SUBSCR_RECPTPRES, errhp);

    printf("Begining Registration for subscription %s\n", subscriptionName);
    checkerr(errhp, OCISubscriptionRegister(svchp, subscrhp, 1, errhp,
        OCI_DEFAULT));
   printf("done\n");

}


int main( argc, argv)
int    argc;
char * argv[];
{
    OCISession *authp = (OCISession *) 0;

/*****************************************************
Initialize OCI Process/Environment
Initialize Server Contexts
Connect to Server
Set Service Context
******************************************************/

/* Registration Code Begins */
/* Each call to initSubscriptionHn allocates
   and initializes a Registration Handle */

/* Register for OCI notification */
    initSubscriptionHn(    &subscrhpSnoop,    /* subscription handle*/
    (char*) "PUBSUB.LOGON:SNOOP", /* subscription name */
                                  /*<queue_name>:<agent_name> */
        (void*)notifySnoop,  /* callback function */
        OCI_SUBSCR_PROTO_OCI, /* receive with protocol */
        (char *)0, /* recipient address */
        OCI_SUBSCR_PRES_DEFAULT); /* receive with presentation */

/* Register for email notification */
    initSubscriptionHn(    &subscrhpSnoopMail,  /* subscription handle */
     (char*) "PUBSUB.LOGON:SNOOP",              /* subscription name */ 
                                                /* <queue_name>:<agent_name> */
        (void*)0, /* callback function */
        OCI_SUBSCR_PROTO_MAIL, /* receive with protocol */
        (char*)  "xyz@company.com", /* recipient address */
        OCI_SUBSCR_PRES_DEFAULT); /* receive with presentation */

/* Register for server to server notification */
    initSubscriptionHn(    &subscrhpSnoopServer, /* subscription handle */
       (char*)  "PUBSUB.LOGON:SNOOP",            /* subscription name */
                                                 /* <queue_name>:<agent_name> */
        (void*)0, /* callback function */
        OCI_SUBSCR_PROTO_SERVER, /* receive with protocol */
         (char*) "pubsub.plsqlnotifySnoop", /* recipient address */
        OCI_SUBSCR_PRES_DEFAULT); /* receive with presentation */

    checkerr(errhp, OCITransCommit(svchp, errhp, (ub4) OCI_DEFAULT));

/*****************************************************
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);
}

If user IX logs on to the database, the client is notified by email, and the callback function notifySnoop is called. An email notification is sent to the address xyz@company.com and the PL/SQL procedure plsqlnotifySnoop is also called in the database.

Publish-Subscribe LDAP Registration Example

Example 9-20 shows a code fragment that illustrates how to do LDAP registration. Please read all the program comments.

...

  /* To use the LDAP registration feature, OCI_EVENTS | OCI_EVENTS |OCI_USE_LDAP*/
  /*   must be set in OCIEnvCreate or OCIEnvNlsCreate */
  /*     (Note: OCIInitialize is deprecated): */
  (void) OCIInitialize((ub4) OCI_EVENTS|OCI_OBJECT|OCI_USE_LDAP, (void *)0,
                       (void * (*)(void *, size_t)) 0,
                       (void * (*)(void *, void *, size_t))0,
                       (void (*)(void *, void *)) 0 );

...

  /* set LDAP attributes in the environment handle */

  /* LDAP host name */
  (void) OCIAttrSet((void *)envhp, OCI_HTYPE_ENV, (void *)"yow", 3,
                    OCI_ATTR_LDAP_HOST, (OCIError *)errhp);

  /* LDAP server port */ 
  ldap_port = 389;
  (void) OCIAttrSet((void *)envhp, OCI_HTYPE_ENV, (void *)&ldap_port,
                    (ub4)0, OCI_ATTR_LDAP_PORT, (OCIError *)errhp);

  /* bind DN of the client, normally the enterprise user name */
  (void) OCIAttrSet((void *)envhp, OCI_HTYPE_ENV, (void *)"cn=orcladmin",
                    12, OCI_ATTR_BIND_DN, (OCIError *)errhp);

  /* password of the client */
  (void) OCIAttrSet((void *)envhp, OCI_HTYPE_ENV, (void *)"welcome",
                    7, OCI_ATTR_LDAP_CRED, (OCIError *)errhp);

  /* authentication method is "simple", username/password authentication */
  ldap_auth = 0x01;
  (void) OCIAttrSet((void *)envhp, OCI_HTYPE_ENV, (void *)&ldap_auth,
                    (ub4)0, OCI_ATTR_LDAP_AUTH, (OCIError *)errhp);

  /* administrative context: this is the DN above cn=oraclecontext */
  (void) OCIAttrSet((void *)envhp, OCI_HTYPE_ENV, (void *)"cn=acme,cn=com",
                    14, OCI_ATTR_LDAP_CTX, (OCIError *)errhp);

...

  /* retrieve the LDAP attributes from the environment handle */

  /* LDAP host */
  (void) OCIAttrGet((void *)envhp, OCI_HTYPE_ENV, (void *)&buf, 
                    &szp,  OCI_ATTR_LDAP_HOST,  (OCIError *)errhp);

  /* LDAP server port */
  (void) OCIAttrGet((void *)envhp, OCI_HTYPE_ENV, (void *)&intval, 
                    0,  OCI_ATTR_LDAP_PORT,  (OCIError *)errhp);

  /* client binding DN */
  (void) OCIAttrGet((void *)envhp, OCI_HTYPE_ENV, (void *)&buf, 
                    &szp,  OCI_ATTR_BIND_DN,  (OCIError *)errhp);

  /* client password */
  (void) OCIAttrGet((void *)envhp, OCI_HTYPE_ENV, (void *)&buf, 
                    &szp,  OCI_ATTR_LDAP_CRED,  (OCIError *)errhp);

  /* administrative context */
  (void) OCIAttrGet((void *)envhp, OCI_HTYPE_ENV, (void *)&buf, 
                    &szp,  OCI_ATTR_LDAP_CTX,  (OCIError *)errhp);

  /* client authentication method */
  (void) OCIAttrGet((void *)envhp, OCI_HTYPE_ENV, (void *)&intval, 
                    0,  OCI_ATTR_LDAP_AUTH,  (OCIError *)errhp);
  
  ...

  /* to set up the server DN descriptor in the subscription handle */

  /* allocate a server DN descriptor, dn is of type "OCIServerDNs **", 
     subhp is of type "OCISubscription **" */
  (void) OCIDescriptorAlloc((void *)envhp, (void **)dn, 
                         (ub4) OCI_DTYPE_SRVDN, (size_t)0, (void **)0);

  /* now *dn is the server DN descriptor, add the DN of the first database 
     that you want to register */
  (void) OCIAttrSet((void *)*dn, (ub4) OCI_DTYPE_SRVDN, 
                    (void *)"cn=server1,cn=oraclecontext,cn=acme,cn=com",
                    42, (ub4)OCI_ATTR_SERVER_DN, errhp);
  /* add the DN of another database in the descriptor */
  (void) OCIAttrSet((void *)*dn, (ub4) OCI_DTYPE_SRVDN, 
                    (void *)"cn=server2,cn=oraclecontext,cn=acme,cn=com",
                    42, (ub4)OCI_ATTR_SERVER_DN, errhp);

  /* set the server DN descriptor into the subscription handle */
  (void) OCIAttrSet((void *) *subhp, (ub4) OCI_HTYPE_SUBSCRIPTION,
                 (void *) *dn, (ub4)0, (ub4) OCI_ATTR_SERVER_DNS, errhp);

  ...

  /* now you will try to get the server DN information from the subscription
     handle */
 
  /* first, get the server DN descriptor out */
  (void) OCIAttrGet((void *) *subhp, (ub4) OCI_HTYPE_SUBSCRIPTION, 
                    (void *)dn, &szp, OCI_ATTR_SERVER_DNS, errhp);

  /* then, get the number of server DNs in the descriptor */
  (void) OCIAttrGet((void *) *dn, (ub4)OCI_DTYPE_SRVDN, (void *)&intval,
                    &szp, (ub4)OCI_ATTR_DN_COUNT, errhp);

  /* allocate an array of char * to hold server DN pointers returned by
     an Oracle database*/
    if (intval)
    {
      arr = (char **)malloc(intval*sizeof(char *));
      (void) OCIAttrGet((void *)*dn, (ub4)OCI_DTYPE_SRVDN, (void *)arr,
                        &intval, (ub4)OCI_ATTR_SERVER_DN, errhp);
    }

  /* OCISubscriptionRegister() calls have two modes: OCI_DEFAULT and 
     OCI_REG_LDAPONLY. If OCI_DEFAULT is used, there should be only one
     server DN in the server DN descriptor. The registration request will
     be sent to the database. If a database connection is not available,
     the registration request will be detoured to the LDAP server. However,
     if mode OCI_REG_LDAPONLY is used, the registration request
     will be directly sent to LDAP. This mode should be used when there is 
     more than one server DN in the server DN descriptor or you are sure
     that a database connection is not available.

     In this example, two DNs are entered, so you should use mode 
     OCI_REG_LDAPONLY in LDAP registration. */
  OCISubscriptionRegister(svchp, subhp, 1, errhp, OCI_REG_LDAPONLY);

  ...

  /* as OCISubscriptionRegister(), OCISubscriptionUnregister() also has
     mode OCI_DEFAULT and OCI_REG_LDAPONLY. The usage is the same. */

  OCISubscriptionUnRegister(svchp, *subhp, errhp, OCI_REG_LDAPONLY);
}
...