MQTT Messages

To ingest data, send an MQTT message to the MQTT changes topic associated with the target dataset. The general format of the topic name is <applicationID>/dataset/<datasetID>/changes.

You must connect to the Flo.w MQTT broker with an API key that grants 'write' permission to the topic. See [MQTT Authentication] for details of how to connect to the MQTT broker.

Message Payload

Flo.w Real-Time message payloads are JSON-formatted strings with the following general form:

{
  "action": 'insert|upsert|delete',
  "data": {} | [{}, {}, ...]
}

Property	Description
action	The ingest action to perform: insert, upsert or delete
data	The data to ingest.

The data property can be a single data object (with properties specifying dataset attributes) or an array of data objects. An array of data objects is guaranteed to be processed in a single database transaction.

Batch Data

Prefer to use the array format if data is naturally batched.

Ingesting batches of data using the array format minimizes the number of MQTT messages sent and database operations performed.

Data Attribute Types

Flo.w Real-Time converts the attribute types of the supplied JSON data to the corresponding database column types before ingesting. See the following table for notes about exceptions:

Data Type	Notes
Timestamps	Database columns of type TIMESTAMP WITH TIME ZONE should be set using an ISO8601 (opens new window)-formatted string. We recommend always working in Coordinated Universal Time (UTC).
Geometry	Geometry columns should be set using a WKT (opens new window)-formatted string. You should include the SRID, which specifies the spatial reference system of the geometry. For example, SRID=4326;POINT(-1.2 51.4).

Insert Action

The insert action specifies that the supplied data should be inserted into the associated dataset. This is equivalent to performing a SQL INSERT INTO statement.

TIP

Insert actions are typically used to insert real-time data into a dataset to build a time series. See Time Series Data.

All primary key fields and non-nullable fields must be specified in the supplied data. Nullable fields that are not specified will be set to null. Database errors such as primary key constraint errors, unique index errors or missing required fields will be reported to the error topic associated with the target dataset.

An example insert message:

{
  "action": "insert",
  "data": [{"id": 1, "value": "a"}, {"id": 2, "value": "b"}]
}

Upsert Action

The upsert action specifies that the supplied data should be inserted or updated (if already present). How this is performed is dependent on the database driver used. For the standard Postgres driver an 'upsert' is performed using the insert on conflict do update pattern.

TIP

Upsert actions are typically used to maintain a 'last-known values' dataset.

Database errors such as primary key constraint errors, unique index errors or missing required fields will be reported to the error topic associated with the target dataset.

An example upsert message:

{
  "action": "upsert",
  "data": [{"id": 1, "value": "c"}, {"id": 2, "value": "d"}]
}

Delete Action

The delete action specifies that the supplied data should be deleted from the associated dataset. This is equivalent to performing a SQL DELETE FROM statement.

TIP

Delete actions are typically used to remove stale entries from a 'last-known values' dataset.

Supplied data objects should specify the primary key(s) of records to delete. Other fields will be ignored. Database errors such as primary key constraint errors, unique index errors or missing required fields will be reported to the error topic associated with the target dataset.

An example delete message:

{
  "action": "delete",
  "data": [{"id": 1}, {"id": 2}]
}