Legacy conflation versus new topic properties system

Other topic types use a new system based on topic properties, described in Configuring conflation.

If the new system is used to set an "off" policy on a deprecated topic, that policy will be applied and any legacy conflation policy will be ignored.

If the new system is used to set an "unsubscribe" policy on a deprecated topic, the legacy policy is evaluated for each message, and the "unsubscribe" policy is applied when the message queue is full.

If the new system is used to set an "always" or "conflate" policy, the legacy conflation policy will be applied and the new system will not otherwise be used.

Legacy conflation policies

One or more conflation policies can be configured, each defining different conflation mechanisms by using conflation-policy elements. Conflation policies comprise the following:

Having defined one or more conflation policies, you can map topics to them. This is done by specifying a topic path or topic selector which maps to a particular conflation policy.

Conflation policies can be added or removed at runtime and the removal of a conflation policy automatically removes any mappings to it.

Conflation policy mode

The conflation policy mode determines whether the new (or merged) message is to replace the existing message in the client queue or be appended to the end of the client queue.

Available modes are:

If no mode is specified, REPLACE is assumed.

The mode is specified in the mode property of a conflation-policy section in etc/Server.xml.

When defining conflation policies programmatically the mode is specified when creating the policy.

Message matchers

A message matcher is used by a conflation policy when queuing a new message for a client that has conflation enabled for a topic that has a conflation policy defined for it. The message matcher is used to locate the last message queued for a client that is a candidate for conflation.

If no message matcher is explicitly defined for a conflation policy, a default matcher is used which locates a message of the same topic.

A message matcher can be supplied if the matching is to be somehow dependent upon the content of the messages.

An example of a MessageMatcher implementation is shown below:

public class ExampleMessageMatcher implements messageMatcher { 
    @Override 
    public boolean matches(TopicMessage currentMessage,TopicMessage newMessage) { 
        return currentMessage.nextField().equals(newMessage.nextField()); 
    } 
}

MessageMatcher implementations must be thread safe and stateless. The same MessageMatcher instance can be supplied to more than one different conflation policy if so required.

Message mergers

A message merger can be specified on a conflation policy if the action of the policy is to merge the content of an existing queued message with the new message being queued. This technique can be used when message data comprises more than one data item and it is desirable to reduce the number of messages sent to the client whilst preserving the data from all messages.

If no message merger is specified for a conflation policy, the policy replaces the current message with the new.

The action of conflation will depend upon the message that is returned from the merger method, as follows:

An example of a message merger implementation is shown below:

package com.pushtechnology.diffusion.examples;

import com.pushtechnology.diffusion.api.message.MessageReader;
import com.pushtechnology.diffusion.api.message.Record;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.pushtechnology.diffusion.api.APIException;
import com.pushtechnology.diffusion.api.conflation.MessageMerger;
import com.pushtechnology.diffusion.api.message.TopicMessage;
import com.pushtechnology.diffusion.api.publisher.Publishers;

import java.util.ArrayList;
import java.util.List;

public final class MessageMergerExample implements MessageMerger {
    private static final Logger LOG =
            LoggerFactory.getLogger(MessageMergerExample.class);

    @Override
    public TopicMessage merge(TopicMessage currentMessage, TopicMessage newMessage) throws APIException {

        final MessageReader currentMessageReader = currentMessage.getReader();
        final MessageReader newMessageReader = newMessage.getReader();

        final TopicMessage result = Publishers.createDeltaMessage(currentMessage.getTopicName());

        Record cRecord = currentMessageReader.nextRecord();
        Record nRecord = newMessageReader.nextRecord();

        while (nRecord != null) {
            final List<String> mergedRecord = new ArrayList<>(nRecord.size());
            for (int i = 0; i < nRecord.size(); i++) {
                final String nField = nRecord.getField(i);
                if (!nField.isEmpty() || cRecord == null || i >= cRecord.size()) {
                    mergedRecord.add(nField);
                }
                else {
                    mergedRecord.add(cRecord.getField(i));
                }
            }

            result.putRecord(mergedRecord);

            nRecord = newMessageReader.nextRecord();
            cRecord = currentMessageReader.nextRecord();
        }

        if (LOG.isTraceEnabled()) {
            LOG.trace("MessageMerger merging - currentMessage: {}, newMessage: {}, merged: {}",
                    currentMessage.asRecords(), newMessage.asRecords(), result.asRecords());
        }

        return result;
    }
}

The above example merges delta messages for record topics with variable records and fields.

MessageMerger implementations must be thread safe and stateless. The same MessageMerger instance can be supplied to more than one different conflation policy if so required.

Default conflation policy

You can specify a default conflation policy that is used for any topics that do not have explicit policy mappings.

Use a default conflation policy only if you want to apply conflation to all topics when conflation is enabled for a client.

This can be specified using the default-conflation-policy property in the conflation section of etc/Server.xml. Alternatively it can be set programmatically at any time using the setDefaultPolicy method on ConflationConfig.

If no default policy is set, conflation will not occur for topics that have no explicit mappings even when conflation is enabled.

Mapping topics to policies

Having defined one or more conflation policies, you can map topics to the conflation policies that are to be used for them.

Use a topic selector pattern when mapping to a conflation policy.

As the use of topic selectors makes it possible for more than one mapping to potentially apply to the same topic, the last mapping defined that matches a specific topic is the one that is used for conflating messages of that topic.

A default conflation policy can be specified which is selected to map to if no other mapping matches a topic.

Messages for topics that have no mappings (when there is no default policy) are not conflated, even if conflation is enabled for a client.

Conflation mappings can be defined using topic-conflation elements within the conflation section of the etc/Server.xml property file.

Conflation mappings can also be set programmatically using the setTopicPolicy method of ConflationConfig. Mappings can be set at any time during the running of a server. Mappings can also be removed at any time using unsetTopicPolicy.

Enabling conflation

Specify conflation for a queue-definition by setting the conflates property to true. This queue definition can then be used wherever required, for example by connectors that have conflation enabled for all clients.