MODEL Publishing

When to Use Action Models

Action models (COLLAPSED models) are templates for structured data that you publish explicitly from within actions. They’re ideal when you need:

Control over timing - Publish only when specific conditions are met
Dynamic routing - Send data to different topics based on logic
Data transformation - Process and enrich data before structuring

Unlike basic models that trigger automatically on topic changes, action models let you decide exactly when and where structured data is created. With WITH FORMAT PROTOBUF, that output is binary protobuf bytes rather than JSON — see Protobuf Models.

Example Use Case

Imagine you receive raw alarm data and need to:

Extract relevant fields from JSON
Add a unique ID and timestamp
Route critical alarms to a special topic
Publish a structured record

DEFINE MODEL AlarmRecord COLLAPSED
    ADD STRING "alarm_id"
    ADD STRING "equipment_id"
    ADD STRING "severity"
    ADD STRING "timestamp"

DEFINE ACTION ProcessAlarm
ON TOPIC "alarms/+/input" DO
    SET "equipment_id" WITH TOPIC POSITION 2
    SET "severity" WITH (GET JSON "severity" IN PAYLOAD AS STRING)
    
    PUBLISH MODEL AlarmRecord TO "alarms/structured/" + {equipment_id} WITH
        alarm_id = (RANDOM UUID)
        equipment_id = {equipment_id}
        severity = {severity}
        timestamp = TIMESTAMP "UTC"
    
    IF {severity} EQUALS "CRITICAL" THEN
        PUBLISH TOPIC "alerts/critical/" + {equipment_id} WITH PAYLOAD

Step-by-Step Guide

Define a COLLAPSED Model

Create a template with all required fields. The COLLAPSED keyword indicates this model has no automatic trigger:

DEFINE MODEL SensorRecord COLLAPSED
    ADD STRING "sensor_id"
    ADD STRING "sensor_type"
    ADD DOUBLE "temperature"
    ADD DOUBLE "humidity"
    ADD STRING "quality_status"
    ADD STRING "timestamp"

Create the Processing Action

Define an action that extracts data and publishes the model:

DEFINE ACTION ProcessSensorJSON
ON TOPIC "sensors/+/json_data" DO
    SET "sensor_id" WITH TOPIC POSITION 2
    
    // Extract from JSON payload
    SET "temp" WITH (GET JSON "temperature" IN PAYLOAD AS DOUBLE)
    SET "hum" WITH (GET JSON "humidity" IN PAYLOAD AS DOUBLE)
    SET "type" WITH (GET JSON "type" IN PAYLOAD AS STRING)
    SET "quality" WITH (GET JSON "quality" IN PAYLOAD AS STRING)
    
    // Publish structured record
    PUBLISH MODEL SensorRecord TO "sensors/processed/" + {sensor_id} WITH
        sensor_id = {sensor_id}
        sensor_type = {type}
        temperature = {temp}
        humidity = {hum}
        quality_status = {quality}
        timestamp = TIMESTAMP "UTC"

Test the Flow

Publish to: sensors/temp001/json_data

{
  "temperature": 23.5,
  "humidity": 65,
  "type": "environmental",
  "quality": "GOOD"
}

Result at: sensors/processed/temp001

{
  "sensor_id": "temp001",
  "sensor_type": "environmental",
  "temperature": 23.5,
  "humidity": 65,
  "quality_status": "GOOD",
  "timestamp": "2025-10-25T14:30:15Z"
}

Use Case Examples

Alarm Processing
Quality Reports
Event Logging

Route alarms by severity while creating structured records:

DEFINE MODEL AlarmRecord COLLAPSED
    ADD STRING "alarm_id"
    ADD STRING "equipment_id"
    ADD STRING "alarm_type"
    ADD DOUBLE "trigger_value"
    ADD STRING "severity"
    ADD STRING "timestamp"

DEFINE ACTION ProcessAlarmJSON
ON TOPIC "alarms/+/input" DO
    SET "equipment_id" WITH TOPIC POSITION 2
    
    PUBLISH MODEL AlarmRecord TO "alarms/structured/" + {equipment_id} WITH
        alarm_id = (RANDOM UUID)
        equipment_id = {equipment_id}
        alarm_type = (GET JSON "type" IN PAYLOAD AS STRING)
        trigger_value = (GET JSON "value" IN PAYLOAD AS DOUBLE)
        severity = (GET JSON "severity" IN PAYLOAD AS STRING)
        timestamp = TIMESTAMP "UTC"
    
    // Route critical alarms
    SET "sev" WITH (GET JSON "severity" IN PAYLOAD AS STRING)
    IF {sev} EQUALS "CRITICAL" THEN
        PUBLISH TOPIC "alerts/critical/" + {equipment_id} WITH PAYLOAD

Calculate metrics and generate quality reports:

DEFINE MODEL QualityReport COLLAPSED
    ADD STRING "report_id"
    ADD STRING "station_id"
    ADD INT "items_tested"
    ADD INT "items_passed"
    ADD DOUBLE "pass_rate"
    ADD STRING "timestamp"

DEFINE ACTION GenerateQualityReport
ON TOPIC "quality/+/results" DO
    SET "station" WITH TOPIC POSITION 2
    SET "tested" WITH (GET JSON "total" IN PAYLOAD AS INT)
    SET "passed" WITH (GET JSON "passed" IN PAYLOAD AS INT)
    SET "rate" WITH ({passed} AS DOUBLE / {tested} AS DOUBLE * 100)
    
    PUBLISH MODEL QualityReport TO "reports/quality/" + {station} WITH
        report_id = (RANDOM UUID)
        station_id = {station}
        items_tested = {tested}
        items_passed = {passed}
        pass_rate = {rate}
        timestamp = TIMESTAMP "UTC"
    
    IF {rate} < 90 THEN
        PUBLISH TOPIC "alerts/quality/low_rate" WITH {station} + ": " + {rate} + "%"

Create structured event logs with context:

DEFINE MODEL EventLog COLLAPSED
    ADD STRING "event_id"
    ADD STRING "event_type"
    ADD STRING "source"
    ADD STRING "description"
    ADD STRING "severity"
    ADD STRING "timestamp"

DEFINE ACTION LogEvent
ON TOPIC "events/+/occurred" DO
    SET "event_type" WITH TOPIC POSITION 2
    
    PUBLISH MODEL EventLog TO "logs/events/" + {event_type} WITH
        event_id = (RANDOM UUID)
        event_type = {event_type}
        source = (GET JSON "source" IN PAYLOAD AS STRING)
        description = (GET JSON "message" IN PAYLOAD AS STRING)
        severity = (GET JSON "severity" IN PAYLOAD AS STRING)
        timestamp = TIMESTAMP "UTC"

JSON vs protobuf on the destination topic

PUBLISH MODEL writes whatever format the model declares:

Model format	Payload on `TO "…"`
`WITH FORMAT JSON` (default)	JSON string
`WITH FORMAT PROTOBUF` / `PROTO`	Raw protobuf bytes
`WITH FORMAT BOTH`	JSON on the main topic + binary on a `/protobuf` sidecar topic

Trigger-based models (WITH TOPIC + AS TRIGGER) still auto-publish JSON. Use a COLLAPSED model and PUBLISH MODEL when a gateway must receive protobuf. Nested messages you fill on publish use ADD MODEL Child "field", not indented ADD OBJECT.

DEFINE MODEL TelemetryMessage WITH FORMAT PROTOBUF COLLAPSED
    ADD INT "message_type" PROTO_TAG 1
    ADD DOUBLE "temperature" PROTO_TAG 2
    ADD DOUBLE "pressure" PROTO_TAG 3

The Action copies JSON tags into that envelope and publishes bytes:

DEFINE ACTION PublishTelemetry
ON TOPIC "plant/+/tags" DO
    SET "temp" WITH GET JSON "Temperature" IN PAYLOAD AS DOUBLE
    SET "pressure" WITH GET JSON "Pressure" IN PAYLOAD AS DOUBLE
    PUBLISH MODEL TelemetryMessage TO "plant/gateway/telemetry" WITH
        message_type = 2
        temperature = {temp}
        pressure = {pressure}

Full type mapping, PROTO_TAG, and GET PROTO are on Protobuf Models.

Best Practices

Always Use Type Casting with GET JSON

Explicit type casting prevents runtime errors:

// Good
SET "temp" WITH (GET JSON "temperature" IN PAYLOAD AS DOUBLE)

// Risky
SET "temp" WITH (GET JSON "temperature" IN PAYLOAD)

Validate Extracted Data

Check data before publishing:

SET "value" WITH (GET JSON "value" IN PAYLOAD AS DOUBLE)

IF {value} >= 0 AND {value} <= 100 THEN
    PUBLISH MODEL ValidData TO "output" WITH value = {value}
ELSE
    PUBLISH TOPIC "errors/validation" WITH "Invalid value: " + {value}

Use Dynamic Topics

Include identifiers in output topics for easy filtering:

// For topic "sensors/temp001/data" - wildcard is at position 2
SET "sensor_id" WITH TOPIC POSITION 2
PUBLISH MODEL SensorData TO "processed/" + {sensor_id} + "/data" WITH ...

Generate Unique IDs

Use RANDOM UUID for record identification:

PUBLISH MODEL Record TO "records/new" WITH
    record_id = (RANDOM UUID)
    ...

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Using AI with Coreflux

Product Updates

When to Use Action Models

Example Use Case

Step-by-Step Guide

Use Case Examples

JSON vs protobuf on the destination topic

Best Practices

Next Steps

Model Examples

Action Operations

​When to Use Action Models

​Example Use Case

​Step-by-Step Guide

​Use Case Examples

​JSON vs protobuf on the destination topic

​Best Practices

​Next Steps

Model Examples

Action Operations

When to Use Action Models

Example Use Case

Step-by-Step Guide

Use Case Examples

JSON vs protobuf on the destination topic

Best Practices

Next Steps