Actions Overview

Overview

Actions are executable logic blocks that run directly inside the Coreflux MQTT broker. They react to events—incoming messages, time intervals, or manual triggers—and perform operations like data transformation, routing, and alerting.

Think of Actions like a mail sorting machine. When a letter (message) arrives, the machine reads the address (topic), decides where it should go (routing logic), and sends it to the right mailbox (publishes to a topic). All of this happens instantly, without needing a person to manually sort each letter.

Action Capabilities

Capability	Description
Event-Driven Execution	Trigger on MQTT messages with wildcard support for flexible topic matching
Time-Based Scheduling	Execute at regular intervals—seconds to weeks—for monitoring and reporting
Data Transformation	Extract JSON fields, perform calculations, and restructure payloads inline
Conditional Logic	Route data, set thresholds, and make decisions with IF/THEN/ELSE

When to Use Each Pattern

Pattern	Use When
Time-Based (`ON EVERY`)	Periodic tasks: heartbeats, polling, scheduled reports
Topic-Based (`ON TOPIC`)	Event-driven processing: sensor data, commands, alerts
Callable (no trigger)	Reusable utilities invoked by other actions or commands
Python-Extended (`CALL PYTHON`)	Complex calculations, external libraries, data validation

In This Section

Syntax Reference

Start here to learn the DEFINE ACTION structure and build your first action.

Action Triggers

Choose the right trigger type for your use case: time-based, topic-based, or callable.

Operations

Master SET, GET, PUBLISH, conditionals, and all available LoT operations.

Python Integration

Extend actions with Python when you need complex logic or external libraries.

Action Types

LoT supports three action patterns:

Time-Based Actions

Execute at regular intervals using ON EVERY:

DEFINE ACTION SystemHeartbeat
ON EVERY 5 SECONDS DO
    PUBLISH TOPIC "system/status" WITH "online"
    PUBLISH TOPIC "system/timestamp" WITH TIMESTAMP "UTC"

Key characteristics:

Runs on a schedule regardless of incoming data
Useful for monitoring, polling, and periodic reports
Supported units: MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS, WEEKS, MONTHS, YEARS
Calendar-aware scheduling: ON EVERY WEEKDAY AT "HH:MM:SS" and ON EVERY WEEKEND AT "HH:MM:SS"

Topic-Based Actions

Execute when MQTT messages arrive using ON TOPIC:

DEFINE ACTION TemperatureProcessor
ON TOPIC "sensors/+/temperature" DO
    SET "sensor_id" WITH TOPIC POSITION 2
    SET "temp" WITH (GET JSON "value" IN PAYLOAD AS DOUBLE)
    
    IF {temp} > 80 THEN
        PUBLISH TOPIC "alerts/high_temp/" + {sensor_id} WITH PAYLOAD
    ELSE
        PUBLISH TOPIC "processed/" + {sensor_id} WITH PAYLOAD

Key characteristics:

Triggers instantly when matching messages arrive
Supports + (single-level) and # (multi-level) MQTT wildcards
Access payload data via PAYLOAD and GET JSON
Extract topic segments with TOPIC POSITION

Callable Actions

Actions without triggers are invoked manually or by other actions:

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

Key characteristics:

No automatic trigger—must be invoked explicitly
Can accept input parameters and return values
Invoked via $SYS/Coreflux/Command or CALL ACTION
Ideal for reusable utility functions

Learn more about invoking callable actions in Action Triggers.

Core Operations

Actions use these operations to process data:

SET, GET TOPIC, GET JSON, PUBLISH, and IF/THEN are the core operations for working with data in actions. See the Operations Reference for complete details.

Variables
Topic Operations
JSON Extraction
Conditionals

Create and use variables within an action:

SET "sensor_id" WITH TOPIC POSITION 2
SET "temperature" WITH (GET TOPIC "sensors/temp" AS DOUBLE)
SET "humidity" WITH (GET JSON "humidity" IN PAYLOAD AS DOUBLE)

PUBLISH TOPIC "output/" + {sensor_id} WITH {temperature}

Read from and write to MQTT topics:

// Read from a topic
SET "last_value" WITH (GET TOPIC "sensors/temperature" AS DOUBLE)

// Publish to subscribers
PUBLISH TOPIC "output/data" WITH {temperature}

// Store internally (not broadcast)
KEEP TOPIC "cache/last_reading" WITH PAYLOAD

Parse JSON payloads and extract fields:

// For payload: {"temp": 23.5, "humidity": 65}
SET "temp" WITH (GET JSON "temp" IN PAYLOAD AS DOUBLE)
SET "humidity" WITH (GET JSON "humidity" IN PAYLOAD AS INT)

// Nested paths
SET "city" WITH (GET JSON "location.city" IN PAYLOAD AS STRING)

Implement branching logic:

IF {temperature} > 80 THEN
    PUBLISH TOPIC "alerts/high" WITH {temperature}
ELSE IF {temperature} < 20 THEN
    PUBLISH TOPIC "alerts/low" WITH {temperature}
ELSE
    PUBLISH TOPIC "status/normal" WITH "OK"

Example: Complete Sensor Processor

This example combines topic matching, JSON extraction, conditional routing, and state updates in a realistic sensor processing action.

DEFINE ACTION SensorDataProcessor
ON TOPIC "sensors/+/raw" DO
    // Extract sensor ID from topic path
    // Topic: sensors/temp001/raw → POSITION 2 = "temp001"
    SET "sensor_id" WITH TOPIC POSITION 2
    
    // Parse JSON payload
    SET "temp_value" WITH (GET JSON "temperature" IN PAYLOAD AS DOUBLE)
    SET "humidity" WITH (GET JSON "humidity" IN PAYLOAD AS DOUBLE)
    
    // Conditional routing based on thresholds
    IF {temp_value} > 80 THEN
        PUBLISH TOPIC "alerts/high_temp/" + {sensor_id} WITH PAYLOAD
    ELSE IF {temp_value} < 10 THEN
        PUBLISH TOPIC "alerts/low_temp/" + {sensor_id} WITH PAYLOAD
    ELSE
        PUBLISH TOPIC "sensors/normal/" + {sensor_id} WITH PAYLOAD
    
    // Always update the processed output
    PUBLISH TOPIC "sensors/processed/" + {sensor_id} WITH PAYLOAD
    
    // Record the last update timestamp
    PUBLISH TOPIC "sensors/" + {sensor_id} + "/last_update" WITH TIMESTAMP "UTC"

What this does:

Triggers when any message arrives at sensors/*/raw
Extracts the sensor ID from the topic path
Parses temperature and humidity from the JSON payload
Routes alerts based on temperature thresholds
Publishes processed data and timestamps

Actions vs. Models

Actions and Models serve different purposes in LoT:

Aspect	Actions	Models
Purpose	Process and route data	Structure and format data
Execution	Event-driven logic	Schema definition
Output	Any MQTT operation	Structured JSON
Complexity	Conditionals, loops, Python	Field mapping, calculations
Best For	Routing, alerting, transformation	Data aggregation, formatting

Use them together: Actions process and route data, then publish it using a Model to produce structured JSON output.

Quick Start

Coreflux HUB

MQTT Broker

AI

LoT Language

VS Code - LoT Notebooks

Actions

Routes

Models

Rules

Overview

Action Capabilities

When to Use Each Pattern

In This Section

Syntax Reference

Action Triggers

Operations

Python Integration

Action Types

Time-Based Actions

Topic-Based Actions

Callable Actions

Core Operations

Example: Complete Sensor Processor

Actions vs. Models

Next Steps

Syntax Reference

Action Triggers

Quick Start

Coreflux HUB

MQTT Broker

AI

LoT Language

VS Code - LoT Notebooks

Actions

Routes

Models

Rules

​Overview

​Action Capabilities

​When to Use Each Pattern

​In This Section

Syntax Reference

Action Triggers

Operations

Python Integration

​Action Types

​Time-Based Actions

​Topic-Based Actions

​Callable Actions

​Core Operations

​Example: Complete Sensor Processor

​Actions vs. Models

​Next Steps

Syntax Reference

Action Triggers

Overview

Action Capabilities

When to Use Each Pattern

In This Section

Action Types

Time-Based Actions

Topic-Based Actions

Callable Actions

Core Operations

Example: Complete Sensor Processor

Actions vs. Models

Next Steps