Conditional topic updates

Multiple overlapping producers

In another situation, you might have multiple independent producers that are updating the same topic. Again, you don’t want to overwrite a topic with an older value than it currently contains. Consider two producers, sending the same data at slightly different times:

Helpfully, Diffusion (as long as you are not using the DONT_RETAIN_VALUE flag), doesn’t send an update when there is no change to a topic, so the client doesn’t get all 8 possible updates. But, for some period of time, :green-4: is overwritten by :red-3: and old data is given to the client. Clearly, we don’t want this to happen!

Example usage

Let’s take a sample series of messages with which we want to update a topic:

Like our first use case, we have data here that is not in order. We will use UpdateConstraints to make sure that the message with seqnum = 3 does not overwrite the topic after seqnum = 4. The rules that govern updating the topic are:

1. If the topic does not exist, we can add it.

2. If the topic exists, but has no data, we can update it.

3. If the topic exists and has data, the seqnum field must contain a value that is less than the one in the message we are using for the update.

To do this, we create an UpdateConstraint with the following rules:

final UpdateConstraint constraint =
    Diffusion.updateConstraints()
             .noTopic()
             .or(Diffusion.updateConstraints().noValue())
             .or(Diffusion.updateConstraints().jsonValue()
                 .with("/seqnum",
                       UpdateConstraint.Operator.LT,
                       thisSeqNum));

Available condition operators

LT is not the only comparison operator available. Here is a full list:

In all cases except for IS, when the value can be interpreted as a number such as a Long or a Double it is, and this includes when it is formatted as a string. For example, LT 5 is interpreted the same as LT "5.0".

The difference between IS and EQ is that IS performs an exact comparison at the byte level, whereas EQ will check to see if the current and new values are equivalent:

Full example

The following class shows how to send JSON messages as per the above example.

package com.diffusiondata.examples;

import com.pushtechnology.diffusion.client.Diffusion;
import com.pushtechnology.diffusion.client.features.TopicUpdate;
import com.pushtechnology.diffusion.client.features.UpdateConstraint;
import com.pushtechnology.diffusion.client.session.Session;
import com.pushtechnology.diffusion.client.topics.details.TopicSpecification;
import com.pushtechnology.diffusion.client.topics.details.TopicType;
import com.pushtechnology.diffusion.datatype.json.JSON;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

public class Main {
    public static void main(String[] args) throws Exception {
        final Session session = Diffusion.sessions()
                                         .principal("admin")
                                         .password("password")
                                         .open("ws://localhost:8090");

        final TopicSpecification topicSpec = Diffusion.newTopicSpecification(TopicType.JSON);

        // A sequence of JSON-formatted messages to send to Diffusion.
        // Note the out-of-order seqnum 3.
        final String[] messages = {
            "{\"seqnum\": 1, \"price\": 1.50}",
            "{\"seqnum\": 2, \"price\": 1.52}",
            "{\"seqnum\": 4, \"price\": 1.54}",
            "{\"seqnum\": 3, \"price\": 1.53}",
            "{\"seqnum\": 5, \"price\": 1.55}",
        };

        final ObjectMapper mapper = new ObjectMapper();

        for (String thisMessage : messages) {
            final JsonNode node = mapper.readTree(thisMessage);
            final int thisSeqNum = node.get("seqnum").asInt();

            final UpdateConstraint constraint =
                Diffusion.updateConstraints()
                         .noTopic()
                         .or(Diffusion.updateConstraints().noValue())
                         .or(Diffusion.updateConstraints().jsonValue().with("/seqnum",
                                                                            UpdateConstraint.Operator.LT,
                                                                            thisSeqNum));

            final JSON myJsonMessage = Diffusion.dataTypes().json().fromJsonString(thisMessage);

            session.feature(TopicUpdate.class)
                   .addAndSet("my-topic", topicSpec, JSON.class, myJsonMessage, constraint)
                   .whenComplete((result, err) -> {
                       if (err == null) {
                           System.out.println("Updated topic with seqnum="
                                              + thisSeqNum);
                       } else {
                           System.err.println("Failed to update topic with seqnum="
                                              + thisSeqNum + ": " + err.getMessage());
                       }
                   });
    }

    // Give some time for futures to complete
    Thread.sleep(1000);

    // Done.
    session.close();
}