4.1.2.1. UMP Flight Size

UMP supports a flight size mechanism that tracks messages in flight from a particular source and responds when a send would exceed the configured flight size ( ume_flight_size and/or ume_flight_size_bytes). You can configure ume_flight_size_behavior to either:

• block any sends that would exceed the flight size or,

• allow the sends while notifying your application.

UMP considers a sent message in flight until the following two conditions are met.

1. The source receives the configured number of stability acknowledgements from the store(s).

2. The source has received the configured number of delivery confirmation notifications. (See ume_retention_unique_confirmations.)

If configuring both ume_flight_size and ume_flight_size_behavior, UMP uses the smaller of the two flight sizes on a per send basis.

When using stores in a Quorum/Consensus configuration, intragroup and intergroup stability settings affect whether UMP considers a messages in flight. Consider a case with three stores in a single QC group, and two receivers. Given the default configuration, until a source receives a stability notification from two of the three stores, UMP considers a given message in-flight. In addition, if you set ume_retention_unique_confirmations to 2, that same message would be considered in flight until the source receives two stability notifications AND two delivery confirmation notifications. See also Sources Using Quorum/Consensus Store Configuration.

4.2.1.1. Late Registering Receiver

Late joining receivers that register after the first RPP topic message has been sent cannot receive any messages sent prior to their registration, except for messages not yet consumed by all RPP receivers. This behavior also applies to the very first receiver of a RPP group that registers after the source sends the first message. Any messages published prior to RPP receiver registration are not available for recovery.

4.2.1.2. Early Exiting Receiver

Should a registered receiver's activity timer expire and be declared by the repository to be inactive, the repository retains all messages published since the receiver's last acknowledged message (or initial sequence number if no messages were acknowledged) until its receiver state lifetime expires and the repository deletes the receiver state information. Deleting receiver state removes all knowledge of the receiver from the repository. As a result, the repository also deletes all messages being held solely for this receiver.

Should an early exiting receiver reregister (or otherwise become active) before the expiration of its state lifetime, that receiver can recover all messages retained for that receiver.

4.2.1.3. UMP Version RPP Compatibility Matrix

The following table indicates the result of registration requests across UMP versions.

• Granted - Source Error indicates that the store granted the registration but the source detected that RPP behavior was not acknowledged by the store.

• Granted - Receiver Error indicates that the store granted the registration but the receiver detected that RPP behavior was not acknowledged by the store.

• * Refers only to the re-registration of a source with an existing source repository because the source determines the repository's behavior for new registrations.

4.2.6.1. UM Configuration File

The following example UM configuration file contains RPP options in the ##Persistence Options### section.

• The source uses the same repository size values as the store. In this case, you do not need to specify these option values again in the source's UM Configuration File. They appear in this file for the sake of completeness.

• The source reconfigures ume_flight_size_bytes to 1,000,000 bytes, which is less than the repository's 4 MB default. (The source can reconfigure this option to a value less than or equal to the repository's configured value.)

• The source reconfigures ume_write_delay from the default of 0 ms to 1000 ms or 1 second.

• The option, ume_session_id 5353, is commented out because this file specifies RegIDs 2929 and 2930, respectively, for the stores in the ume_store_name option. The option, ume_session_id, appears in this file as a reminder that you can use either RegIDs or Session IDs, but not both.

#Sample UM Configuration File, UMP Version 5.3
#Major Options
source transport lbtrm
# in order and reassembled
receiver ordered_delivery 1
#Multicast Resolver Network Options
context resolver_multicast_address 225.8.17.29
context resolver_multicast_interface 10.29.3.0/24
# Transport LBT-RM Netowrk Options
source transport_lbtrm_multicast_address 225.8.17.30
context transport_lbtrm_multicast_address_low 225.12.17.10
context transport_lbtrm_multicast_address_high 225.12.17.14
#Transport LBT-RM Operation Options
context transport_lbtrm_data_rate_limit 10000000
context transport_lbtrm_retransmit_rate_limit 5000000
# Transport LBT-RM Reliability Options
receiver transport_lbtrm_nak_initial_backoff_interval 40000
receiver transport_lbtrm_nak_initial_backoff_interval 500
receiver transport_lbtrm_nak_generation_interval 10000
##Turn off NAKs
receiver transport_lbtrm_send_naks 0
#Request Network Options
context request_tcp_port_low 55000
context request_tcp_port_high 55500

## Persistence Options ###
source ume_store_group 0:1
source ume_store_name store0rpp:2929:0
source ume_store_group 1:1
source ume_store_name store1rpp:2930:1
source ume_store_behavior qc
source ume_flight_size 500
source ume_flight_size_bytes 1000000
source ume_receiver-paced-persistence 1
source ume_repository_size_threshold 104857600
source ume_repository_size_limit 209715200
source ume_repository_disk_file_size_limit 1073741824
source ume_repository_ack_on_reception 1
source ume_write_delay 1000
receiver ume_receiver-paced-persistence 1
receiver ume_explicit_ack_only 1
source ume_proxy_source 1
#source ume_session_id 535353
context ume_source_liveness_timeout 4000
context ume_receiver_liveness_interval 1000
source ume_confirmed_delivery_notification 1
                   

4.2.6.2. umestored Configuration File

The following example store configuration file contains RPP options in the ABC* topic section.

• The store has raised the repository-size-limit from the default of 48 MB to 200 MB, the repository-size-threshold from the default of 0 to 100 MB, and the repository-disk-file-size-limit from the default of 100MB to 1 GB.

• The store does not specify a source-flight-size-bytes-maximum, using the default of 4 MB.

<?xml version="1.0"?>
<ume-store version="1.2">
  <daemon>
    <log>/configs/stores/umestored1/umestored.log</log>
    <lbm-license-file>/bin/umq_exp_license.txt</lbm-license-file> 
    <lbm-license-file>/bin/lbm_ume_umq_udx_rdma_license.txt</lbm-license-file> 
    <lbm-config>/configs/lbm_4_store.cfg</lbm-config>
    <pidfile>/configs/stores/umestored1/umestored.pid</pidfile>
    <web-monitor>*:15404</web-monitor>
  </daemon>
  <stores>
    <store name="rpp-ump-test-store-1" port="14667">
      <ume-attributes>
        <option type="store" name="disk-cache-directory" value="/stores/store1/cache"/>
        <option type="store" name="disk-state-directory" value="/stores/store1/state"/>  
        <option type="store" name="allow-proxy-source" value="0" />
        <option type="store" name="context-name" value="store1rpp"/>
      </ume-attributes>
      <topics>
        <topic pattern="ABC*" type="PCRE">
          <ume-attributes>
             <option type="store" name="repository-allow-receiver-paced-persistence" value="1"/>
             <option type="store" name="repository-type" value="disk"/>
             <option type="store" name="repository-size-threshold" value="104857600"/>
             <option type="store" name="repository-size-limit" value="209715200"/>
             <option type="store" name="repository-disk-file-size-limit" value="1073741824"/>
             <option type="store" name="repository-allow-ack-on-reception" value="1"/>
             <option type="store" name="repository-disk-write-delay" value="1000"/> 
             <option type="store" name="receiver-new-registration-rollback" value="0"/>
             <option type="store" name="source-activity-timeout" value="120000"/>
             <option type="store" name="receiver-activity-timeout" value="30000"/>
             <option type="store" name="retransmission-request-forwarding" value="0"/>
          </ume-attributes>
        </topic>
      </topics>
    </store>
  </stores>
</ume-store>
                   

4.3.1.1. Source Application Registration

At a high level, the following registration activity occurs before a source application submits messages to a queue topic.

1. umq_queue_name must be set for queue Q1.

2. umestored must be started and configured for queue Q1 and topic SubjectA.

3. The source application creates a source object for topic SubjectA.

4. The source application's context automatically registers with Q1 if not already registered.

5. The Queue sends a Queue Registration Complete context event to the source application.

6. The source object automatically registers as SubjectA with Q1.

7. The Queue sends a Queue Registration Complete source event to the source application.

4.3.1.2. Receiver Application Registration

At a high level, the following registration activity occurs before a receiver application can receive queue topic messages.

1. umestored must be started and configured for queue Q1 and topic SubjectA.

2. The receiver application creates a receiver object for topic SubjectA.

3. The receiver discovers through topic resolution that Q1 has messages for topic SubjectA. (Topic advertisements in a queuing operation contain the queue name.)

4. The receiver application's context automatically registers with Q1 if not already registered.

5. The Queue sends a Queue Context Registration Complete message to the receiver application after the receiver application's context registers.

6. The receiver object automatically registers as SubjectA with Q1 If not using UMQ Sessions IDs, the receiver includes its Assignment ID with the registration request.

7. The Queue sends a Queue Receiver Registration Complete message to the receiver application. If using UMQ Sessions IDs, the Queue includes the receiver's Assignment ID with the registration complete message.

4.3.2.1. Serial Queue Dissemination (SQD)

The Serial Queue Dissemination (SQD) model (Serial Queue Dissemination (SQD)) uses direct serial unicast from the queue to the individual receivers. Receivers only receive the messages they are assigned to process. The term serial indicates that the queue sends each message via unicast only to the message's assigned receivers (one in each application set). This dissemination model creates less work for receivers than either PQD or SD, which require receivers to decipher control information to determine the messages they must consume.

4.3.2.2. Parallel Queue Dissemination (PQD)

The Parallel Queue Dissemination (PQD) model (Parallel Queue Dissemination (PQD)) uses normal UM transport sessions to disseminate messages. In fact, the queue uses individual UM topics to send messages to all receivers. In addition, the Queue sends control information (Receiver Control Record - RCR) over a specific topic, configured with the control-topic-name Queue option in the Queue's XML configuration file. (See Queue Element. Receivers listen to all data and control information and deliver all their assigned messages to the application, ignoring all other messages.

4.3.2.3. Source Dissemination (SD)

The Source Dissemination (SD) model also uses normal UM transport sessions from source application to send message data to the receivers. The queue sends control information (Receiver Control Record - RCR) as in the PQD model that instructs receivers via assignments what to process and what to ignore. However, the queue does not send data messages on topics.

4.3.7.1. Queue Browser Authentication

UM queues can authenticate UM applications using the Ultra Messaging JMS Queue Browser feature. UM applications can also authenticate the queue.

The use of any of the queue browser APIs mentioned in Queue Browser, initiates authentication automatically between your application and the queue. You can require authentication between your application and the queue, ensuring that authentication must be successful before queue browsing can occur.

You require authentication by setting the UM Configuration option, umq_require_queue_authentication, to the default setting of 1 (authentication required). In addition, set the queue (umestored) configuration option, require-client-authentication to the default value of 1.

4.3.7.2. Setting Queue Browser Authentication Credentials

Perform the following two tasks to set the Queue Browser authentication credentials.

1. Use the C API lbm_auth_set_credentials() or the Java API LBMAuthUserInfo() in your application to create users and passwords in your application. A Credential callback provides a way for you to supply alternate credentials in the event of authentication failure.

2. Generate a password.xml file that contains the usernames and passwords used by your application. Place password.xml in the directory configured in the umestored XML configuration file with the option, lbm-password-file. See Daemon Element. The queue accesses this file during authentication to verify usernames and passwords. You can generate password.xml in one of the two following ways.

   • Use the lbm_authstorage_*() API calls in an auxiliary application to create password.xml. This application can import any existing user credentials (i.e. from LDAP). (No Java equivalent exists for these functions.)

   • Use the UMQ utility, /bin/lbmpwdgen to generate password.xml. Usage information appears within the file. (This utility uses lbm_authstorage_*() API calls.)

A sample password.xml appears below. Notice that you can also create and assign user roles. In the sample you may also notice that /bin/lbmpwdgen creates an anonymous user (<user name="">). UMQ requires this user when authentication has not been enabled. You should not delete or edit this user.

<?xml version="1.0"?>
<um-configuration version="1.0">
  <users>
    <user name="userSmith">
      <verifier>69A4aAE70US3NiuZr/TvAbSWztu5na5TbFo8bdHxU5.ILdMu8rLd5ragE3p4Qcuz/nXxAj6kGnwIF2JrKdCf704U/Lxs7jvK.b2uyOQoqKG</verifier>
        <salt>3BmRWUznvzy0n2</salt>
        <roles>
          <role name="admin"></role>
          <role name="normal"></role>
        </roles>
    </user>
    <user name="">
      <verifier/>
      <salt/>
      <roles>
        <role name="admin"/>
      </roles>
    </user>
  </users>
  <roles>
    <role name="admin">
      <action>MSG_LIST</action>
      <action>MSG_RETRIEVE</action>
      <action>TOPIC_LIST</action>
     </role>
    <role name="normal">
      <action>TOPIC_LIST</action>
    </role>
  </roles>
</um-configuration>
           

7.1.1.1. Use Static RegIDs

The simplest method uses static RegIDs for individual applications. This method works best if:

• Applications use separate stores

• Multiple instances of an application also use separate stores

In the latter case, the same static source RegID can be used in every instance of the application because receivers will identify every Store/Source RegID tuple as unique.

The following source code examples assign a static RegID to a source by adding the RegID, 1000, to the ume_store attribute. (See also ume-example-src-2.c.)

C API

lbm_src_topic_attr_t * sattr;

if (lbm_src_topic_attr_create(&sattr) == LBM_FAILURE) {
        fprintf(stderr, "lbm_src_topic_attr_create: %s\n", lbm_errmsg());
        exit(1);
}
if (lbm_src_topic_attr_str_setopt(sattr, "ume_store", "127.0.0.1:14567:1000")
== LBM_FAILURE) {
        fprintf(stderr, "lbm_src_topic_attr_str_setopt: %s\n", lbm_errmsg());
        exit(1);
}
       

JAVA API

LBMSourceAttributes sattr = null;
try {
sattr = new LBMSourceAttributes();
sattr.setValue("ume_store", "127.0.0.1:14567:1000");
}
catch (LBMException ex) {
System.err.println("Error creating source attribute: " + ex.toString());
System.exit(1);
}
       

.NET API

LBMSourceAttributes sattr = null;
try {
sattr = new LBMSourceAttributes();
sattr.setValue("ume_store", "127.0.0.1:14567:1000");
}
catch (LBMException ex) {
System.Console.Error.WriteLine ("Error creating source attribute: " + ex.toString());
System.Environment.Exit(1);
       }
       

7.1.1.2. Save Assigned RegIDs

Your application can save the RegID assigned to a source or receiver from the store because the UMP API informs your application of the RegID used for each registration. This method of managing RegIDs is perhaps the most flexible, but also requires some work by the application to save RegIDs and retrieve them in some way.

The following source code examples save the RegID assigned to a source to a file. (See also ume-example-src-3.c.)

C API

typedef struct src_info_t_stct {
    int existing_regid;       
    int message_num;          
} src_info_t;

#define SRC_REGID_SAVE_FILENAME "UME-example-src-RegID"

int save_src_regid_to_file(const char *filename, lbm_src_event_ume_registration_ex_t *reg)
{
    FILE *fp;           
    
    if ((fp = fopen(filename, "w")) == NULL)
        return -1;
    fprintf(fp, "%s:%u", reg->store, reg->registration_id);
    printf("saving RegID info to \"%s\" - %s:%u\n", filename, reg->store, reg->registration_id);
    fflush(fp);
    fclose(fp);
    return 0;
}
       

7.1.1.3. Managing RegIDs with Session IDs

The RegIDs used by stores to identify sources and receivers must be unique. Rather than maintaining RegIDs (either statically or dynamically), applications can use a Session ID, which is simply a 64-bit value that uniquely identifies any set of sources with unique topics and receivers with unique topics. A single Session ID allows UMP stores to correctly identify all the sources and receivers for a particular application.

Combinations of sources and receivers that make up a single valid session include the following.

• Sources for topics A, B, and C

• Receivers for topics A, B, and C

• Sources for topics A, B, and C, and receivers for topics X, Y and Z

• Sources for topics A, B, and C, and receivers for topics A, B, and C

The UMP configuration option, ume_session_id , specifies a Session ID for a source, receiver or a context. If you want all sources and receivers for a particular context to use the same Session ID, use (context) ume_session_id . Any source or receiver that does not specify its own Session ID inherits the context's session ID. If a source or receiver specifies its own Session ID, it overrides the context Session ID for that individual source or receiver.

Of the two mutually exclusive methods for managing RegIDs, ...

1. Enable your application to assign and manage every RegID, ensuring no two objects registered with an individual store share the same RegID.

2. Allow the store to assign every RegID and enable your application to persist the RegIDs.

... using Session IDs simplifies the second management method. Since you cannot combine these two strategies at any single store, you also cannot combine the first method with the use of Session IDs at a single store.

7.1.2.1. New or Re-Registration

Any source needs to know at start-up if it is a new registration or a re-registration. The answer determines how a source registers with the store. UMP can not answer this question. Therefore, it is essential that the developer consider what identifies the lifetime of a source and how a source determines the appropriate value to use as the RegID when it is ready to register. RegIDs are per source per topic per store, thus a single RegID per store is needed.

The following source code examples look for an existing RegID from a file and uses a new RegID assigned from the store if it finds no existing RegID. (See also ume-example-src-3.c.)

C API

    err = lbm_context_create(&ctx, NULL, NULL, NULL);
    if (err) {printf("line %d: %s\n", __LINE__, lbm_errmsg()); exit(1);}

    srcinfo.message_num = 1;
    srcinfo.existing_regid = 0;
    
    err = read_src_regid_from_file(SRC_REGID_SAVE_FILENAME, store_info, sizeof(store_info));
    if (!err) { srcinfo.existing_regid = 1; }

        err = lbm_src_topic_attr_create(&attr);
        if (err) {printf("line %d: %s\n", __LINE__, lbm_errmsg()); exit(1);}

        err = lbm_src_topic_attr_str_setopt(attr, "ume_store", store_info);
        if (err) {printf("line %d: %s\n", __LINE__, lbm_errmsg()); exit(1);}
       

The use of Session IDs allows UMP , as opposed to your application, to accomplish the same RegID management. See Managing RegIDs with Session IDs.

7.1.2.2. Sources Must Be Able to Resume Sending

A source sends messages unless UMP prevents it, in which case, the send function returns an error. A source may lose the ability to send messages temporarily if the store(s) in use become unresponsive, e.g. the store(s) die or become disconnected from the source. Once the store(s) are responsive again, sending can continue. Thus source applications need to take into account that sending may fail temporarily under specific failure cases and be able to resume sending when the failure is removed.

The following source code examples demonstrate how a failed send function can sleep for a second and try again.

C API

   while (lbm_src_send(src, message, len, 0) == LBM_FAILURE) {
        If (lbm_errnum() == LBM_EUMENOREG) {
            printf("Send unsuccessful. Waiting...\n");
            sleep(1);
            continue;
        }
        fprintf(stderr, "lbm_src_send: %s\n", lbm_errmsg());
                        exit(1);
                }
       

JAVA API

   for (;;) {
        try {
            src.send(message, len, 0);
        }
        catch (UMENoRegException ex) {
            System.out.println("Send unsuccessful. Waiting...");
            try {
                Thread.sleep(1000);
            }
            catch (InterruptedException e) { }
            continue;
        }
        catch (LBMException ex) {
            System.err.println("Error sending message: " + ex.toString());
System.exit(1);
        }
        break;
    }
       

.NET API

   for (;;) {
        try {
            src.send(message, len, 0);
        }
        catch (UMENoRegException ex) {
            System.Console.Out.WriteLine("Send unsuccessful. Waiting...");
            System.Threading.Thread.Sleep(1000);
            continue;
        }
        catch (LBMException ex) {
            System.Console.Out.WriteLine ("Error sending message: " + ex.toString());
System.exit(1);
        }
        break;
    }

       

7.1.2.3. Source Message Retention and Release

UMP allows streaming of messages from a source without regard to message stability at a store, which is one reason for UMP's performance advantage over other persistent messaging systems. Sources retain all messages until notified by the active store(s) that they are stable. This provides a method for stores to be brought up to date when restarted or started anew.

When messages are considered stable at the store, the source can release them which frees up source retention memory for new messages. Generally, the source releases older stable messages first. To release the oldest retained message, all the following conditions must be met:

• message must meet stability requirements of the source, which can range from a single stability notice from the active store to stability notices from a group of stores (See Sources Using Quorum/Consensus Store Configuration)

  and

• message must have been confirmed as delivered by a configured number of receivers (ume_retention_unique_confirmations),

  and

• the aggregate amount of buffered messages exceeds retransmit_retention_size_threshold bytes in payload and headers.

Some things to note:

• If the retransmit_retention_size_threshold is not met, no messages will be released regardless of stability.

• If the source registered with a "no-cache" store (See UMP Stores) or ume_message_stability_notification is turned off, ume_retention_unique_confirmations is the only way to allow the source to release messages before retention size options come into play.

• If the aggregate amount of buffered messages exceeds retransmit_retention_size_limit bytes in payload and headers, then the oldest retained message is forcibly released even if it does not meet one or more of the conditions above. This condition should be avoided and suggests increasing the retransmit_retention_size_limit or lowering the retransmit_retention_size_threshold.

7.1.2.4. Source Release Policy Options

sources use a set of configuration options to release messages that, in effect, specify the source's release policy. The following configuration options directly impact when the source may release retained messages.

• ume_message_stability_notification

• ume_retention_unique_confirmations

• retransmit_retention_size_threshold

• retransmit_retention_size_limit

7.1.2.5. Confirmed Delivery

As mentioned earlier, ume_retention_unique_confirmations requires a message to have a minimum number of unique confirmations from different receivers before the message may be released. This retains messages that have not been confirmed as being received and processed and keeps them available to fulfill any retransmission requests.

The following code samples show how to require a message to have 10 unique receiver confirmations

C API

lbm_src_topic_attr_t * sattr;

if (lbm_src_topic_attr_create(&sattr) == LBM_FAILURE) {
        fprintf(stderr, "lbm_src_topic_attr_create: %s\n", lbm_errmsg());
        exit(1);
}
if (lbm_src_topic_attr_str_setopt(sattr, "ume_retention_unique_confirmations",
"10") 
== LBM_FAILURE) {
        fprintf(stderr, "lbm_src_topic_attr_str_setopt: %s\n", lbm_errmsg());
        exit(1);
}
       

JAVA API

LBMSourceAttributes sattr = null;
try {
sattr = new LBMSourceAttributes();
sattr.setValue("ume_retention_unique_confirmations", "10");
}
catch (LBMException ex) {
System.err.println("Error creating source attribute: " + ex.toString());
System.exit(1);
}
       

.NET API

LBMSourceAttributes sattr = null;
try {
sattr = new LBMSourceAttributes();
sattr.setValue("ume_retention_unique_confirmations", "10");
}
catch (LBMException ex) {
System.Console.Error.WriteLine ("Error creating source attribute: " + ex.toString());
System.Environment.Exit(1);
       }
       

7.1.2.6. Sources Using Round-Robin Store Configuration

The source retains messages until they are considered stable at the active store(s). For Round-Robin store behavior, this means the current active store notifies the source that it has stabilized the message via a message stability notification. The following configuration file statements implement Round-Robin behavior among 3 stores.

source ume_store 10.29.3.77:15313:150000:0
source ume_store 10.29.3.76:16313:160000:0 
source ume_store 10.29.3.75:17313:170000:0
source ume_message_stability_notification 1
source ume_store_behavior rr 
       

See also Round-Robin Store Usage

7.1.2.7. Sources Using Quorum/Consensus Store Configuration

In the case of Quorum/Consensus store behavior, a message is considered stable after it has been successfully stored within a group of stores or among groups of stores according to the two settings, intergroup behavior and intragroup behavior, described below.

• The intragroup behavior specifies the requirements needed to stabilize a message among the stores within a group. A message is stable for the group once it is successfully stored at a quorum (majority) of the group's stores or successfully stored in all the stores in the group.

• The intergroup behavior specifies the requirements needed to stabilize a message among groups of stores. A message is stable among the groups if it is successfully stored at any group, a majority of groups, or all groups.

Notice that a message needs to meet intragroup stability requirements before it can meet intergroup stability requirements. These options provide a number of possibilities for retention of messages for the source.

The following configuration file statements implement a 3-group Quorum/Consensus configuration with each group on a different machine, in which a message is considered stable when it has been successfully stored at a quorum of stores in at least one group. (See Quorum/Consensus - Single Location Groups for more information about this configuration.)

source ume_store 10.29.3.77:10313:101000:0
source ume_store 10.29.3.77:11313:110000:0
source ume_store 10.29.3.77:12313:120000:0
source ume_store 10.29.3.77:13313:130000:0
source ume_store 10.29.3.77:14313:140000:0
source ume_store 10.29.3.78:15313:150000:1
source ume_store 10.29.3.78:16313:160000:1
source ume_store 10.29.3.78:17313:170000:1
source ume_store 10.29.3.79:18313:180000:2
source ume_store 10.29.3.79:19313:190000:2
source ume_store 10.29.3.79:29313:290000:2
source ume_store 10.29.3.79:39313:390000:2
source ume_store 10.29.3.79:49313:490000:2

source ume_message_stability_notification 1
source ume_store_behavior qc

source ume_store_group 0:5
source ume_store_group 1:3
source ume_store_group 2:5

source ume_retention_intragroup_stability_behavior quorum
source ume_retention_intergroup_stability_behavior any
       

See also Quorum/Consensus Store Usage and Quorum/Consensus - Mixed Location Groups.

7.1.2.8. Source Event Handler

The Source Event Handler is a function callback initialized at source creation to provide source events to your application related to the operation of the source. The following source code examples illustrate the use of a source event handler for registration events. To accept other source events, additional case statements would be required, one for each additional source event. See also UMP and UMQ Events.

C API

int handle_src_event(lbm_src_t *src, int event, void *ed, void *cd)
{
    switch (event) {
    case LBM_SRC_EVENT_UME_REGISTRATION_ERROR:
{
const char *errstr = (const char *)ed;
        printf("Error registering source with UME store: %s\n", errstr);
}
break;
        case LBM_SRC_EVENT_UME_REGISTRATION_SUCCESS_EX:
            {
                lbm_src_event_ume_registration_ex_t *reg = 
                        (lbm_src_event_ume_registration_ex_t *)ed;

                    printf("UME store %u: %s registration success. RegID %u. Flags %x ", 
                                reg->store_index, reg->store, reg->registration_id, 
                                reg->flags);
                    if (reg->flags & LBM_SRC_EVENT_UME_REGISTRATION_SUCCESS_EX_FLAG_OLD)
                        printf("OLD[SQN %x] ", reg->sequence_number);
                    if (reg->flags & LBM_SRC_EVENT_UME_REGISTRATION_SUCCESS_EX_FLAG_NOACKS)
                        printf("NOACKS ");
                    printf("\n");
            }
         break;
        case LBM_SRC_EVENT_UME_REGISTRATION_COMPLETE_EX:
            {
                    lbm_src_event_ume_registration_complete_ex_t *reg;

                        reg  = (lbm_src_event_ume__complete_ex_t *)ed;
                    printf("UME registration complete. SQN %x. Flags %x ", reg->sequence_number, 
                        reg->flags);
                    if (reg->flags & LBM_SRC_EVENT_UME_REGISTRATION_COMPLETE_EX_FLAG_QUORUM)
                     printf("QUORUM ");
                    printf("\n");
            }
         break;
        case LBM_SRC_EVENT_UME_STORE_UNRESPONSIVE:
        {
            const char *infostr = (const char *)ed;
            printf("UME store: %s\n", infostr);
        }
        break;
    default:
            printf("Unknown source event %d\n", event);
            break;
    }
   return 0;
}
       

JAVA API

public int onSourceEvent(Object arg, LBMSourceEvent sourceEvent)
{
    switch (sourceEvent.type()) {
    case LBM.SRC_EVENT_UME_REGISTRATION_ERROR:
System.out.println("Error registering source with UME store: "
+ sourceEvent.dataString());
break;
    case LBM.SRC_EVENT_UME_REGISTRATION_SUCCESS_EX:
 UMESourceEventRegistrationSuccessInfo reg = sourceEvent.registrationSuccessInfo();
 System.out.print("UME store " + reg.storeIndex() + ": " + reg.store()
+ " registration success. RegID " + reg.registrationId() + ". Flags "
+ reg.flags() + " ");
                if (((reg.flags() & LBM.SRC_EVENT_UME_REGISTRATION_SUCCESS_EX_FLAG_OLD)) 
                != 0) {
 System.out.print("OLD[SQN " + reg.sequenceNumber() + "] ");
                }
                if (((reg.flags() & LBM.SRC_EVENT_UME_REGISTRATION_SUCCESS_EX_FLAG_NOACKS)) 
                != 0) {
System.out.print("NOACKS ");
                }
                System.out.println();
                break;
    case LBM.SRC_EVENT_UME_REGISTRATION_COMPLETE_EX:
                UMESourceEventRegistrationCompleteInfo regcomp = 
                sourceEvent.registrationCompleteInfo();
                System.out.print("UME registration complete. SQN " + regcomp.sequenceNumber()
+ ". Flags " + regcomp.flags() + " ");
                if ((regcomp.flags() & 
                LBM.SRC_EVENT_UME_REGISTRATION_COMPLETE_EX_FLAG_QUORUM) != 0) {
System.out.print("QUORUM ");
                }
                System.out.println();
                break;
    case LBM.SRC_EVENT_UME_STORE_UNRESPONSIVE:
                System.out.println("UME store: "
                    + sourceEvent.dataString());
                break;
    ...
    default:
                System.out.println("Unknown source event "
+ sourceEvent.type());
                break;
        }
        return 0;
}
       

.NET API

public int onSourceEvent(Object arg, LBMSourceEvent sourceEvent)
{
    switch (sourceEvent.type()) {
    case LBM.SRC_EVENT_UME_REGISTRATION_ERROR:
System.Console.Out.WriteLine("Error registering source with UME store: "
+ sourceEvent.dataString());
break;
    case LBM.SRC_EVENT_UME_REGISTRATION_SUCCESS_EX:
 UMESourceEventRegistrationSuccessInfo reg = sourceEvent.registrationSuccessInfo();
 System.Console.Out.Write("UME store " + reg.storeIndex() + ": " + reg.store()
+ " registration success. RegID " + reg.registrationId() + ". Flags "
+ reg.flags() + " ");
                if (((reg.flags() & LBM.SRC_EVENT_UME_REGISTRATION_SUCCESS_EX_FLAG_OLD)) 
                != 0) {
 System.Console.Out.Write("OLD[SQN " + reg.sequenceNumber() + "] ");
                }
                if (((reg.flags() & LBM.SRC_EVENT_UME_REGISTRATION_SUCCESS_EX_FLAG_NOACKS)) 
                != 0) {
System.Console.Out.Write("NOACKS ");
                }
                System.Console.Out.WriteLine();
                break;
    case LBM.SRC_EVENT_UME_REGISTRATION_COMPLETE_EX:
                UMESourceEventRegistrationCompleteInfo regcomp = 
                sourceEvent.registrationCompleteInfo();
                System.Console.Out.Write("UME registration complete. SQN " + 
                regcomp.sequenceNumber()
+ ". Flags " + regcomp.flags() + " ");
                if ((regcomp.flags() & 
                LBM.SRC_EVENT_UME_REGISTRATION_COMPLETE_EX_FLAG_QUORUM) != 0) {
System.Console.Out.Write("QUORUM ");
                }
                System.Console.Out.WriteLine();
                break;
    case LBM.SRC_EVENT_UME_STORE_UNRESPONSIVE:
                System.Console.Out.WriteLine("UME store: "
                          + sourceEvent.dataString());
                break;
    ...
    default:
                System.Console.Out.WriteLine("Unknown source event "
+ sourceEvent.type());
                break;
        }
        return 0;
}
       

7.1.2.9. Source Event Handler - Stability, Confirmation and Release

As shown in Section 7.1.2.8 above, the Source Event Handler can be expanded to handle more source events by adding additional case statements. The following source code examples show case statements to handle message stability events, delivery confirmation events and message release (reclaim) events. See also UMP and UMQ Events.

C API

case LBM_SRC_EVENT_UME_MESSAGE_STABLE_EX:  
/* requires that source ume_message_stability_notification attribute is enabled */
        {
            lbm_src_event_ume_ack_ex_info_t *info = (lbm_src_event_ume_ack_ex_info_t *)ed;
    
            printf("UME store %u: %s message stable. SQN %x (msgno %d). Flags %x ", 
            info->store_index, info->store,
                    info->sequence_number, (int)info->msg_clientd - 1, info->flags);
            if (info->flags & LBM_SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_INTRAGROUP_STABLE)
                    printf("IA "); /* Stable within store group */
            if (info->flags & LBM_SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_INTERGROUP_STABLE)
                    printf("IR "); /* Stable amongst all stores */
            if (info->flags & LBM_SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_STABLE)
                    printf("STABLE ");  /* Just plain stable */
            if (info->flags & LBM_SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_STORE)
                    printf("STORE ");   /* Stability reported by UME Store */
            printf("\n");   
        }
        break;

case LBM_SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX:   
/* requires that source ume_confirmed_delivery_notification attribute is enabled */
        {
            lbm_src_event_ume_ack_ex_info_t *info = (lbm_src_event_ume_ack_ex_info_t *)ed;

                printf("UME delivery confirmation. SQN %x, Receiver RegID %u (msgno %d). Flags %x ",
                       info->sequence_number, info->rcv_registration_id, 
                       (int)info->msg_clientd - 1, info->flags);
                if (info->flags & LBM_SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_UNIQUEACKS)
                    printf("UNIQUEACKS ");   
                    /* Satisfied number of unique ACKs requirement */
                if (info->flags & LBM_SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_UREGID)
                    printf("UREGID ");       
                    /* Confirmation contains receiver application registration ID */
                if (info->flags & LBM_SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_OOD)
                    printf("OOD ");          
                    /* Confirmation received from arrival order receiver */
                if (info->flags & LBM_SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_EXACK)
                    printf("EXACK ");        
                    /* Confirmation explicitly sent by receiver */
                printf("\n");
        }
        break;
    
case LBM_SRC_EVENT_UME_MESSAGE_RECLAIMED:  
/* requires that source ume_confirmed_delivery_notification or ume_message_stability_notification 
attributes are enabled */
        {
                lbm_src_event_ume_ack_info_t *ackinfo = (lbm_src_event_ume_ack_info_t *)ed;

                printf("UME message released - sequence number %x (msgno %d)\n",
                       ackinfo->sequence_number, (int)ackinfo->msg_clientd - 1);
        }
        break;

       

JAVA API

case LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX: 
// requires that source ume_message_stability_notification attribute is enabled
        UMESourceEventAckInfo staInfo = sourceEvent.ackInfo();
        System.out.print("UME store " + staInfo.storeIndex() + ": "
                    + staInfo.store() + " message stable. SQN " + staInfo.sequenceNumber()
                    + " (msgno " + staInfo.clientObject() + "). Flags " 
                    + staInfo.flags() + " ");
        if ((staInfo.flags() & LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_INTRAGROUP_STABLE) 
        != 0) {
                System.out.print("IA "); // Stable within store group
        }
        if ((staInfo.flags() & LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_INTERGROUP_STABLE) 
        != 0) {
                System.out.print("IR ");  // Stable amongst all stores
        }
        if ((staInfo.flags() & LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_STABLE) != 0) {
                System.out.print("STABLE ");  // Just plain stable
        }
        if ((staInfo.flags() & LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_STORE) != 0) {
                System.out.print("STORE ");   // Stability reported by UME Store
        }
        System.out.println();
        break;

case LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX:  
// requires that source ume_confirmed_delivery_notification attribute is enabled
        UMESourceEventAckInfo cdelvinfo = sourceEvent.ackInfo();
        System.out.print("UME delivery confirmation. SQN " + cdelvinfo.sequenceNumber()
                    + ", RcvRegID " + cdelvinfo.receiverRegistrationId() + " (msgno "
                    + cdelvinfo.clientObject() + "). Flags " + cdelvinfo.flags() + " ");
        if ((cdelvinfo.flags() & LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_UNIQUEACKS) 
        != 0) {
                System.out.print("UNIQUEACKS "); // Satisfied number of unique ACKs requirement
        }
        if ((cdelvinfo.flags() & LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_UREGID) 
        != 0) {
                System.out.print("UREGID ");    // Confirmation contains receiver application 
                registration ID
        }
        if ((cdelvinfo.flags() & LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_OOD) 
        != 0) {
                System.out.print("OOD ");      // Confirmation received from arrival order 
                receiver
        }
        if ((cdelvinfo.flags() & LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_EXACK) 
        != 0) {
                System.out.print("EXACK ");    // Confirmation explicitly sent by receiver
        }
        System.out.println();
        break;

case LBM.SRC_EVENT_UME_MESSAGE_RECLAIMED:  
// requires that source ume_confirmed_delivery_notification or 
// ume_message_stability_notification attributes are enabled
        System.out.println("UME message released - sequence number "
                + Long.toHexString(sourceEvent.sequenceNumber())
                + " (msgno "
                + Long.toHexString(((Integer)sourceEvent.clientObject()).longValue())
                + ")");
        break;
       

.NET API

case LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX: 
// requires that source ume_message_stability_notification attribute is enabled
            UMESourceEventAckInfo staInfo = sourceEvent.ackInfo();
            System.Console.Out.Write("UME store " + staInfo.storeIndex() + ": "
                        + staInfo.store() + " message stable. SQN " + staInfo.sequenceNumber()
                        + " (msgno " + ((int)staInfo.clientObject()).ToString("x") + "). 
                        Flags " + staInfo.flags() + " ");
            if ((staInfo.flags() & LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_INTRAGROUP_STABLE) 
            != 0)
            {
                    System.Console.Out.Write("IA ");  // Stable within store group
            }
            if ((staInfo.flags() & LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_INTERGROUP_STABLE) 
            != 0) 
            {
                    System.Console.Out.Write("IR ");  // Stable amongst all stores
            }
            if ((staInfo.flags() & LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_STABLE) != 0)
            {
                    System.Console.Out.Write("STABLE ");  // Just plain stable
            }
            if ((staInfo.flags() & LBM.SRC_EVENT_UME_MESSAGE_STABLE_EX_FLAG_STORE) != 0)
            {
                    System.Console.Out.Write("STORE ");  // Stability reported by UME Store
            }
            System.Console.Out.WriteLine();
            break;

case LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX:  
// requires that source ume_confirmed_delivery_notification attribute is enabled

            UMESourceEventAckInfo cdelvinfo = sourceEvent.ackInfo();
            
            System.Console.Out.Write("UME delivery confirmation. SQN " + 
            cdelvinfo.sequenceNumber()
                        + ", RcvRegID " + cdelvinfo.receiverRegistrationId() + " (msgno "
                        + ((int)cdelvinfo.clientObject()).ToString("x") + "). Flags " + 
                        cdelvinfo.flags() + " ");
            if ((cdelvinfo.flags() & 
            LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_UNIQUEACKS) != 0)
            {
                    System.Console.Out.Write("UNIQUEACKS ");  // Satisfied number of unique 
                    ACKs requirement
            }
            if ((cdelvinfo.flags() & LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_UREGID) 
            != 0)
            {
                    System.Console.Out.Write("UREGID ");  // Confirmation contains receiver 
                    application registration ID
            }
            if ((cdelvinfo.flags() & LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_OOD) 
            != 0)
            {
                    System.Console.Out.Write("OOD ");  // Confirmation received from arrival 
                    order receiver
            }
            if ((cdelvinfo.flags() & LBM.SRC_EVENT_UME_DELIVERY_CONFIRMATION_EX_FLAG_EXACK) 
            != 0)
            {
                    System.Console.Out.Write("EXACK ");  // Confirmation explicitly sent by 
                    receiver
            }
            System.Console.Out.WriteLine();                    
            break;

case LBM.SRC_EVENT_UME_MESSAGE_RECLAIMED: 
// requires that source ume_confirmed_delivery_notification or 
// ume_message_stability_notification attributes are enabled
            
            System.Console.Out.WriteLine("UME message released - sequence number "
                               + sourceEvent.sequenceNumber().ToString("x")
                               + " (msgno "
                               + ((int)sourceEvent.clientObject()).ToString("x")
                               + ")");
            break;

       

7.1.2.10. Mapping Your Message Numbers to UMS/UMP Sequence Numbers

lbm_src_sendv_ex() allows you to create a pointer to an object or structure. This pointer will be returned to your application along with all source events. You can then update the object or structure with source event information. For example, if your messages exceed 8K - which requires fragmentation your application's message into more than one UM message - receiving sequence number events with this pointer allows you to determine all the UM sequence numbers for the message and, therefore, how many release (reclaim) events to expect. The following two source code examples show how to:

• Enable message sequence number information

• Handle sequence number source events to determine the application message number in the Source Event Handler

C API - Enable Message Information

lbm_src_send_ex_info_t exinfo;

/* Enable message sequence number info to be returned */
exinfo.flags = LBM_SRC_SEND_EX_FLAG_UME_CLIENTD | LBM_SRC_SEND_EX_FLAG_SEQUENCE_NUMBER_INFO;
exinfo.ume_msg_clientd = (void *)(msgno + 1); 
/* msgno set to application message number (can't evaluate to NULL) */
while (lbm_src_send_ex(src, message, msglen, 0, &exinfo) == LBM_FAILURE)
{
    if (lbm_errnum() == LBM_EUMENOREG)
        {
        printf("Send unsuccessful. Waiting...\n");
                SLEEP_MSEC(1000);   /* Sleep for 1 second */
        }
    else
    {
        fprintf(stderr, "lbm_src_send: %s\n", lbm_errmsg());
        break;
    }
}
       

C API - Sequence Number Event Handler

int handle_src_event(lbm_src_t *src, int event, void *ed, void *cd)
{
      switch (event) {
      case LBM_SRC_EVENT_SEQUENCE_NUMBER_INFO:
            {
                  lbm_src_event_sequence_number_info_t *info = 
                  (lbm_src_event_sequence_number_info_t *)ed;

                  if (info->first_sequence_number != info->last_sequence_number) {
                        printf("SQN [%x,%x] (msgno %d)\n", info->first_sequence_number, 
                        info->last_sequence_number, (int)info->msg_clientd - 1);
                  } else {
                        printf("SQN %x (msgno %d)\n", info->last_sequence_number, 
                        (int)info->msg_clientd - 1);
                }
            }
            break;
      ...
      }
      return 0;
}
       

JAVA API - Enable Message Information

LBMSourceSendExInfo exinfo = new LBMSourceSendExInfo();
exinfo.setClientObject(new Integer(msgno));  // msgno set to application message number
exinfo.setFlags(LBM.SRC_SEND_EX_FLAG_SEQUENCE_NUMBER_INFO);  
// Enable message sequence number info to be returned 
for (;;)
{
    try 
    {
        src.send(message, msglen, 0, exinfo);
    }
    catch(UMENoRegException ex)
    {
            try
            {
                Thread.sleep(1000);
            }
            catch (InterruptedException e) { }
        continue;

    }
    catch (LBMException ex)
    {
            System.err.println("Error sending message: " + ex.toString());  
    }
    break;
}
       

JAVA API - Sequence Number Event Handler

public int onSourceEvent(Object arg, LBMSourceEvent sourceEvent)
{
      switch (sourceEvent.type())
      {
            case LBM.SRC_EVENT_SEQUENCE_NUMBER_INFO:
                  LBMSourceEventSequenceNumberInfo info = sourceEvent.sequenceNumberInfo();
                  if (info.firstSequenceNumber() != info.lastSequenceNumber()) {
                        System.out.println("SQN [" + info.firstSequenceNumber()
                            + "," + info.lastSequenceNumber() + "] (msgno "
                            + info.clientObject() + ")");
                  }
                  else {
                        System.out.println("SQN " + info.lastSequenceNumber()
                            + " (msgno " + info.clientObject() + ")");
                  }
                  break;
          ...
      }
      return 0;
}
       

.NET API - Enable Message Information

LBMSourceSendExInfo exinfo = new LBMSourceSendExInfo();
exinfo.setClientObject(msgno);  // msgno set to application message number
exinfo.setFlags(LBM.SRC_SEND_EX_FLAG_SEQUENCE_NUMBER_INFO);  
// Enable message sequence number info to be returned
for (;;)
{
    try 
    {
        src.send(message, msglen, 0, exinfo);
    }
    catch(UMENoRegException ex)
    {
            System.Threading.Thread.Sleep(100);
        continue;

    }
    catch (LBMException ex)
    {
            System.Console.Out.WriteLine("Error sending message: " + ex.Message()); 
    }
    break;
}
       

.NET API - Sequence Number Event Handler

public void onSourceEvent(Object arg, LBMSourceEvent sourceEvent)
{
      switch (sourceEvent.type())
      {
            case LBM.SRC_EVENT_SEQUENCE_NUMBER_INFO:
                  LBMSourceEventSequenceNumberInfo info = sourceEvent.sequenceNumberInfo();
                  if (info.firstSequenceNumber() != info.lastSequenceNumber())
                  {
                        System.Console.Out.WriteLine("SQN [" + info.firstSequenceNumber()
                              + "," + info.lastSequenceNumber() + "] (cd "
                              + ((int)info.clientObject()).ToString("x") + ")");
                  }
                  else
                  {
                        System.Console.Out.WriteLine("SQN " + info.lastSequenceNumber()
                              + " (msgno " + ((int)info.clientObject()).ToString("x") + ")");
                  }
                  break;
          ...

      }
      return 0;
}
       

7.1.2.11. Receiver Liveness Detection

As an extension to Confirmed Delivery, you can set receivers to send a keepalive to a source during a measured absence of delivery confirmations (due to traffic lapse). In the event that neither message reaches the source within a designated interval, or if the delivery confirmation TCP connection breaks down, the receiver is assumed to have "died". UM then notifies the publishing application via context event callback. This lets the publisher assign a new subscriber.

To use this feature, set these five configuration options:

• ume_source_liveness_timeout

• ume_receiver_liveness_interval

• ume_confirmed_delivery_notification

• ume_user_receiver_registration_id

• ume_session_id

This specialized feature is not recommended for general use. If you are considering it, please note the following caveats:

• Do not use in conjunction with a UM Gateway.

• There is a variety of potential network occurrences that can break or reset the TCP connection and falsely indicate the death of a receiver.

• In cases where a receiver object is deleted while its context is not, the publisher may still falsely assume the receiver to be alive. Other false receiver-alive assumptions could be caused by the following:

• TCP connections can enter a half-open or otherwise corrupted state.

• Failed TCP connections sometimes do not fully close, or experience objectionable delays before fully closing.

• A switch or router failure along the path does not affect the TCP connection state.

7.1.3.1. Receiver RegID Management

RegIDs are slightly more involved for receivers than for sources. Since RegIDs are per source per topic per store and a topic may have several sources, a receiver may have to manage several RegIDs per store in use. Fortunately, receivers in UMP can leverage the RegID of the source with the use of a callback as discussed in Adding Fault Recovery with Registration IDs and shown in ume-example-rcv-2.c. Your application can determine the correct RegID to use and return it to UMP . You can also use Session IDs to enable UMP to manage receiver RegIDs. See Managing RegIDs with Session IDs.

Much like sources, receivers typically have a lifetime based on an amount of work, perhaps an infinite amount. And just like sources, it may be helpful to consider that a RegID is "assigned" at the start of that work and is out of use at the end. In between, the RegID is in use by the instance of the receiver application. However, the nature of RegIDs being per source means that the expected lifetime of a source should play a role in how RegIDs on the receiver are managed. Thus, it may be helpful for the application developer to consider the source application lifetime when deciding how best to handle RegIDs on the receiver.

7.1.3.2. Receiver Message and Event Handler

The Receiver Message and Event Handler is a function callback started at receiver initialization to provide Receiver messages to your application on behalf of the receiver. The following source code examples illustrate the use of a receiver message and event handler for registration messages. To accept other receiver events, additional case statements would be required, one for each additional event. See also UMP and UMQ Events.

C API

int rcv_handle_msg(lbm_rcv_t *rcv, lbm_msg_t *msg, void *clientd)
{
    switch (msg->type) {
    ...
    case LBM_MSG_UME_REGISTRATION_ERROR:
            printf("[%s][%s] UME registration error: %s\n", msg->topic_name, msg->source, 
                msg->data); 
            exit(0);
    break;
    case LBM_MSG_UME_REGISTRATION_SUCCESS:
            {
                    lbm_msg_ume_registration_t *reg = (lbm_msg_ume_registration_t *)
                        (msg->data);

                    printf("[%s][%s] UME registration successful. SrcRegID %u RcvRegID %u\n",
                     msg->topic_name, msg->source, reg->src_registration_id, 
                         reg->rcv_registration_id);
            }
         break;
    case LBM_MSG_UME_REGISTRATION_SUCCESS_EX:
            {
                    lbm_msg_ume_registration_ex_t *reg = (lbm_msg_ume_registration_ex_t *)
                        (msg->data);

                    printf("[%s][%s] store %u: %s UME registration successful. SrcRegID %u 
                        RcvRegID %u. Flags %x ",
                msg->topic_name, msg->source, reg->store_index, reg->store,
                reg->src_registration_id, reg->rcv_registration_id, reg->flags);
                if (reg->flags & LBM_MSG_UME_REGISTRATION_SUCCESS_EX_FLAG_OLD)
                     printf("OLD[SQN %x] ", reg->sequence_number);
                    if (reg->flags & LBM_MSG_UME_REGISTRATION_SUCCESS_EX_FLAG_NOCACHE)
                        printf("NOCACHE ");
                    printf("\n");
            }
            break;
    case LBM_MSG_UME_REGISTRATION_COMPLETE_EX:
         {
                    lbm_msg_ume_registration_complete_ex_t *reg;

reg  = (lbm_msg_ume_registration_complete_ex_t *)(msg->data);
printf("[%s][%s] UME registration complete. SQN %x. Flags %x ",
                    msg->topic_name, msg->source, reg->sequence_number, reg->flags);
                    if (reg->flags & LBM_MSG_UME_REGISTRATION_COMPLETE_EX_FLAG_QUORUM)
                     printf("QUORUM ");
                 if (reg->flags & LBM_MSG_UME_REGISTRATION_COMPLETE_EX_FLAG_RXREQMAX)
                        printf("RXREQMAX ");
                    printf("\n");
            }
        break;
    case LBM_MSG_UME_REGISTRATION_CHANGE:
            printf("[%s][%s] UME registration change: %s\n", msg->topic_name, msg->source, 
                msg->data);
            break;
    ...
    default:
            printf("Unknown lbm_msg_t type %x [%s][%s]\n", msg->type, msg->topic_name, 
                msg->source);
         break;
    }
    return 0;
}
       

JAVA API

public int onReceive(Object cbArg, LBMMessage msg) 
{
    case LBM.MSG_UME_REGISTRATION_ERROR:
            System.out.println("[" + msg.topicName() + "][" + msg.source()
                    + "] UME registration error: " + msg.dataString());
            break;
    case LBM.MSG_UME_REGISTRATION_SUCCESS_EX:
            UMERegistrationSuccessInfo reg = msg.registrationSuccessInfo();
            System.out.print("[" + msg.topicName() + "][" + msg.source()
                    + "] store " + reg.storeIndex() + ": "
                    + reg.store() + " UME registration successful. SrcRegID "
                    + reg.sourceRegistrationId() + " RcvRegID " + reg.receiverRegistrationId()
                    + ". Flags " + reg.flags() + " ");
            if ((reg.flags() & LBM.MSG_UME_REGISTRATION_SUCCESS_EX_FLAG_OLD) != 0)
                System.out.print("OLD[SQN " + reg.sequenceNumber() + "] ");
            if ((reg.flags() & LBM.MSG_UME_REGISTRATION_SUCCESS_EX_FLAG_NOCACHE) != 0)
                System.out.print("NOCACHE ");
            System.out.println();
            break;
    case LBM.MSG_UME_REGISTRATION_COMPLETE_EX:
            UMERegistrationCompleteInfo regcomplete = msg.registrationCompleteInfo();
            System.out.print("[" + msg.topicName() + "][" + msg.source()
                    + "] UME registration complete. SQN " + regcomplete.sequenceNumber()
                    + ". Flags " + regcomplete.flags() + " ");
            if ((regcomplete.flags() & LBM.MSG_UME_REGISTRATION_COMPLETE_EX_FLAG_QUORUM) 
            != 0) {
                System.out.print("QUORUM ");
            }
            if ((regcomplete.flags() & LBM.MSG_UME_REGISTRATION_COMPLETE_EX_FLAG_RXREQMAX) 
            != 0) {
                System.out.print("RXREQMAX ");
            }
            System.out.println();
            break;
    case LBM.MSG_UME_REGISTRATION_CHANGE:
            System.out.println("[" + msg.topicName() + "][" + msg.source()
                    + "] UME registration change: " + msg.dataString());
            break;
    ...
    default:
            System.err.println("Unknown lbm_msg_t type " + msg.type() + " ["
                    + msg.topicName() + "][" + msg.source() + "]");
            break;
        }
    return 0;
}
       

.NET API

public int onReceive(Object cbArg, LBMMessage msg) 
{
    case LBM.MSG_UME_REGISTRATION_ERROR:
            System. Console.Out.WriteLine("[" + msg.topicName() + "][" + msg.source()
                    + "] UME registration error: " + msg.dataString());
            break;
    case LBM.MSG_UME_REGISTRATION_SUCCESS_EX:
            UMERegistrationSuccessInfo reg = msg.registrationSuccessInfo();
            System.Console.Out.Write("[" + msg.topicName() + "][" + msg.source()
                    + "] store " + reg.storeIndex() + ": "
                    + reg.store() + " UME registration successful. SrcRegID "
                    + reg.sourceRegistrationId() + " RcvRegID " + reg.receiverRegistrationId()
                    + ". Flags " + reg.flags() + " ");
            if ((reg.flags() & LBM.MSG_UME_REGISTRATION_SUCCESS_EX_FLAG_OLD) != 0)
                    System.Console.Out.Write ("OLD[SQN " + reg.sequenceNumber() + "] ");
            if ((reg.flags() & LBM.MSG_UME_REGISTRATION_SUCCESS_EX_FLAG_NOCACHE) != 0)
                    System.Console.Out.Write ("NOCACHE ");
            System.Console.Out.WriteLine();
            break;
    case LBM.MSG_UME_REGISTRATION_COMPLETE_EX:
            UMERegistrationCompleteInfo regcomplete = msg.registrationCompleteInfo();
            System.Console.Out.Write("[" + msg.topicName() + "][" + msg.source()
                    + "] UME registration complete. SQN " + regcomplete.sequenceNumber()
                    + ". Flags " + regcomplete.flags() + " ");
            if ((regcomplete.flags() & LBM.MSG_UME_REGISTRATION_COMPLETE_EX_FLAG_QUORUM) 
            != 0) {
                System.Console.Out.Write("QUORUM ");
            }
            if ((regcomplete.flags() & LBM.MSG_UME_REGISTRATION_COMPLETE_EX_FLAG_RXREQMAX) 
            != 0) {
                System.Console.Out.Write("RXREQMAX ");
            }
System.Console.Out.WriteLine();
            break;
    case LBM.MSG_UME_REGISTRATION_CHANGE:
            System.Console.Out.WriteLine("[" + msg.topicName() + "][" + msg.source()
                    + "] UME registration change: " + msg.dataString());
            break;
    ...
    default:
            System.Console.Out.WriteLine("Unknown lbm_msg_t type " + msg.type() + " ["
                    + msg.topicName() + "][" + msg.source() + "]");
            break;
        }
    return 0;
}
       

7.1.3.3. Recovery Management

Recovery management for receivers is fairly simple because UMP requests any missing messages from the store(s) and delivers them as they are retransmitted. However, your application can specify a different message to begin retransmission with using either the retransmit_request_maximum configuration option or lbm_ume_rcv_recovery_info_ex_func_t.

For example, assume a source sends 7 messages with sequence numbers 0-6 which are stabilized at the store. The receiver, configured with the retransmit_request_maximum set to 2, consumes message 0, goes down, then comes back at message 6. lbm_ume_rcv_recovery_info_ex_func_t returns the following:

high_sequence_number = 6
low_rxreq_max_sequence_number = 4
low_sequence_number = 1

NOTE: low_rxreq_max_sequence_number = high_sequence_number - retransmit_request_maximum
       

• UMP obeys the retransmit_request_maximum configuration option and restarts with message 4. This is the default.

• If you modify the low_sequence_number to satisfy some other requirements, you can override the configuration option and restart at message 0, 2, 3, 5 or 6. See Setting Callback Function to Set Recovery Sequence Number below.

• The only way to restart at message 1 in this case, is to set the retransmit_request_maximum configuration option to its default value of 0. If your application changes the low_sequence_number and for whatever reason, the calculation results in the same value as the low_sequence_number, UMP ignores the calculation and restarts with message 4.

All messages retransmitted to a receiver are marked as retransmissions via a flag in the message structure. Thus it is easy for an application to determine if a message is a new message from the source or a retransmission, which may or may not have been processed before the failure. The presence or absence of the retransmit flag gives the application a hint of how best to handle the message with regard to it being processed previously or not.

7.1.3.4. Setting Callback Function to Set Recovery Sequence Number

The sample source code below demonstrates how to use the recovery sequence number info function to determine the stored message with which to restart a receiver. This method retrieves the low sequence number from the recovery sequence number structure and adds an offset to determine the beginning sequence number. The offset is a value completely under the control of your application. For example, if a receiver was down for a "long" period and you only want the receiver to receive the last 10 messages, use an offset to start the receiver with the 10th most recent message. If you wish not to receive any messages, set the low_sequence_number to the high_sequence_number plus one.

C API

lbm_ume_rcv_recovery_info_ex_func_t cb;

cb.func = ume_rcv_seqnum_ex;
cb.clientd = NULL;
if (lbm_rcv_topic_attr_setopt(&rcv_attr, "ume_recovery_sequence_number_info_function", 
&cb, sizeof(cb)) == LBM_FAILURE) {
            fprintf(stderr, 
            "lbm_rcv_topic_attr_setopt:ume_recovery_sequence_number_info_function: %s\n", 
            lbm_errmsg());
            exit(1);
}
printf("Will use seqnum info with low offset %u.\n", seqnum_offset); 

int ume_rcv_seqnum_ex(lbm_ume_rcv_recovery_info_ex_func_info_t *info, void *clientd)
{
     lbm_uint_t new_lo = info->low_sequence_number + seqnum_offset;

     printf("[%s] SQNs Low %x (will set to %x), Low rxreqmax %x, High %x (CD %p)\n",
info->source, info->low_sequence_number,
             new_lo, info->low_rxreq_max_sequence_number, info->high_sequence_number, 
info->source_clientd);
        info->low_sequence_number = new_lo;
        return 0;
}
       

JAVA API

UMERcvRecInfo umerecinfocb = new UMERcvRecInfo(seqnum_offset);
rcv_attr.setRecoverySequenceNumberCallback(umerecinfocb, null);
System.out.println("Will use seqnum info with low offset " + seqnum_offset + ".");

class UMERcvRecInfo implements UMERecoverySequenceNumberCallback {
    private long _seqnum_offset = 0;

    public UMERcvRecInfo(long seqnum_offset) {
        _seqnum_offset = seqnum_offset;
    }

    public int setRecoverySequenceNumberInfo(Object cbArg, 
    UMERecoverySequenceNumberCallbackInfo cbInfo) 
    {
long new_low = cbInfo.lowSequenceNumber() + _seqnum_offset;
if (new_low < 0) {
System.out.println("New low sequence number would be negative.  
Leaving low SQN unchanged.");
new_low = cbInfo.lowSequenceNumber();
         }
            System.out.println("SQNs Low " + cbInfo.lowSequenceNumber() + " (will set to "
                    + new_low + "), Low rxreqmax " + cbInfo.lowRxReqMaxSequenceNumber()
                 + ", High " + cbInfo.highSequenceNumber());
            try {
                    cbInfo.setLowSequenceNumber(new_low);
         }
            catch (LBMEInvalException e) {
                    System.err.println(e.getMessage());
        }
            return 0;
 }
       

.NET API

UMERcvRecInfo umerecinfocb = new UMERcvRecInfo(seqnum_offset);
rcv_attr.setRecoverySequenceNumberCallback(umerecinfocb, null);
System.Console.Out.WriteLine("Will use seqnum info with low offset " + seqnum_offset + ".");

class UMERcvRecInfo implements UMERecoverySequenceNumberCallback {
    private long _seqnum_offset = 0;

    public UMERcvRecInfo(long seqnum_offset) {
        _seqnum_offset = seqnum_offset;
    }

    public int setRecoverySequenceNumberInfo(Object cbArg, 
    UMERecoverySequenceNumberCallbackInfo cbInfo) 
    {
long new_low = cbInfo.lowSequenceNumber() + _seqnum_offset;
if (new_low < 0) {
System.Console.Out.WriteLine ("New low sequence number would be negative.  
Leaving low SQN unchanged.");
new_low = cbInfo.lowSequenceNumber();
         }
            System.Console.Out.WriteLine ("SQNs Low " + cbInfo.lowSequenceNumber() + " 
                (will set to "
                    + new_low + "), Low rxreqmax " + cbInfo.lowRxReqMaxSequenceNumber()
                 + ", High " + cbInfo.highSequenceNumber());
            try {
                    cbInfo.setLowSequenceNumber(new_low);
         }
            catch (LBMEInvalException e) {
                    System.Console.Out.WriteLine (e.getMessage());
        }
            return 0;
 }
       

7.1.3.5. Message Consumption

Receivers use message consumption, defined as message deletion, to indicate that UMP should notify the store(s) that the application consumed the message. This notification takes the form of an acknowledgement, or ACK, to the store(s) in use as well as to the source if you configured the source for delivery confirmation.

• In the C API, message deletion happens by default when the receive callback returns, unless the callback uses lbm_msg_retain(). If the callback uses lbm_msg_retain() then the application has responsibility to use lbm_msg_delete() when it has finished processing the message.

• In the Java API and .NET API, message deletion must be triggered explicitly by the application by using the dispose() method of the message object. Without explicit usage of dispose(), UMP does not know when the application has finished processing the message.

int rcv_handle_msg(lbm_rcv_t *rcv, lbm_msg_t *msg, void *clientd)
{
    lbm_ume_rcv_ack_t *ack = NULL;
...

    ack = lbm_msg_extract_ume_ack(msg);
    lbm_ume_ack_send_explicit_ack(ack, msg->sequence_number);
    lbm_ume_ack_delete(ack);
...

}
       

public int onReceive(Object cbArg, LBMMessage msg)
{
    UMEMessageAck ack;
...

    ack = msg.extractUMEAck();
    ack.sendExplicitAck(msg.sequenceNumber());
    ack.dispose();

...

}
       

7.1.4.1. Round-Robin Store Usage

Stores can be used in a Round-Robin fashion by a source during failover. A source provides UMP with a list of stores to use. The first is the primary, the second is the secondary, the third is the tertiary, etc. The source uses a single store at any one time. If the currently active store becomes unresponsive due to a crash or network disconnect, UMP tries other stores in the list one by one until it finds a responsive store.

With round-robin store usage, inactive stores do not receive data from the source. Thus, a store that becomes the active store will not have any data from the source. In this case, the source may be configured to retain messages and stream those messages to the new store using Late Join. Cascading failures of sources, stores and receivers may require using stores in a Quorum/Consensus fashion.

See also Sources Using Round-Robin Store Configuration.

7.1.4.2. Quorum/Consensus Store Usage

To provide the highest degree of resiliency in the face of failures, UMP provides the Quorum/Consensus failover strategy which allows a source to provide UMP with a number of stores to be used at the same time. Multiple stores can fail and UMP can continue operation unhindered. Moreover, Late Join is not needed as in Round-Robin.

Quorum/Consensus, also called QC, allows a source and the associated receivers to have their persisted state maintained at several stores at the same time. Central to QC is the concept of a group of stores, which is a logical grouping of stores that are intended to signify a single entity of resilience. Within the group, individual stores may fail but for the group as a whole to be viable and provide resiliency, a quorum must be available. In UMP , a quorum is a simple majority. For example, in a group of five stores, three stores are required to maintain a quorum. One or two stores may fail and the group continues to provide resiliency. UMP requires a source to have a quorum of stores available in the group in order to send messages. A group can consist of a single store.

QC also provides the ability to use multiple groups. As long as a single group maintains quorum, then UMP allows a source to proceed. Groups are logical in nature and can be combined in any way imaginable, such as by store location, store type, etc. In addition, QC provides the ability to specify backup stores within groups. Backups may be used if or when a store in the group becomes unresponsive to the source. Quorum/Consensus allows a source many different failure scenarios simply not available in other persistent messaging systems.

See also Sources Using Quorum/Consensus Store Configuration, Quorum/Consensus - Single Location Groups and Quorum/Consensus - Mixed Location Groups.

10.1.2.1. Acknowledgement Generation

Receivers in a persistence implementation of UMP send an a message consumption acknowledgement to stores and the message source. Some applications may want to control this acknowledgement explicitly themselves. In this case, ume_explicit_ack_only can be used.

10.1.2.2. Controlling Retransmission

Receivers in UMP during fault recovery are another matter entirely. Receivers send retransmission requests and receive and process retransmissions. Control over this process is crucial when handling very long recoveries, such as hundreds of thousands or millions of messages. A receiver only sends a certain number of retransmission requests at a time.

This means that a receiver will not, unless configured to with retransmit_request_outstanding_maximum, request everything at once. The value of the low sequence number (Receiver Recovery) has a direct impact on how many requests need to be handled. A receiving application can decide to only handle the last X number of messages instead of recovering them all using the option, retransmit_request_maximum. The timeout used between requests, if the retransmission does not arrive, is totally controllable with retransmit_request_interval. And the total time given to recover all messages is also controllable.

10.1.2.3. Recovery Process

Theoretically, receivers can handle up to roughly 2 billion messages during recovery. This limit is implied from the sequence number arithmetic and not from any other limitation. For recovery, the crucial limiting factor is how a receiver processes and handles retransmissions which come in as fast as UMP can request them and a store can retransmit them. This is perhaps much faster than an application can handle them. In this case, it is crucial to realize that as recovery progresses, the source may still be transmitting new data. This data will be buffered until recovery is complete and then handed to the application. It is prudent to understand application processing load when planning on how much recovery is going to be needed and how it may need to be configured within UMP .

10.1.3.1. Configuring Store Usage per Source

A store handles persisted state on a per topic per source basis. Based on the load of topics and sources, it may be prudent to spread the topic space, or just source space, across stores as a way to handle large loads. As configuration of store usage is per source, this is extremely easy to do. It is easy to spread CPU load via multi-threading as well as hard disk usage across stores. A single store process can have a set of virtual stores within it, each with their own thread.

10.1.3.2. Disk vs. Memory

As mentioned previously in UMP Stores, stores can be memory based or disk based. Disk stores also have the ability to spread hard disk usage across multiple physical disks by using multiple virtual stores within a single store process. This gives great flexibility on a per source basis for spreading data reception and persistent data load.

UMP stores provide settings for controlling memory usage and for caching messages for retransmission in memory as well as on disk. See Options for a Topic's ume-attributes Element. All messages in a store, whether in memory or on disk, have some small memory state. This is roughly about 72 bytes per message. For very large caches of messages, this can become non-trivial in size.

10.1.3.3. Activity Timeouts

UMP stores are NOT archives and are not designed for archival. Stores persist source and receiver state with the aim of providing fault recovery. Central to this is the concept that a source or receiver has an activity timeout attached to it. Once a source or receiver suspends operation or has a failure, it has a set time before the store will forget about it. This activity timeout needs to be long enough to handle the recovery demands of sources and receivers. However, it can not and should not be infinite. Each source takes up memory and disk space, therefore an appropriate timeout should be chosen that meets the requirements of recovery, but is not excessively long so that the limited resources of the store are exhausted.

10.1.4.1. UMP Configuration with NAT/Firewall

Although the diagram, Normal Operation, demonstrates the typical message interaction in UMP , sources, receivers, and stores may be arranged in almost limitless configurations. Some configurations make more sense than others for certain situations. One of those situations involves a Network Address Translation configuration (NAT) and/or Firewall. In such configurations, the source is the key element behind the NAT or Firewall. Although not the only viable NAT/Firewall configuration for UMP , the figure below demonstrates one approach to such an arrangement.

The lbmrd (29west Resolution Daemon) is an optional piece, but used in most situations where a NAT or Firewall is involved. It provides unicast support for topic resolution. The lbmrd and the store are placed on the outside (or at least are non-NATed or on a DMZ). Important characteristics of this configuration are:

• The LBMRD acts as a proxy for the topic resolution information.

• The store is accessible by the source and receiver directly.

In this situation, receivers and stores unicast control information back to the source, therefore the NAT or Firewall router needs to port forward information back to the source.

10.1.4.2. Quorum/Consensus - Single Location Groups

Quorum/Consensus provides a huge set of options for store arrangements in UMP . Between backups and groups, the number of viable approaches is practically limitless. Below are two approaches using single location groups and multiple location groups.

In short, as long as one of the groups in the figure maintains quorum, then the source can continue. See Sources Using Quorum/Consensus Store Configuration to view a UM configuration file for this example.

The above figure shows three groups arranged on a location basis. Each group is a single location. Just SOME possible failure scenarios are:

• Failure of any 3 stores in Group 0

• Failure of any 1 store in Group 1

• Failure of any 2 stores in Group 2

• Failure of all stores in Group 0 and 1

• Failure of all stores in Group 1 and 2

• Failure of all stores in Group 0 and 2

10.1.4.3. Quorum/Consensus - Mixed Location Groups

Groups of stores can be configured across locations. Such an arrangement would ensure continued operation in the event of a site-wide failure at any location.

The figure above shows two groups arranged in a mixed location manner. Essentially, one location can totally fail and a source can continue sending because the other location has a group with a quorum. See below for an UM configuration file for this example.

source ume_store 10.16.3.77:10313:101000:0
source ume_store 10.16.3.78:11313:110000:1
source ume_store 10.16.3.79:12313:120000:1
source ume_store 192.168.0.44:15313:150000:1
source ume_store 192.168.0.45:16313:160000:0
source ume_store 192.168.0.46:17313:170000:0

source ume_message_stability_notification 1
source ume_store_behavior qc

source ume_store_group 0:3
source ume_store_group 1:3

source ume_retention_intragroup_stability_behavior quorum
source ume_retention_intergroup_stability_behavior any
       

12.2.1.1. Child Elements of the Store Element

The following table gives the child elements allowed in the store configuration section.

12.2.1.2. Options for a Store's ume-attributes Element

You can configure context (scope) options with a type attribute of lbm-context. UM passes such options through the normal receiver and context configuration option setting mechanisms. See the UM Configuration Guide for details. Store options without a type attribute or those explicitly given a type attribute of store simply configure the store itself.

The following table gives options allowed for a store element. Use the store Option Type for these options. A Store's ume-attributes Element can also accept the lbm-context Option Type. See Option Types for ume-attributes Elements for more information.

12.2.2.1. Topic Element

The Topic Element defines an individual topic persisted on the UMP store. The following table gives attributes for the topic element.

The Topic Element has one child element, ume-attributes, the options for which appear in Options for a Topic's ume-attributes Element.

12.3.1.1. Queue Groups Element

The queue-groups element contains queue-group elements that define all the queue groups that make up the queue. In the abbreviated Queues section shown in Queues Element, the queue element,

  <queue name="Queue 1" interface=10.29.3.24" port="20555" group-index="0">
       

specifies Queue 1 as a Queue Instance in Group 0. The queue-groups element in the same sample specifies that Queue 1 comprises two groups, Group 0, which has 5 queue instances and Group 1, which has 1 queue instance.

    <queue-groups>
      <queue-group index="0" size="5"/>
      <queue-group index="1" size="1"/>
    </queue-groups>
   

To completely configure Queue 1, you must specify 6 queue instances in either one umestored.xml file as individual queue elements within the queues element or in separate umestored.xml files, one for each instance. The Queue Element for all 6 queue instances would be the same except for the interface, port and group-index because every instance of Queue 1 must have the same name. And all 6 queue instances would also have the same queue-groups configured.

12.3.1.2. General Options for a Queue's ume-attributes Element

The table below displays the general options available for a Queue Element. Use the queue Option Type for these options. A Queue's ume-attributes Element can also accept Option Types lbm-receiver, lbm-context and lbm-source. See Option Types for ume-attributes Elements for more information.

12.3.1.3. Message Storage Options for a Queue's ume-attributes Element

UMQ provides 3 basic modes or operation for message storage.

• Memory Storage: UMQ stores messages and message state only in memory. A restarted umestored does not resume any previous operation. In this configuration, sinc-data-filename, sinc-queue-swap-filename, and sinc-log-filename are not set.

• Disk Storage, no Persistence: UMQ swaps messages and message state from memory to disk as needed to maintain low memory overhead. A restarted umestored does not resume any previous operation. In this configuration, sinc-data-filename and sinc-queue-swap-filename are set. The sinc-log-filename is not set.

• Disk Storage, Persistence: UMQ swaps messages and message state from memory to disk as needed to maintain low memory overhead. In addition, UMQ maintains a separate log file that contains all operations needed to reconstruct state when UMQ restores umestored. In this configuration, sinc-data-filename, sinc-queue-swap-filename, and sinc-log-filename are set.

Use the queue Option Type for these options. A Queue's ume-attributes Element can also accept Option Types lbm-receiver, lbm-context and lbm-source. See Option Types for ume-attributes Elements for more information.

12.3.1.4. Queue Management Options for a Queue's ume-attributes Element

The Queue Management options control how queue instances communicate regarding elections and how they monitor each other.

Use the queue Option Type for these options. A Queue's ume-attributes Element can also accept Option Types lbm-receiver, lbm-context and lbm-source. See Option Types for ume-attributes Elements for more information.

12.3.1.5. Queue Slave Instance (QSI) Options for a Queue's ume-attributes Element

Queue Slave Instances (QSIs) receive assignment information from the master queue using the control topic. These messages are called Receiver Control Records (RCR). If a queue instance misses an RCR, it can request it. The options below control the operation of requesting RCRs.

The acronym, qrcrr, refers to Queue Receiver Control Records Request.

Use the queue Option Type for these options. A Queue's ume-attributes Element can also accept Option Types lbm-receiver, lbm-context and lbm-source. See Option Types for ume-attributes Elements for more information.

12.3.5.1. Application Set Element

The Application Set element defines an individual application set and has only one attribute, name, which identifies the application set. Each application set also requires a Receiver Type. See Options for an Application Set's ume-attributes Element for application set options.

12.3.5.2. Options for an Application Set's ume-attributes Element

Options for an application set appear in the following table. Use the queue Option Type for these options. See Option Types for ume-attributes Elements for more information.

12.3.5.3. Receiver Types Element

The Receiver Types element is a container element for all the receiver types within the Application Set.

<receiver-type id="100">
  <ume-attributes> ... </ume-attributes>
  <index-rules order="allow, deny">
    <index-allow name="Red Messages" />
    <index-deny name="Brown Messages" />
  </index-rules>
</receiver-type>
       

12.3.6.1. Queue Topic Element

The Topic Element for a queue uses the same attributes for store topics listed in Topic Element and must also specify an Application Set that is configured in the Application Sets section. The options listed in Options for a Queue Topic's ume-attributes Element are also available.

12.3.6.2. Options for a Queue Topic's ume-attributes Element

Options for a queue topic appear in the following table. Use the queue Option Type for these options. A Queue's ume-attributes Element can also accept Option Types lbm-receiver, lbm-context and lbm-source. See Option Types for ume-attributes Elements for more information.