Skip to main content

Data Synchronization: MQTTSync

A core feature of Transitive is MQTTSync, the data synchronization framework. MQTTSync synchronizes data between the participants: robots, cloud, and web front-ends. It is build on top of MQTT and abstracts from its message-based pub/sub model to a shared-data model. This means that all participants, whether a process running on the robot or a web-component running in the browser, can read and write the data locally without needing to worry about synchronizing changes with other participants. This is similar to how cloud-backup software keeps files in-sync.

Data Model​

The model of the shared data is a humongeous JSON object. However, that's just the model. This JSON object is never materialized in one place as such. Instead it is represented by a collection of MQTT topics and their values, and stored in the retained messages store of an MQTT broker (currently mosquitto).

For illustration: imagine you'd like to store the JSON object

{
  "a": {
    "b": {
      "c": {
        "d": {
          "e": "test"
        },
        "f": "test2"
      }
    }
  }
}

This can be presented by the following two MQTT topics and values:

/a/b/c/d/e: "test"
/a/b/c/f: "test2"

We call this representation "flat". Note that this is not a unique representation. It would be equally valid to represent the object as:

/a/b/c/d: {"e": "test"}
/a/b/c/f: "test2"

or

/a/b/c: {"d": {"e": "test"}, "f": "test2"}

This way of representing the JSON object has several advantages: It makes it easy to selectively share and update sub-objects, it allows for efficient distributed storage (if necessary), and still allows for the specification of "atomic" updates, i.e., sub-objects that are required to be updated as one, rather than as flat topics that may or may not arrive on different devices at different times. This can be very important for the application logic as we will see.

Layout​

Specifically, the humongeous JSON object is structured as follows:

{
  "orgId": {
    "deviceId": {
      "capabilityScope": {
        "capabilityName": {
          "capabilityVersion": {
            // whatever the capability wants to store
          }
        },
      }
    }
  }
}

Accordingly, the underlying MQTT topics take the form:

/orgId/deviceId/capabilityScope/capabilityName/capabilityVersion/...

Here:

  • orgId corresponds to the username on the Portal.
  • deviceId is an id generated at the time the Transitive agent is installed on a device. It is only required to be unique within the orgId namespace.
  • capabilityScope is @local in capabilities only available locally in a self-deployed environment and @transitive-robotics when the capability is available from transitiverobotics.com.
  • capabilityName is the name of the capability, e.g., webrtc-video.
  • capabilityVersion is any sub-set of a semantic version, i.e., M, M.m, or M.n.p, where M stands for major, m for minor, and p for patch. It is up to the capability to choose at what level of versioning it wants to version its namespace. The default is to use the whole version, M.n.p, in which case each new patch version of the capability will operate in a separate namespace. But for many capabilities it is beneficial to share a namespace between patches. This reduces the need for migrations from one namespace to another upon upgrades, and allow for better semantic versioning.

Within that namespace, the capability has full control over the data. A common pattern is to further sub-divide into robot, cloud, and client namespaces such that each is only written by the corresponding participant. But this is not required. It is also fine for multiple participants to write to the same fields.

Usage​

Capabilities read from and write to this shared data model via the MQTTSync class. This class provides functions to selectively subscribe to parts of the model and publish changes to it.

Publish and subscribe​

If a participant wants to read from a sub-object, it first subscribes to it. This means it will receive the current state of that sub-object and it will receive updates in real-time as they are made by other participants. Likewise, sub-objects the participant wants to write to are published.

For example, in the health-monitoring capability, the robot publishes:

mqttSync.publish('/diagnostics');

which the web-client subscribes to:

mqttSync.subscribe(`${prefix}/diagnostics`);

Note in this example that on the robot there is no need to prefix the topic with the namespace as described above. This is taken care of the local MQTT bridge running on the robot, which is part of the sandboxing mechanism. On the client, prefix would be something like /user1/device123/@transitive-robotics/health-monitoring/0.6.3/.

Reading and changing data​

Data that is subscribed or published is locally stored in MqttSync.data, an instance of the DataCache class. This object provides mechanisms for reading and writing.

Reading data is straightforward, since locally that model materializes as a JSON object fragment. Hence it can be accessed like any other JSON object. When using React on the web, one can use the useMqttSync hook to further get a reactive data object. Changes to this object will trigger reactive re-renders in React.

Writing to the shared data happens via DataCache.update(key, value) where key is either a string representing a topic, e.g., /diagnostics/temp, or an array representing a path, e.g., ['diagnostics', 'temp']. The value can be a primitive or a JSON object, as described above.

One of the roles the DataCache class plays in the context of MQTTSync is that of de-duplication: when an update does not actually change the value of a field (because the new value is the same as the old), no messages will be propagated over MQTT. This can save a lot of bandwidth, especially when ingesting data from ROS topics, where it is very common that only part of the fields in the topic change while others stay the same.

Try it​

Perhaps the best way to learn about this data model is by using it. When you create a new capability using the template code it will setup a basic capability that uses MQTTSync in all three places: the robot code, the cloud code, and the front-end. Using that code as a starting point, you can explore and experiment with its usage.