20 Working with Oracle Messaging Gateway

Database Connection

The Messaging Gateway agent runs as a process external to the database. To access Oracle Streams AQ and the Messaging Gateway packages, the Messaging Gateway agent needs to establish connections to the database. You can use DBMS_MGWADM.DB_CONNECT_INFO to set the username, password and the database connect string that the Messaging Gateway agent will use for creating database connections. The user must be granted the role MGW_AGENT_ROLE before the Messaging Gateway agent can be started.

You can call DBMS_MGWADM.DB_CONNECT_INFO to alter connection information when the Messaging Gateway agent is running.

Example 20-1 shows Messaging Gateway being configured for user mgwagent with password mgwagent_password using net service name mydatabase.

SQL> exec dbms_mgwadm.db_connect_info(username => 'mgwagent', 
                                      password => 'mgwagent_password',
                                      database => 'mydatabase');

Resource Limits

You can use DBMS_MGWADM.ALTER_AGENT to set the maximum number of messaging connections used by the Messaging Gateway agent, the heap size of the Messaging Gateway agent process, and the number of propagation threads in the agent process. The default values are one connection, 64 MB of memory heap, and one propagation thread.

Example 20-2 sets the number of database connections to two, the heap size to 64MB, and the number of propagation threads to two.

SQL> exec dbms_mgwadm.alter_agent(max_connections => 2,
                                  max_memory => 64,
                                  max_threads => 2);

You can alter the maximum number of connections when the Messaging Gateway agent is running. The memory heap size and the number of propagation threads cannot be altered when the Messaging Gateway agent is running. Example 20-3 updates the maximum number of connections to three but leaves the maximum memory and the number of propagation threads unchanged.

SQL> exec dbms_mgwadm.alter_agent(max_connection => 3);

Starting the Oracle Messaging Gateway Agent

After the Messaging Gateway agent is configured, you can start the agent with DBMS_MGWADM.STARTUP, as shown in Example 20-4.

SQL> exec dbms_mgwadm.startup;

You can use the MGW_GATEWAY view to check the status of the Messaging Gateway agent, as described in Chapter 22, " Monitoring Oracle Messaging Gateway".

Shutting Down the Oracle Messaging Gateway Agent

You can use DBMS_MGWADM.SHUTDOWN to shut down the Messaging Gateway agent, as shown in Example 20-5.

SQL> exec dbms_mgwadm.shutdown;

You can use the MGW_GATEWAY view to check if the Messaging Gateway agent has shut down successfully, as described in Chapter 22, " Monitoring Oracle Messaging Gateway".

Oracle Messaging Gateway Agent Job Queue Job

Messaging Gateway uses a job queue job to start the Messaging Gateway agent. This job is created when procedure DBMS_MGWADM.STARTUP is called. When the job is run, it calls an external procedure that creates the Messaging Gateway agent in an external process. The job is removed after:

• The agent shuts down because DBMS_MGWADM.SHUTDOWN was called

• The agent terminates because a non-restartable error occurs

The DBMS_JOB package creates a repeatable job with a repeat interval of two minutes. The job is owned by SYS. A repeatable job enables the Messaging Gateway agent to restart automatically when a given job instance ends because of a database shutdown, database malfunction, or a restartable error. Only one instance of an Messaging Gateway agent job runs at a given time.

If the agent job encounters an error, then the error is classified as either a restartable error or non-restartable error. A restartable error indicates a problem that might go away if the agent job were to be restarted. A non-restartable error indicates a problem that is likely to persist and be encountered again if the agent job restarts. ORA-01089 (immediate shutdown in progress) and ORA-28576 (lost RPC connection to external procedure) are examples of restartable errors. ORA-06520 (error loading external library) is an example of a non-restartable error.

Messaging Gateway uses a database shutdown trigger. If the Messaging Gateway agent is running on the instance being shut down, then the trigger notifies the agent of the shutdown, and upon receipt of the notification, the agent will terminate the current run. The job scheduler will automatically schedule the job to run again at a future time.

If an Messaging Gateway agent job instance ends because of a database malfunction or a restartable error detected by the agent job, then the job will not be removed and the job scheduler will automatically schedule the job to run again at a future time.

The MGW_GATEWAY view shows the agent status, the job identifier, and the database instance on which the Messaging Gateway agent is currently running.

Running the Oracle Messaging Gateway Agent on RAC

While the Messaging Gateway job startup and shutdown principles are the same for RAC and non-RAC environments, there are some things to keep in mind for a RAC environment.

Only one Messaging Gateway agent process can be running at a given time, even in a RAC environment. The job scheduler determines which database instance will run a job based on parameters specified when the job is created. The DBMS_MGWADM.STARTUP procedure has two optional parameters, instance and force, that can be used to set the instance affinity of the Messaging Gateway agent job.

When a database instance is shut down in a RAC environment, the Messaging Gateway shutdown trigger will notify the agent to shut down only if the Messaging Gateway agent is running on the instance being shut down. The job scheduler will automatically schedule the job to be run again at a future time, either on another instance, or if the job can only run on the instance being shut down, when that instance is restarted.

Creating a WebSphere MQ Base Java Link

A WebSphere MQ Base Java link is created by calling DBMS_MGWADM.CREATE_MSGSYSTEM_LINK with the following information provided:

• Interface type: DBMS_MGWADM.MQSERIES_BASE_JAVA_INTERFACE

• WebSphere MQ connection information:

  • Host name and port number of the WebSphere MQ server

  • Queue manager name

  • Channel name

  • User name and password

• Maximum number of messaging connections allowed

• Log queue names for inbound and outbound propagation

• Optional information such as:

  • Send, receive, and security exits

  • Character sets

Example 20-6 configures a WebSphere MQ Base Java link 'mqlink'. The link is configured to use the WebSphere MQ queue manager 'my.queue.manager' on host 'myhost.mydomain' and port 1414, using WebSphere MQ channel 'mychannel'.

This example also sets the option to register a WebSphere MQ SendExit class. The class 'mySendExit' must be in the CLASSPATH set in mgw.ora.

declare
  v_options sys.mgw_properties;
  v_prop sys.mgw_mqseries_properties;
begin
  v_prop := sys.mgw_mqseries_properties.construct();
  
  v_prop.interface_type := dbms_mgwadm.MQSERIES_BASE_JAVA_INTERFACE;
  v_prop.max_connections := 1;
  v_prop.username := 'mqm';
  v_prop.password := 'mqm';
  v_prop.hostname := 'myhost.mydomain';
  v_prop.port     := 1414;
  v_prop.channel  := 'mychannel';
  v_prop.queue_manager := 'my.queue.manager';
  v_prop.outbound_log_queue := 'mylogq';
  
  -- Specify a WebSphere MQ send exit class 'mySendExit' to be associated with
  -- the queue. 
  -- Note that this is used as an example of how to use the options parameter, 
  -- but is not an option that is usually set.
  v_options := sys.mgw_properties(sys.mgw_property('MQ_SendExit',
                                                   'mySendExit'));
  dbms_mgwadm.create_msgsystem_link(
      linkname => 'mqlink', properties => v_prop, options => v_options );
end;

Creating a WebSphere MQ JMS Link

A WebSphere MQ JMS link is created by calling DBMS_MGWADM.CREATE_MSGSYSTEM_LINK with the following information provided:

• Interface type

  Java Message Service (JMS) distinguishes between queue and topic connections. Each type of connection can be used only for operations involving that JMS destination type:

  • A WebSphere MQ JMS queue link must be created with interface type DBMS_MGWADM.JMS_QUEUE_CONNECTION to access WebSphere MQ JMS queues

  • A WebSphere MQ JMS topic link must be created with interface type DBMS_MGWADM.JMS_TOPIC_CONNECTION to access WebSphere MQ JMS topics

• WebSphere MQ connection information:

  • Host name and port number of the WebSphere MQ server

  • Queue manager name

  • Channel name

  • User name and password

• Maximum number of messaging connections allowed

  A messaging connection is mapped to a JMS session.

• Log destination (JMS queue or JMS topic) for inbound and outbound propagation

  The log destination type must match the link type:

  • For a WebSphere MQ JMS queue link, the log queue name must be the name of a physical WebSphere MQ JMS queue created using WebSphere MQ administration tools.

  • For a WebSphere MQ JMS topic link, the log queue name must be the name of a WebSphere MQ JMS topic. The physical WebSphere MQ queue used by that topic must be created using WebSphere MQ administration tools. By default, the physical queue used is SYSTEM.JMS.D.SUBSCRIBER.QUEUE. A link option can be used to specify a different physical queue.

• Optional information such as:

  • Send, receive, and security exits

  • Character sets

  • WebSphere MQ publish/subscribe configuration used for JMS topics

Example 20-7 configures an Messaging Gateway link to a WebSphere MQ queue manager using a JMS topic interface. The link is named 'mqjmslink' and is configured to use the WebSphere MQ queue manager 'my.queue.manager' on host 'myhost.mydomain' and port 1414, using WebSphere MQ channel 'mychannel'.

This example also uses the options parameter to specify a nondefault durable subscriber queue to be used with the log topic.

declare
  v_options sys.mgw_properties;
  v_prop sys.mgw_mqseries_properties;
begin
  v_prop := sys.mgw_mqseries_properties.construct();
  v_prop.max_connections := 1;
  
  v_prop.interface_type := dbms_mgwadm.JMS_TOPIC_CONNECTION;
  v_prop.username := 'mqm';
  v_prop.password := 'mqm';
  v_prop.hostname := 'myhost.mydomain';
  v_prop.port     := 1414;
  v_prop.channel  := 'mychannel';
  v_prop.queue_manager := 'my.queue.manager'; 
  
  v_prop.outbound_log_queue := 'mylogtopic' 
  
  -- Specify a WebSphere MQ durable subscriber queue to be used with the
  -- log topic.
  v_options := sys.mgw_properties(
  sys.mgw_property('MQ_JMSDurSubQueue', 'myDSQueue'));
  
  dbms_mgwadm.create_msgsystem_link(
      linkname => 'mqjmslink', properties => v_prop, options => v_options );
end;

Creating a TIB/Rendezvous Link

A TIB/Rendezvous link is created by calling DBMS_MGWADM.CREATE_MSGSYSTEM_LINK with three parameters (service, network and daemon) for the agent to create a corresponding transport of TibrvRvdTransport type.

A TIB/Rendezvous message system link does not need propagation log queues. Logging information is stored in memory. Therefore, Messaging Gateway can only guarantee at-most-once message delivery.

Example 20-8 configures a TIB/Rendezvous link named 'rvlink' that connects to the rvd daemon on the local computer.

declare
  v_options sys.mgw_properties;
  v_prop     sys.mgw_tibrv_properties;
begin
  v_prop := sys.mgw_tibrv_properties.construct();
  
  dbms_mgwadm.create_msgsystem_link(linkname => 'rvlink', properties => v_prop);
end;

Altering a Messaging System Link

Some link information can be altered after the link is created. You can alter link information with the Messaging Gateway agent running or shut down. Example 20-9 alters the link 'mqlink' to change the max_connections and password properties.

declare
  v_options sys.mgw_properties;
  v_prop sys.mgw_mqseries_properties;
begin
  -- use alter_construct() for initialization
  v_prop := sys.mgw_mqseries_properties.alter_construct();
  v_prop.max_connections := 2;
  v_prop.password := 'newpasswd';
  
  dbms_mgwadm.alter_msgsystem_link(
    linkname => 'mqlink', properties => v_prop);
end;

Removing a Messaging System Link

You can remove an Messaging Gateway link to a non-Oracle messaging system with DBMS_MGWADM.REMOVE_MSGSYSTEM_LINK, but only if all registered queues associated with this link have already been unregistered. The link can be removed with the Messaging Gateway agent running or shut down. Example 20-10 removes the link 'mqlink'.

begin
  dbms_mgwadm.remove_msgsystem_link(linkname =>'mqlink');
end;

Views for Messaging System Links

You can use the MGW_LINKS view to check links that have been created. It lists the name and link type, as shown in Example 20-11.

SQL> select link_name, link_type from MGW_LINKS;
 
LINK_NAME     LINK_TYPE 
------------------------
MQLINK        MQSERIES
RVLINK        TIBRV

You can use the MGW_MQSERIES_LINK and MGW_TIBRV_LINKS views to check messaging system type-specific configuration information, as shown in Example 20-12.

SQL> select link_name, queue_manager, channel, hostname from mgw_mqseries_link;
 
LINK_NAME   QUEUE_MANAGER      CHANNEL     HOSTNAME
----------------------------------------------------------
MQLINK      my.queue.manager  mychannel  myhost.mydomain
 
SQL> select link_name, service, network, daemon from mgw_tibrv_links;
 
LINK_NAME   SERVICE      NETWORK     DAEMON
-----------------------------------------------------
RVLINK 

Registering a Non-Oracle Queue

Registering a non-Oracle queue provides information for the Messaging Gateway agent to access the queue. However, it does not create the physical queue in the non-Oracle messaging system. The physical queue must be created using the non-Oracle messaging system administration interfaces before the Messaging Gateway agent accesses the queue.

The following information is used to register a non-Oracle queue:

• Name of the messaging system link used to access the queue

• Native name of the queue (its name in the non-Oracle messaging system)

• Domain of the queue

  • DBMS_MGWADM.DOMAIN_QUEUE for a point-to-point queue

  • DBMS_MGWADM.DOMAIN_TOPIC for a publish/subscribe queue

• Options specific to the non-Oracle messaging system

  These options are a set of {name, value} pairs, both of which are strings.

Example 20-13 shows how to register the WebSphere MQ Base Java queue my_mq_queue as an Messaging Gateway queue 'destq'.

begin
  dbms_mgwadm.register_foreign_queue(
    name => 'destq',
    linkname => 'mqlink',
    provider_queue => 'my_mq_queue',
    domain => dbms_mgwadm.DOMAIN_QUEUE);
end;

Unregistering a Non-Oracle Queue

A non-Oracle queue can be unregistered with DBMS_MGWADM.UNREGISTER_FOREIGN_QUEUE, but only if there are no subscribers or schedules referencing it.

Example 20-14 unregisters the queue 'destq' of the link 'mqlink'.

begin
  dbms_mwgadm.unregister_foreign_queue(name =>'destq', link_name=>'mqlink');
end;

View for Registered Non-Oracle Queues

You can use the MGW_FOREIGN_QUEUES view to check which non-Oracle queues are registered and what link each uses, as shown in Example 20-15.

SQL> select name, link_name, provider_queue from MGW_FOREIGN_QUEUES;
 
NAME   LINK_NAME   PROVIDER_QUEUE
------------------------------------
DESTQ  MQLINK      my_mq_queue

Propagation Subscriber Overview

A propagation subscriber specifies what messages are propagated and how the messages are propagated.

Messaging Gateway allows bi-directional message propagation. An outbound propagation moves messages from Oracle Streams AQ to non-Oracle messaging systems. An inbound propagation moves messages from non-Oracle messaging systems to Oracle Streams AQ.

If the propagation source is a queue (point-to-point), then the Messaging Gateway agent moves all messages from the source queue to the destination queue. If the propagation source is a topic (publish/subscribe), then the Messaging Gateway agent creates a subscriber of the propagation source queue in the messaging system. The agent only moves messages that are published to the source queue after the subscriber is created.

When propagating a message, the Messaging Gateway agent converts the message from the format in the source messaging system to the format in the destination messaging system. Users can customize the message conversion by providing a message transformation. If message conversion fails, then the message will be moved to an exception queue, if one has been provided, so that the agent can continue to propagate messages for the subscriber.

An Messaging Gateway exception queue is different from an Oracle Streams AQ exception queue. Messaging Gateway moves a message to an Messaging Gateway exception queue when message conversion fails. Oracle Streams AQ moves a message to an Oracle Streams AQ exception queue after MAX_RETRIES dequeue attempts on the message.

Messages moved to an Oracle Streams AQ exception queue may result in unrecoverable failures on the associated Messaging Gateway subscriber. To avoid the problem, the MAX_RETRIES parameter of any Oracle Streams AQ queue that is used as the propagation source of an Messaging Gateway propagation job should be set to a value much larger than 16.

If the messaging system of the propagation source queue supports message selection, then a message selection rule can be specified for a propagation subscriber. Only messages that satisfy the message selector will be propagated.

Users can also specify subscriber options for certain types of propagation subscribers to control how messages are propagated, such as options for JMS message delivery mode and TIB/Rendezvous queue policies.

Messaging Gateway provides MGW_SUBSCRIBERS and MGW_SCHEDULES views for users to check configuration and status of Messaging Gateway subscribers and schedules.

Creating an Oracle Messaging Gateway Propagation Subscriber

Messaging Gateway subscribers are created by DBMS_MGWADM.ADD_SUBSCRIBER.

If the propagation source for non-JMS propagation is an Oracle Streams AQ queue, then the queue must be a multiconsumer queue. Messaging Gateway creates a corresponding Oracle Streams AQ subscriber 'MGW_subscriber_id' for the messaging system subscriber subscriber_id when DBMS_MGWADM.ADD_SUBSCRIBER is called.

If the propagation source is a JMS topic, such as an Oracle Java Message Service (OJMS) topic or a WebSphere MQ JMS topic, then a JMS subscriber 'MGW_subscriber_id' is created on the topic in the source messaging system by the Messaging Gateway agent. If the agent is not running, then the subscriber will not be created until the agent is restarted.

If the propagation source is a queue, then only one propagation job can be created using that queue as the propagation source. If the propagation source is a topic, then multiple propagation jobs can be set up using that topic as the propagation source with each propagation job having its own corresponding subscriber on the topic in the messaging system.

Example 20-16 creates Messaging Gateway propagation subscriber sub_aq2mq.

begin
  dbms_mgwadm.add_subscriber(
    subscriber_id => 'sub_aq2mq',
    propagation_type => dbms_mgwadm.outbound_propagation,
    queue_name => 'mgwuser.srcq', 
    destination => 'destq@mqlink');
end;

Creating an Oracle Messaging Gateway Propagation Schedule

A propagation subscriber is not processed until an associated propagation schedule is created and enabled. A propagation schedule is associated with a propagation subscriber when the propagation type, source and destination match.

The latency parameter in a propagation schedule controls the polling interval of a propagation job. The polling interval determines how soon the agent can discover the available messages to propagate in the propagation source queue. The default polling interval is 5 seconds or the value set for oracle.mgw.polling_interval in Messaging Gateway initialization file mgw.ora.

Example 20-17 creates Messaging Gateway propagation schedule sch_aq2mq.

begin
  dbms_mgwadm.schedule_propagation(
    schedule_id => 'sch_aq2mq',
    propagation_type => dbms_mgwadm.outbound_propagation,
    source => 'mgwuser.srcq',
    destination => 'destq@mqlink',
    latency => 2);
end;

Enabling and Disabling a Propagation Job

A propagation job is enabled if its propagation schedule is created and enabled. A propagation job is disabled if its propagation schedule is disabled or removed. Users can call DBMS_MGWADM.ENABLE_PROPAGATION_SCHEDULE to enable a propagation schedule and DBMS_MGWADM.DISABLE_PROPAGATION_SCHEDULE to disable a propagation schedule.

Example 20-18 enables the propagation schedule for propagation subscriber sub_aq2mq.

begin
  dbms_mgwadm.enable_propagation_schedule('sch_aq2mq');
end;

Example 20-19 disables the propagation schedule for propagation subscriber sub_aq2mq.

begin
  dbms_mgwadm.disable_propagation_schedule('sch_aq2mq');
end;

By default, the propagation schedule is enabled when it is first created.

To create a propagation job that is initially disabled, call the following APIs in the indicated order:

1. DBMS_MGWADM.SCHEDULE_PROPAGATION

2. DBMS_MGWADM.DISABLE_PROPAGATION_SCHEDULE

3. DBMS_MGWADM.ADD_SUBSCRIBER

Resetting a Propagation Job

When a problem occurs with a propagation job, the Messaging Gateway agent retries the failed operation up to 16 times in an exponential backoff scheme before the propagation job stops. You can use DBMS_MGWADM.RESET_SUBSCRIBER to reset the failure count to zero to allow the agent to retry the failed operation immediately.

Example 20-20 resets the failure count for propagation subscriber sub_aq2mq.

begin
  dbms_mgwadm.reset_subscriber('sub_aq2mq');
end;

Altering a Propagation Subscriber and Schedule

After the propagation subscriber and schedule of a propagation job are created, you can alter the selection rule, transformation, exception queue, subscriber options, and latency of the propagation job. Subscribers and schedules can be altered with the Messaging Gateway agent running or shut down.

Example 20-21 adds an exception queue for subscriber sub_aq2mq.

begin
  dbms_mgwadm.alter_subscriber(
    subscriber_id => 'sub_aq2mq',
    exception_queue => 'mgwuser.my_ex_queue');
end;

Example 20-22 changes the polling interval for schedule sch_aq2mq.

begin
  dbms_mgwadm.alter_propagation_schedule(
    subscriber_id => 'sch_aq2mq',
    latency => 1);
end;

Removing a Propagation Subscriber and Schedule

You can remove an Messaging Gateway propagation subscriber with DBMS_MGWADM.REMOVE_SUBSCRIBER.

Before removing the Messaging Gateway subscriber from the Messaging Gateway configuration, Messaging Gateway does the following cleanup:

• Removes from the messaging system the associated subscriber that may have been created by Messaging Gateway

• Removes propagation log records from log queues for the subscriber being removed

Messaging Gateway may fail to do the cleanup because:

• The Messaging Gateway agent is not running

• Non-Oracle messaging system is not running

• The Messaging Gateway agent is unable to interact with the source or destination messaging system

If Messaging Gateway cleanup fails for any reason, then the Messaging Gateway subscriber being removed is placed in the DELETE_PENDING state. The Messaging Gateway agent tries to clean up subscribers in DELETE_PENDING state when:

• DBMS_MGWADM.REMOVE_SUBSCRIBER is called and the Messaging Gateway agent is running

• The Messaging Gateway agent is starting and finds a subscriber in DELETE_PENDING state

You can specify DBMS_MGWADM.FORCE when calling DBMS_MGWADM.REMOVE_SUBSCRIBER to force Messaging Gateway to remove the Messaging Gateway subscriber from the Messaging Gateway configuration without placing it in the DELETE_PENDING mode in case of cleanup failures.

Calling DBMS_MGWADM.REMOVE_SUBSCRIBER with DBMS_MGWADM.FORCE may result in obsolete log records in the log queues and subscriptions in messaging systems, which may cause unnecessary message accumulation. Oracle recommends not using DBMS_MGWADM.FORCE when calling DBMS_MGWADM.REMOVE_SUBSCRIBER, if possible.

Example 20-23 removes propagation subscriber sub_aq2mq.

begin
  dbms_mgwadm.remove_subscriber(subscriber_id =>'sub_aq2mq');
end;

You can remove propagation schedules with DBMS_MGWADM.UNSCHEDULE_PROPAGATION. Removing a propagation schedule results in disabling the associated propagation job. It does not remove any subscriptions in messaging systems.

Example 20-24 removes propagation schedule sch_aq2mq.

begin
  dbms_mgwadm.unschedule_propagation(schedule_id => 'sch_aq2mq');
end;