Operations Reference - Coreflux Documentation

Overview

Complete reference for all LoT operations available within actions. These operations allow you to manipulate variables, perform calculations, transform strings, read and write MQTT topics, and implement conditional logic.

Quick Reference: Use the sidebar or the index table below to jump to specific operations.

In This Page

Operation	Purpose	Example
`SET`	Create/assign variables	`SET "temp" WITH 25.5`
`GET TOPIC`	Read from MQTT topic	`GET TOPIC "sensors/temp"`
`GET JSON`	Extract from JSON payload	`GET JSON "data.value" IN PAYLOAD`
`GET ENV`	Read environment variable	`GET ENV "DB_HOST"`
`GET SECRET`	Read encrypted secret	`GET SECRET "DB_PASSWORD"`
`+, -, *, /, %`	Arithmetic operations	`SET "sum" WITH ({a} + {b})`
`+ (strings)`	String concatenation	`"Hello " + {name}`
`RANDOM`	Generate random values	`RANDOM BETWEEN 1 AND 100`
`FILTER`	Extract with regex	`FILTER PAYLOAD USING REGEX "[0-9]+"`
`REPLACE`	String replacement	`REPLACE " " WITH "_" IN {text}`
`PUBLISH`	Write to MQTT topic	`PUBLISH TOPIC "output" WITH {temp}`
`KEEP TOPIC`	Persist internal state	`KEEP TOPIC "cache/last" WITH PAYLOAD`
`IF/THEN/ELSE`	Conditional logic	`IF {temp} > 80 THEN ...`
`TIMESTAMP`	Generate timestamps	`TIMESTAMP "UTC"`
`CALL ACTION`	Invoke callable actions	`CALL ACTION Calculate WITH ...`
`TRIGGER`	Trigger route events	`TRIGGER "Route" TO {data} FOR mapping`
`CALL MCP`	Invoke MCP tools	`CALL MCP "Route.tool" WITH ...`

Variable Operations

SET, GET TOPIC, and GET JSON are the core operations for working with data in actions. They allow you to store values, read from MQTT topics, and extract fields from JSON payloads.

SET

Creates or assigns a value to an internal variable:

SET "variable_name" WITH <value>

Variables exist only during action execution and are referenced using {variable_name}.

Static Value
From Topic
Calculated

SET "threshold" WITH 100
SET "status" WITH "active"
SET "enabled" WITH TRUE

SET "current_temp" WITH (GET TOPIC "sensors/temperature")
// For topic "sensors/temp001/data": POSITION 2 gets the sensor ID
SET "sensor_id" WITH TOPIC POSITION 2

SET "total" WITH (GET TOPIC "count/a" + GET TOPIC "count/b")
SET "average" WITH ({sum} / {count})

Variable names must be enclosed in double quotes when using SET, but referenced with curly braces: SET "myVar" WITH 10 then use {myVar}.

GET TOPIC

Retrieves the last known value from a topic. If the topic has no value, the broker returns a default based on the requested type.

GET TOPIC "topic/path"
GET TOPIC "topic/path" AS <TYPE>

Default Values for Missing Topics

Type	Default Value
`STRING`	`""` (empty string)
`DOUBLE`	`0.0`
`INT`	`0`
`BOOL`	`FALSE`

Type Casting

GET TOPIC "sensors/temp" AS DOUBLE
GET TOPIC "config/count" AS INT
GET TOPIC "flags/enabled" AS BOOL

Wildcard Inheritance

When an action is triggered by a wildcard topic pattern, GET TOPIC inherits the matched wildcard values:

DEFINE ACTION ProcessWithContext
ON TOPIC "sensors/+/temperature" DO
    // If triggered by "sensors/temp001/temperature"
    // The + wildcard matches "temp001"
    
    // This GET TOPIC uses the same wildcard value:
    SET "humidity" WITH (GET TOPIC "sensors/+/humidity" AS DOUBLE)
    // Reads from "sensors/temp001/humidity"

Example Usage

// Read and process
SET "last_temp" WITH (GET TOPIC "sensors/temperature" AS DOUBLE)

// Check before using (EMPTY check)
IF GET TOPIC "config/threshold" == EMPTY THEN
    SET "threshold" WITH 80
ELSE
    SET "threshold" WITH (GET TOPIC "config/threshold" AS DOUBLE)

// Combine multiple topic values
SET "total" WITH (GET TOPIC "counters/a" AS INT + GET TOPIC "counters/b" AS INT)

GET JSON

Extracts fields from JSON payloads with support for nested paths and arrays.

Basic Syntax

GET JSON "field_name" IN PAYLOAD AS <TYPE>

Supported Types

Type	Use For	Example
`STRING`	Text values	`GET JSON "name" IN PAYLOAD AS STRING`
`DOUBLE`	Decimal numbers	`GET JSON "temperature" IN PAYLOAD AS DOUBLE`
`INT`	Whole numbers	`GET JSON "count" IN PAYLOAD AS INT`
`BOOL`	True/false values	`GET JSON "enabled" IN PAYLOAD AS BOOL`

Nested Path Extraction

Use dot notation to access nested objects and bracket notation for arrays:

// Nested object access
SET "nested" WITH (GET JSON "data.sensors.temperature" IN PAYLOAD AS DOUBLE)
SET "deep" WITH (GET JSON "metadata.config.threshold" IN PAYLOAD AS INT)

// Array access (0-indexed)
SET "first_reading" WITH (GET JSON "readings[0].value" IN PAYLOAD AS DOUBLE)
SET "second_sensor" WITH (GET JSON "data.sensors[1].id" IN PAYLOAD AS STRING)

For a payload like:

{
  "data": {
    "sensors": [
      { "id": "temp001", "value": 23.5 },
      { "id": "temp002", "value": 24.1 }
    ]
  },
  "metadata": { "config": { "threshold": 80 } }
}

Error Handling

If a field doesn’t exist, the result depends on the requested type:

Type	Missing Field Result
`STRING`	`""` (empty string)
`DOUBLE`	`0.0`
`INT`	`0`
`BOOL`	`FALSE`

Use EMPTY check for conditional handling:

SET "value" WITH (GET JSON "optional_field" IN PAYLOAD AS DOUBLE)
IF {value} == EMPTY THEN
    SET "value" WITH 100  // Default fallback

Complete Example

DEFINE ACTION ProcessComplexJSON
ON TOPIC "sensors/data" DO
    // Top-level fields
    SET "sensor_id" WITH (GET JSON "id" IN PAYLOAD AS STRING)
    SET "timestamp" WITH (GET JSON "timestamp" IN PAYLOAD AS STRING)
    
    // Nested fields
    SET "temp" WITH (GET JSON "readings.temperature" IN PAYLOAD AS DOUBLE)
    SET "humidity" WITH (GET JSON "readings.humidity" IN PAYLOAD AS DOUBLE)
    
    // Array access
    SET "latest" WITH (GET JSON "history[0].value" IN PAYLOAD AS DOUBLE)
    
    PUBLISH TOPIC "processed/" + {sensor_id} + "/temp" WITH {temp}
    PUBLISH TOPIC "processed/" + {sensor_id} + "/humidity" WITH {humidity}

Environment Variables & Secrets

Version Requirement: GET ENV, GET SECRET, KEEP ENV, KEEP SECRET, DELETE ENV, and DELETE SECRET are available from Coreflux Broker v1.9.3 and above.

LoT Actions can read, write, and delete environment variables and encrypted secrets at runtime. Environment variables store non-sensitive configuration (hostnames, ports, URLs). Secrets store sensitive credentials encrypted with AES-256-GCM.

Use environment variables for configuration that can safely appear in logs (hostnames, ports, feature flags). Use secrets for anything sensitive (passwords, API keys, tokens).

GET ENV

Reads an environment variable. The broker checks the managed .env file first, then falls back to process, user, and machine environment variables. Returns an empty string if not found.

GET ENV "NAME"

Examples

-- Read connection settings
SET "host" WITH (GET ENV "DB_HOST")
SET "port" WITH (GET ENV "DB_PORT")
SET "apiUrl" WITH (GET ENV "API_BASE_URL")

-- Use in conditional logic
IF GET ENV "FEATURE_FLAG" == "enabled" THEN
    PUBLISH TOPIC "features/new" WITH "active"

KEEP ENV

Persists a new environment variable to the .env file on disk:

KEEP ENV "NAME" WITH <value>

-- Save a timestamp
KEEP ENV "LAST_RUN" WITH TIMESTAMP "UTC"

-- Save a computed value
KEEP ENV "DEVICE_COUNT" WITH {count}

DELETE ENV

Removes an environment variable from the managed .env file:

DELETE ENV "NAME"

DELETE ENV "OLD_CONFIG"

GET SECRET

Reads an encrypted secret from secrets.json, decrypting it at runtime. Returns an empty string if the secret does not exist.

GET SECRET "NAME"

Examples

-- Read credentials
SET "dbPass" WITH (GET SECRET "DB_PASSWORD")
SET "apiKey" WITH (GET SECRET "API_KEY")

-- Use in conditional logic
IF GET SECRET "LICENSE_KEY" == EMPTY THEN
    PUBLISH TOPIC "system/warnings" WITH "No license key configured"

KEEP SECRET

Encrypts and persists a new secret to secrets.json:

KEEP SECRET "NAME" WITH "value"

KEEP SECRET "API_KEY" WITH "sk-abc123"

DELETE SECRET

Removes a secret from secrets.json:

DELETE SECRET "NAME"

DELETE SECRET "OLD_API_KEY"

Complete Example

DEFINE ACTION SecureDataForwarder
ON TOPIC "sensors/+/data" DO
    SET "sensor_id" WITH TOPIC POSITION 2

    -- Read configuration from environment
    SET "apiUrl" WITH (GET ENV "API_BASE_URL")
    SET "region" WITH (GET ENV "DEPLOY_REGION")

    -- Read credentials from secrets
    SET "apiKey" WITH (GET SECRET "API_KEY")

    IF {apiUrl} == EMPTY THEN
        PUBLISH TOPIC "system/errors" WITH "API_BASE_URL not configured"
    ELSE
        -- Forward sensor data with region tag
        PUBLISH TOPIC "forwarded/" + {region} + "/" + {sensor_id} WITH PAYLOAD
        PUBLISH TOPIC "forwarded/" + {sensor_id} + "/timestamp" WITH TIMESTAMP "UTC"

    -- Track last forwarded sensor
    KEEP ENV "LAST_FORWARDED_SENSOR" WITH {sensor_id}

For full details on environment variable resolution order, encryption key management, and deployment configuration, see Environment Variables & Secrets.

Arithmetic Operators

LoT supports standard arithmetic operations that work with numeric types (INT, DOUBLE) and can be combined with GET TOPIC, PAYLOAD, and variables.

Supported Operators

Operator	Symbol	Description
Addition	`+`	Adds two values together
Subtraction	`-`	Subtracts right value from left
Multiplication	`*`	Multiplies two values
Division	`/`	Divides left value by right
Modulo	`%`	Returns remainder of division
Unary Negative	`-`	Negates a value

Syntax

<expression> + <expression>
<expression> - <expression>
<expression> * <expression>
<expression> / <expression>
<expression> % <expression>
-<expression>

Basic Examples

// Simple arithmetic with literals
SET "sum" WITH (10 + 5)
SET "difference" WITH (100 - 25)
SET "product" WITH (6 * 7)
SET "quotient" WITH (100 / 4)
SET "remainder" WITH (17 % 5)
SET "negative" WITH (-42)

Using with Variables

SET "a" WITH 100
SET "b" WITH 25
SET "result" WITH ({a} + {b})
SET "average" WITH (({a} + {b}) / 2)

Using with GET TOPIC

// Add values from two different topics
SET "total" WITH (GET TOPIC "sensors/count/a" AS INT + GET TOPIC "sensors/count/b" AS INT)

// Calculate percentage
SET "percentage" WITH (GET TOPIC "current/value" AS DOUBLE / GET TOPIC "max/value" AS DOUBLE * 100)

// Temperature conversion (Celsius to Fahrenheit)
SET "fahrenheit" WITH (GET TOPIC "sensors/temp/celsius" AS DOUBLE * 9 / 5 + 32)

Using with PAYLOAD

DEFINE ACTION CalculateWithPayload
ON TOPIC "input/value" DO
    SET "doubled" WITH (PAYLOAD AS DOUBLE * 2)
    SET "incremented" WITH (PAYLOAD AS INT + 1)
    PUBLISH TOPIC "output/doubled" WITH {doubled}

Operator Precedence

Operations follow standard mathematical precedence:

Parentheses () - highest priority
Unary negation -
Multiplication *, Division /, Modulo %
Addition +, Subtraction -

// Without parentheses: 2 + 3 * 4 = 14 (multiplication first)
SET "result1" WITH (2 + 3 * 4)

// With parentheses: (2 + 3) * 4 = 20
SET "result2" WITH ((2 + 3) * 4)

Complete Example

DEFINE ACTION TemperatureProcessor
ON TOPIC "sensors/+/temperature" DO
    // POSITION 1="sensors", POSITION 2=sensor_id (wildcard), POSITION 3="temperature"
    SET "sensor_id" WITH TOPIC POSITION 2
    SET "celsius" WITH PAYLOAD AS DOUBLE
    
    // Convert to Fahrenheit
    SET "fahrenheit" WITH ({celsius} * 9 / 5 + 32)
    
    // Calculate deviation from setpoint
    SET "setpoint" WITH (GET TOPIC "config/setpoint" AS DOUBLE)
    SET "deviation" WITH ({celsius} - {setpoint})
    SET "deviation_percent" WITH ({deviation} / {setpoint} * 100)
    
    PUBLISH TOPIC "processed/" + {sensor_id} + "/fahrenheit" WITH {fahrenheit}
    PUBLISH TOPIC "processed/" + {sensor_id} + "/deviation" WITH {deviation_percent}

String Concatenation

The + operator concatenates strings, allowing dynamic construction of topics, messages, and values.

Syntax

<string_expression> + <string_expression>

Basic Examples

SET "greeting" WITH ("Hello, " + "World!")
SET "fullname" WITH ({firstname} + " " + {lastname})

Dynamic Topic Construction

// Build topic paths dynamically
PUBLISH TOPIC "devices/" + {device_id} + "/status" WITH "online"
PUBLISH TOPIC "sensors/" + {sensor_type} + "/" + {location} + "/reading" WITH {value}

// Using with KEEP
KEEP TOPIC "cache/" + {sensor_id} + "/last_reading" WITH PAYLOAD

Combining Strings with Other Types

// String + Number (number is converted to string)
SET "message" WITH ("Temperature is: " + {temp})
PUBLISH TOPIC "alerts/temperature" WITH ("High temp: " + {celsius} + "°C")

Complete Example

DEFINE ACTION DynamicPublisher
ON TOPIC "input/+/+/data" DO
    // POSITION 1="input", POSITION 2=region, POSITION 3=device, POSITION 4="data"
    SET "region" WITH TOPIC POSITION 2
    SET "device" WITH TOPIC POSITION 3
    SET "timestamp" WITH TIMESTAMP "UTC"
    
    // Build dynamic output topic
    SET "output_topic" WITH ("processed/" + {region} + "/" + {device})
    
    // Build descriptive message
    SET "message" WITH ("Device " + {device} + " in " + {region} + " reported at " + {timestamp})
    
    PUBLISH TOPIC {output_topic} WITH {message}

RANDOM Function

Generates random values of various types including numeric ranges and unique identifiers.

Syntax

// Random integer in range (inclusive)
RANDOM BETWEEN <min> AND <max>
RANDOM INT BETWEEN <min> AND <max>

// Random double/float in range
RANDOM DOUBLE BETWEEN <min> AND <max>

// Unique identifiers
RANDOM UUID
RANDOM GUID
RANDOM ULID

Random Integers

// Random integer between 1 and 100
SET "dice_roll" WITH (RANDOM BETWEEN 1 AND 6)
SET "random_percent" WITH (RANDOM INT BETWEEN 0 AND 100)

// Using expressions for range
SET "random_in_range" WITH (RANDOM INT BETWEEN {min_value} AND {max_value})

Random Doubles

// Random double between 0.0 and 1.0
SET "probability" WITH (RANDOM DOUBLE BETWEEN 0.0 AND 1.0)

// Random temperature for simulation
SET "simulated_temp" WITH (RANDOM DOUBLE BETWEEN 18.5 AND 25.5)

Unique Identifiers

// Generate UUID (e.g., "550e8400-e29b-41d4-a716-446655440000")
SET "transaction_id" WITH (RANDOM UUID)

// Generate GUID (same format as UUID)
SET "session_id" WITH (RANDOM GUID)

// Generate ULID (e.g., "01ARZ3NDEKTSV4RRFFQ69G5FAV")
// ULIDs are sortable by time
SET "event_id" WITH (RANDOM ULID)

Use Cases

Simulation
Unique IDs
Load Balancing

Generate test data for sensors:

DEFINE ACTION SimulateSensor
ON EVERY 5 SECONDS DO
    SET "temperature" WITH (RANDOM DOUBLE BETWEEN 20.0 AND 30.0)
    SET "humidity" WITH (RANDOM INT BETWEEN 40 AND 80)
    SET "pressure" WITH (RANDOM DOUBLE BETWEEN 1000.0 AND 1020.0)
    
    PUBLISH TOPIC "simulated/sensor/temperature" WITH {temperature}
    PUBLISH TOPIC "simulated/sensor/humidity" WITH {humidity}
    PUBLISH TOPIC "simulated/sensor/pressure" WITH {pressure}

Generate unique record identifiers:

DEFINE ACTION CreateRecord
ON TOPIC "records/create" DO
    SET "record_id" WITH (RANDOM UUID)
    SET "timestamp" WITH TIMESTAMP "UTC"
    
    PUBLISH TOPIC "records/" + {record_id} WITH PAYLOAD
    PUBLISH TOPIC "records/index/latest" WITH {record_id}

Distribute work across workers:

DEFINE ACTION RandomRouter
ON TOPIC "incoming/requests" DO
    SET "worker" WITH (RANDOM INT BETWEEN 1 AND 4)
    PUBLISH TOPIC "workers/" + {worker} + "/queue" WITH PAYLOAD

String Transformations

LoT provides string transformation operations for filtering, replacing, and parsing text data.

FILTER

Extracts content from a string using a regular expression pattern.

Syntax

FILTER <source> USING REGEX "<pattern>"

Supported Sources

PAYLOAD - The incoming message payload
GET TOPIC "path" - Value from an MQTT topic
{variable} - A previously set variable
"string literal" - A hardcoded string

Examples

// Extract all digits from payload
SET "numbers" WITH FILTER PAYLOAD USING REGEX "[0-9]+"

// Extract from a topic value
SET "extracted" WITH FILTER (GET TOPIC "data/raw") USING REGEX "[a-zA-Z]+"

// Extract email pattern
SET "email" WITH FILTER PAYLOAD USING REGEX "[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}"

// Extract numbers from mixed content
SET "value" WITH FILTER "Temperature: 23.5°C" USING REGEX "[0-9]+\\.?[0-9]*"

Common Regex Patterns

Pattern	Description	Example Match
`[0-9]+`	One or more digits	”123”, “4567”
`[a-zA-Z]+`	One or more letters	”Hello”, “World”
`[0-9]+\\.?[0-9]*`	Decimal numbers	”23.5”, “100”
`[^\\s]+`	Non-whitespace	”word”
`\\w+`	Word characters	”test_123”

REPLACE

Replaces occurrences of a pattern with a new value.

Syntax

REPLACE "<find>" WITH "<replacement>" IN <source>

Examples

// Replace in payload
SET "cleaned" WITH REPLACE "old" WITH "new" IN PAYLOAD

// Replace in topic value
SET "modified" WITH REPLACE "error" WITH "warning" IN (GET TOPIC "logs/latest")

// Replace in variable
SET "formatted" WITH REPLACE " " WITH "_" IN {sensor_name}

// Remove characters by replacing with empty string
SET "digits_only" WITH REPLACE "[^0-9]" WITH "" IN {mixed_content}

Multiple Replacements

SET "step1" WITH REPLACE "," WITH ";" IN PAYLOAD
SET "step2" WITH REPLACE " " WITH "_" IN {step1}
SET "final" WITH REPLACE "'" WITH "" IN {step2}

SPLIT CSV

Parses CSV (Comma-Separated Values) data from the payload into a structured format.

Syntax

SPLIT PAYLOAD CSV [SEPARATOR "<char>"] [HEADERS PRESENT|NONE]

Parameters

Parameter	Description	Default
`SEPARATOR`	The delimiter character	`,` (comma)
`HEADERS PRESENT`	First row contains column names	-
`HEADERS NONE`	No header row	-

Example

DEFINE ACTION ProcessCSVData
ON TOPIC "import/csv" DO
    // Payload: "name,value,unit\nTemperature,23.5,C\nHumidity,65,percent"
    SET "parsed" WITH SPLIT PAYLOAD CSV SEPARATOR "," HEADERS PRESENT
    
    PUBLISH TOPIC "import/parsed" WITH {parsed}

SPLIT CSV currently only supports PAYLOAD as input. Support for GET TOPIC and variables is planned for future releases.

Topic Operations

PUBLISH TOPIC

Sends data to an MQTT topic, broadcasting it to all clients subscribed to that topic. Published messages are delivered immediately and can trigger other actions or routes listening to the topic.

Syntax

PUBLISH TOPIC "<topic_path>" WITH <value>

Value Types

You can publish various value types:

// String
PUBLISH TOPIC "system/status" WITH "online"

// Number
PUBLISH TOPIC "sensors/reading" WITH 23.5

// Variable
PUBLISH TOPIC "output/value" WITH {calculated_result}

// Payload passthrough
PUBLISH TOPIC "forwarded/data" WITH PAYLOAD

// Timestamp
PUBLISH TOPIC "events/timestamp" WITH TIMESTAMP "UTC"

Cannot Use Inline JSON ObjectsPUBLISH TOPIC does not support inline JSON object literals. This will NOT work:

// ❌ WRONG - This does not work
PUBLISH TOPIC "output/data" WITH {"temperature": 23.5, "humidity": 65}

Why? PUBLISH TOPIC is designed for simple scalar values (strings, numbers, timestamps, variables). It doesn’t have the type system or schema validation needed for structured JSON data.Solution: For structured JSON output, use Models with PUBLISH MODEL:

// ✅ CORRECT - Define a model and publish it
DEFINE MODEL SensorData COLLAPSED
    ADD DOUBLE "temperature"
    ADD DOUBLE "humidity"

DEFINE ACTION PublishStructuredData
ON TOPIC "sensors/input" DO
    PUBLISH MODEL SensorData TO "output/data" WITH
        temperature = 23.5
        humidity = 65

Models provide type safety, schema validation, and proper JSON serialization. See Publishing Models for complete guidance.

Dynamic Topics

Build topic paths dynamically using string concatenation:

PUBLISH TOPIC "devices/" + {device_id} + "/status" WITH "active"
PUBLISH TOPIC "sensors/" + {sensor_type} + "/" + {location} + "/reading" WITH {value}

Multiple Publishes

Actions can publish to multiple topics:

DEFINE ACTION MultiPublish
ON TOPIC "input/data" DO
    PUBLISH TOPIC "output/raw" WITH PAYLOAD
    PUBLISH TOPIC "output/timestamp" WITH TIMESTAMP "UTC"
    PUBLISH TOPIC "output/source" WITH "sensor_001"

Triggering Other Actions: When you publish to a topic, any action with ON TOPIC matching that path will execute. This enables chaining actions together for complex workflows.

KEEP TOPIC

Stores a value in an internal broker topic that persists between action executions. Think of KEEP as creating a persistent variable - unlike SET (which exists only during one execution), KEEP values survive and can be read later.

KEEP vs PUBLISH: Both write to topics, but PUBLISH sends to the MQTT bus (external clients can subscribe). KEEP stores internally for your actions to read later - like a private scratchpad that doesn’t broadcast to subscribers.

When to Use KEEP

Use Case	Example
Cache last known value	`KEEP TOPIC "cache/last_temp" WITH {temperature}`
Store state between runs	`KEEP TOPIC "state/counter" WITH {count}`
Save intermediate calculations	`KEEP TOPIC "calc/running_avg" WITH {avg}`

Syntax

KEEP TOPIC "<topic_path>" WITH <value>

Example: Persistent Counter

DEFINE ACTION PersistentCounter
ON EVERY 10 SECONDS DO
    SET "current" WITH (GET TOPIC "state/counter")
    IF {current} == EMPTY THEN
        KEEP TOPIC "state/counter" WITH 1
    ELSE
        KEEP TOPIC "state/counter" WITH ({current} + 1)
    
    PUBLISH TOPIC "stats/count" WITH (GET TOPIC "state/counter")

KEEP vs PUBLISH Comparison

Operation	Network Traffic	External Access	Use Case
`PUBLISH`	Yes	Yes	Data for external systems
`KEEP`	No	No	Internal caching, state storage

Control Flow

IF … THEN … ELSE

Execute different logic based on conditions.

IF <condition> THEN
    <statements>
ELSE
    <statements>

Comparison Operators

Operator	Description	Example
`EQUALS` or `==`	Equality	`IF {value} EQUALS 100`
`!=`	Not equal	`IF {status} != "error"`
`>`	Greater than	`IF {temp} > 80`
`<`	Less than	`IF {count} < 10`
`>=`	Greater or equal	`IF {level} >= 50`
`<=`	Less or equal	`IF {pressure} <= 100`

Basic Conditionals

IF {temperature} > 80 THEN
    PUBLISH TOPIC "alerts/high_temp" WITH {temperature}
ELSE
    PUBLISH TOPIC "status/normal" WITH "OK"

Nested Conditionals

For multiple conditions, nest IF statements:

IF {temperature} > 90 THEN
    PUBLISH TOPIC "alerts/critical" WITH "CRITICAL: " + {temperature}
ELSE
    IF {temperature} > 80 THEN
        PUBLISH TOPIC "alerts/warning" WITH "WARNING: " + {temperature}
    ELSE
        IF {temperature} > 70 THEN
            PUBLISH TOPIC "alerts/info" WITH "Elevated: " + {temperature}
        ELSE
            PUBLISH TOPIC "status/normal" WITH "OK"

Checking for Empty Topics

Use EMPTY to check if a topic has no value:

IF GET TOPIC "data/value" == EMPTY THEN
    PUBLISH TOPIC "data/value" WITH 0
ELSE
    PUBLISH TOPIC "data/value" WITH (GET TOPIC "data/value" + 1)

Combining with Expressions

// Check calculated values
SET "average" WITH (({temp1} + {temp2}) / 2)
IF {average} > {threshold} THEN
    PUBLISH TOPIC "alerts/average_high" WITH {average}

// String comparisons
IF {status} EQUALS "active" THEN
    PUBLISH TOPIC "devices/active_count" WITH (GET TOPIC "devices/active_count" + 1)

Timestamps

Add timestamps to your messages for tracking, logging, and debugging. LoT supports two timestamp formats.

Supported Formats

Format	Output	Use Case
`TIMESTAMP "UTC"`	`2025-10-25T14:30:15Z`	Human-readable logs, APIs, ISO 8601 compliance
`TIMESTAMP "UNIX"`	`1729864215`	Databases, calculations, cross-platform compatibility

Syntax

TIMESTAMP "UTC"   // ISO 8601 format
TIMESTAMP "UNIX"  // Unix epoch (seconds since 1970-01-01)

Using with PUBLISH and KEEP

// Add timestamp to published messages
PUBLISH TOPIC "events/created" WITH TIMESTAMP "UTC"
PUBLISH TOPIC "sensors/reading/time" WITH TIMESTAMP "UNIX"

// Store timestamp for later reference
KEEP TOPIC "cache/last_update" WITH TIMESTAMP "UNIX"

Using with SET

// Store timestamp in a variable
SET "event_time" WITH TIMESTAMP "UTC"
SET "process_start" WITH TIMESTAMP "UNIX"

// Use in string concatenation
SET "log_message" WITH ("Event occurred at " + TIMESTAMP "UTC")
PUBLISH TOPIC "logs/events" WITH {log_message}

Complete Example

DEFINE ACTION AuditLogger
ON TOPIC "actions/+/executed" DO
    SET "action_name" WITH TOPIC POSITION 2
    SET "utc_time" WITH TIMESTAMP "UTC"
    SET "unix_time" WITH TIMESTAMP "UNIX"
    
    // Human-readable log
    PUBLISH TOPIC "audit/" + {action_name} + "/log" 
        WITH ("Action " + {action_name} + " executed at " + {utc_time})
    
    // Machine-readable timestamp for database storage
    PUBLISH TOPIC "audit/" + {action_name} + "/timestamp" WITH {unix_time}

Callable Actions

LoT supports callable actions - actions that can be invoked by other actions with input parameters and return values. This enables modular, reusable logic blocks.

Defining a Callable Action

DEFINE ACTION ActionName
INPUT param1 AS <TYPE>
INPUT param2 AS <TYPE>
DO
    <action_logic>
RETURN
    OUTPUT variable1
    OUTPUT variable2

Supported Input Types

STRING
INT
DOUBLE
BOOL
JSON

Calling an Action

CALL ACTION ActionName
    WITH param1 = <value>, param2 = <value>
    RETURN var1, var2

Basic Example

Define
Call

Create a reusable calculation:

DEFINE ACTION CalculateAverage
INPUT value1 AS DOUBLE
INPUT value2 AS DOUBLE
DO
    SET "sum" WITH ({value1} + {value2})
    SET "avg" WITH ({sum} / 2)
RETURN
    OUTPUT avg

Use the callable action:

DEFINE ACTION ProcessSensorPair
ON TOPIC "sensors/pair/data" DO
    SET "temp1" WITH (GET JSON "sensor1" IN PAYLOAD AS DOUBLE)
    SET "temp2" WITH (GET JSON "sensor2" IN PAYLOAD AS DOUBLE)
    
    CALL ACTION CalculateAverage
        WITH value1 = {temp1}, value2 = {temp2}
        RETURN result
    
    PUBLISH TOPIC "sensors/average" WITH {result}

Multiple Outputs

DEFINE ACTION AnalyzeValue
INPUT value AS DOUBLE
INPUT threshold AS DOUBLE
DO
    SET "is_above" WITH ({value} > {threshold})
    SET "difference" WITH ({value} - {threshold})
    SET "percentage" WITH ({value} / {threshold} * 100)
RETURN
    OUTPUT is_above
    OUTPUT difference
    OUTPUT percentage

DEFINE ACTION CheckThreshold
ON TOPIC "sensors/+/value" DO
    SET "sensor_id" WITH TOPIC POSITION 2
    SET "current_value" WITH PAYLOAD AS DOUBLE
    SET "threshold" WITH (GET TOPIC "config/threshold" AS DOUBLE)
    
    CALL ACTION AnalyzeValue
        WITH value = {current_value}, threshold = {threshold}
        RETURN above, diff, pct
    
    PUBLISH TOPIC "analysis/" + {sensor_id} + "/above_threshold" WITH {above}
    PUBLISH TOPIC "analysis/" + {sensor_id} + "/difference" WITH {diff}
    PUBLISH TOPIC "analysis/" + {sensor_id} + "/percentage" WITH {pct}

Utility Action Library

Build reusable utility actions for common operations:

// Reusable temperature conversion
DEFINE ACTION ConvertCelsiusToFahrenheit
INPUT celsius AS DOUBLE
DO
    SET "fahrenheit" WITH ({celsius} * 9 / 5 + 32)
RETURN
    OUTPUT fahrenheit

// Reusable percentage calculation
DEFINE ACTION CalculatePercentage
INPUT value AS DOUBLE
INPUT total AS DOUBLE
DO
    SET "pct" WITH ({value} / {total} * 100)
RETURN
    OUTPUT pct

// Main processing action using utilities
DEFINE ACTION ProcessTemperatureReading
ON TOPIC "sensors/+/celsius" DO
    SET "sensor_id" WITH TOPIC POSITION 2
    SET "temp_c" WITH PAYLOAD AS DOUBLE
    SET "max_temp" WITH (GET TOPIC "config/max_temperature" AS DOUBLE)
    
    CALL ACTION ConvertCelsiusToFahrenheit
        WITH celsius = {temp_c}
        RETURN temp_f
    
    CALL ACTION CalculatePercentage
        WITH value = {temp_c}, total = {max_temp}
        RETURN capacity
    
    PUBLISH TOPIC "sensors/" + {sensor_id} + "/fahrenheit" WITH {temp_f}
    PUBLISH TOPIC "sensors/" + {sensor_id} + "/capacity_percent" WITH {capacity}

TRIGGER Statement

The TRIGGER statement allows actions to programmatically trigger route events or mappings. This is useful for initiating data pipeline operations or triggering external integrations.

Syntax

TRIGGER "<route_name>" TO <data> FOR <mapping_name>

Components

Component	Description
`route_name`	The name of the route to trigger
`data`	The data to send (PAYLOAD, variable, or expression)
`mapping_name`	The specific mapping or event within the route

Basic Examples

// Trigger a route with payload data
TRIGGER "EmailRoute" TO PAYLOAD FOR "SendAlert"

// Trigger with a variable
SET "message" WITH ("Alert: " + {sensor_id} + " exceeded threshold")
TRIGGER "NotificationRoute" TO {message} FOR "EmailMapping"

// Trigger with a literal value
TRIGGER "DatabaseRoute" TO "backup_requested" FOR "BackupEvent"

Complete Example

DEFINE ACTION AlertHandler
ON TOPIC "alerts/+/critical" DO
    SET "alert_type" WITH TOPIC POSITION 2
    SET "alert_data" WITH PAYLOAD
    SET "timestamp" WITH TIMESTAMP "UTC"
    
    // Build alert message
    SET "message" WITH ("Critical alert [" + {alert_type} + "] at " + {timestamp})
    
    // Trigger email notification route
    TRIGGER "EmailNotification" TO {message} FOR "CriticalAlertEmail"
    
    // Also trigger database storage
    TRIGGER "DatabaseStorage" TO {alert_data} FOR "StoreAlert"
    
    PUBLISH TOPIC "alerts/processed" WITH {alert_type}

Use Cases

Email Notifications
Database Operations
External APIs

DEFINE ACTION SendDailyReport
ON EVERY 24 HOURS DO
    SET "report_data" WITH (GET TOPIC "reports/daily/summary")
    TRIGGER "EmailRoute" TO {report_data} FOR "DailyReportEmail"

DEFINE ACTION ArchiveOldData
ON EVERY 1 HOUR DO
    SET "cutoff_time" WITH TIMESTAMP "UNIX"
    TRIGGER "DatabaseRoute" TO {cutoff_time} FOR "ArchiveEvent"

DEFINE ACTION NotifyExternalSystem
ON TOPIC "events/important" DO
    TRIGGER "RestAPIRoute" TO PAYLOAD FOR "WebhookEvent"

CALL MCP

CALL MCP allows actions to invoke tools provided by MCP (Model Context Protocol) routes. This enables integration with AI assistants and external tools.

Prerequisites

An MCP route must be defined and active
The route must expose the tool you want to call

Syntax

CALL MCP "RouteName.tool_name"
    WITH (arg1 = <value1>, arg2 = <value2>, ...)
    RETURN AS {variable_name}

Example MCP Route

DEFINE ROUTE FileSystem WITH TYPE MCP
    ADD MCP_CONFIG
        WITH SERVER_COMMAND "npx"
        WITH SERVER_ARGUMENTS "-y, @modelcontextprotocol/server-filesystem, /data"
        WITH CLIENT_NAME "CorefluxBroker"

Basic Examples

// Call an MCP tool with arguments
CALL MCP "FileSystem.read_file"
    WITH (path = "/data/config.json")
    RETURN AS {file_content}

// Call without arguments
CALL MCP "MyMCPRoute.list_resources"
    RETURN AS {resources}

// Call with multiple arguments
CALL MCP "MyMCPRoute.echo"
    WITH (message = "Hello", count = 3)
    RETURN AS {response}

Complete Example

DEFINE ACTION ReadDeviceConfig
ON TOPIC "devices/+/request_config" DO
    SET "device_id" WITH TOPIC POSITION 2
    SET "config_path" WITH "/configs/" + {device_id} + ".json"
    
    CALL MCP "FileSystem.read_file"
        WITH (path = {config_path})
        RETURN AS {config_data}
    
    IF {config_data} == EMPTY THEN
        PUBLISH TOPIC "devices/" + {device_id} + "/config/error" WITH "Config not found"
    ELSE
        PUBLISH TOPIC "devices/" + {device_id} + "/config" WITH {config_data}

Complete Example

This example combines the key operations covered on this page—SET, GET TOPIC, GET JSON, arithmetic, IF/THEN, PUBLISH, and KEEP—in a realistic sensor processing action.

DEFINE ACTION ComprehensiveProcessor
ON TOPIC "sensors/+/data" DO
    // Extract sensor ID from topic path (POSITION 2 = the wildcard)
    SET "sensor_id" WITH TOPIC POSITION 2
    
    // Parse JSON payload
    SET "temperature" WITH (GET JSON "temp" IN PAYLOAD AS DOUBLE)
    
    // Convert to Fahrenheit
    SET "fahrenheit" WITH ({temperature} * 9 / 5 + 32)
    
    // Conditional alerting
    IF {temperature} > 80 THEN
        PUBLISH TOPIC "alerts/" + {sensor_id} WITH "High temp: " + {temperature}
    
    // Publish processed data
    PUBLISH TOPIC "processed/" + {sensor_id} + "/temperature" WITH {temperature}
    PUBLISH TOPIC "processed/" + {sensor_id} + "/fahrenheit" WITH {fahrenheit}
    PUBLISH TOPIC "processed/" + {sensor_id} + "/timestamp" WITH TIMESTAMP "UTC"
    
    // Cache for later reference
    KEEP TOPIC "cache/" + {sensor_id} + "/last_reading" WITH PAYLOAD

What This Action Does

Triggers on any message to sensors/+/data topics
Extracts the sensor ID from the topic path
Parses the JSON payload to get temperature value
Converts Celsius to Fahrenheit using arithmetic
Alerts if temperature exceeds threshold
Publishes formatted data to processed topics
Caches the raw payload for future reference

Quick Reference

Operation	Category	Description
`+, -, *, /, %`	Arithmetic	Mathematical operations
`+ (strings)`	String	Concatenation
`RANDOM BETWEEN`	Generation	Random numbers
`RANDOM UUID/GUID/ULID`	Generation	Unique identifiers
`FILTER ... USING REGEX`	Transformation	Extract with regex
`REPLACE ... WITH ... IN`	Transformation	String replacement
`SPLIT ... CSV`	Transformation	Parse CSV data
`GET ENV`	Configuration	Read environment variable
`GET SECRET`	Configuration	Read encrypted secret
`KEEP ENV`	Configuration	Persist environment variable
`KEEP SECRET`	Configuration	Persist encrypted secret
`DELETE ENV`	Configuration	Remove environment variable
`DELETE SECRET`	Configuration	Remove secret
`CALL PYTHON`	Integration	Execute Python functions
`CALL MCP`	Integration	Invoke MCP tools
`CALL ACTION`	Control	Call other actions
`TRIGGER`	Control	Trigger route events

Quick Start

MQTT Broker

AI

LoT Language

LoT Notebooks

Actions

Routes

Models

Rules

​Overview

​In This Page

​Variable Operations

​SET

​GET TOPIC

​Default Values for Missing Topics

​Type Casting

​Wildcard Inheritance

​Example Usage

​GET JSON

​Basic Syntax

​Supported Types

​Nested Path Extraction

​Error Handling

​Complete Example

​Environment Variables & Secrets

​GET ENV

​Examples

​KEEP ENV

​DELETE ENV

​GET SECRET

​Examples

​KEEP SECRET

​DELETE SECRET

​Complete Example

​Arithmetic Operators

​Supported Operators

​Syntax

​Basic Examples

​Using with Variables

​Using with GET TOPIC

​Using with PAYLOAD

​Operator Precedence

​Complete Example

​String Concatenation

​Syntax

​Basic Examples

​Dynamic Topic Construction

​Combining Strings with Other Types

​Complete Example

​RANDOM Function

​Syntax

​Random Integers

​Random Doubles

​Unique Identifiers

​Use Cases

​String Transformations

​FILTER

​Syntax

​Supported Sources

​Examples

​Common Regex Patterns

​REPLACE

​Syntax

​Examples

​Multiple Replacements

​SPLIT CSV

​Syntax

​Parameters

​Example

​Topic Operations

​PUBLISH TOPIC

​Syntax

​Value Types

​Dynamic Topics

​Multiple Publishes

​KEEP TOPIC

​When to Use KEEP

​Syntax

​Example: Persistent Counter

​KEEP vs PUBLISH Comparison

Overview

In This Page

Variable Operations

SET

GET TOPIC

Default Values for Missing Topics

Type Casting

Wildcard Inheritance

Example Usage

GET JSON

Basic Syntax

Supported Types

Nested Path Extraction

Error Handling

Complete Example

Environment Variables & Secrets

GET ENV

Examples

KEEP ENV

DELETE ENV

GET SECRET

Examples

KEEP SECRET

DELETE SECRET

Complete Example

Arithmetic Operators

Supported Operators

Syntax

Basic Examples

Using with Variables

Using with GET TOPIC

Using with PAYLOAD

Operator Precedence

Complete Example

String Concatenation

Syntax

Basic Examples

Dynamic Topic Construction

Combining Strings with Other Types

Complete Example

RANDOM Function

Syntax

Random Integers

Random Doubles

Unique Identifiers

Use Cases

String Transformations

FILTER

Syntax

Supported Sources

Examples

Common Regex Patterns

REPLACE

Syntax

Examples

Multiple Replacements

SPLIT CSV

Syntax

Parameters

Example

Topic Operations

PUBLISH TOPIC

Syntax

Value Types

Dynamic Topics

Multiple Publishes

KEEP TOPIC

When to Use KEEP

Syntax

Example: Persistent Counter

KEEP vs PUBLISH Comparison

Control Flow

IF … THEN … ELSE

Comparison Operators

Basic Conditionals

Nested Conditionals

Checking for Empty Topics

Combining with Expressions

Timestamps

Supported Formats