MySQL Route

MySQL Overview

The MYSQL route stores MQTT messages in MySQL databases. It supports SSL connections, parameterized queries, and automatic reconnection for reliable data persistence.

MySQL is ideal for web applications and general-purpose data storage. Its wide ecosystem support makes it a reliable choice for most IoT data persistence needs.

Basic Syntax

DEFINE ROUTE SensorDB WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.example.com"
        WITH PORT '3306'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT StoreSensorReading
        WITH SOURCE_TOPIC "sensors/+/reading"
        WITH QUERY "INSERT INTO readings (ts, sensor_id, value) VALUES (NOW(), '{sensor_id}', '{value.json}')"

Connection Configuration

SQL_CONFIG Parameters

SERVER

string

required

MySQL server hostname or IP address.

DATABASE

string

required

Target database name.

USERNAME

string

required

Database username.

PASSWORD

string

required

Database password.

Optional Settings

PORT

integer

MySQL port. Default: 3306.

USE_SSL

boolean

Enable SSL connection. Default: false.

CONNECTION_TIMEOUT

integer

Connection timeout in seconds. Default: 30.

Never hardcode credentials in production. Use environment variables and encrypted secrets to keep sensitive values out of your route definitions:

WITH SERVER GET ENV "DB_HOST"
WITH USERNAME GET ENV "DB_USER"
WITH PASSWORD GET SECRET "DB_PASSWORD"

See Environment Variables & Secrets for setup and usage.

Event Configuration

EVENT Parameters

SOURCE_TOPIC

string

required

MQTT topic pattern that triggers the INSERT. Supports wildcards.

QUERY

string

required

SQL INSERT statement with placeholders for dynamic values.

Optional Settings

DESTINATION_TOPIC

string

MQTT topic to publish query result status.

Query Placeholders

Placeholder	Description
`{value.json}`	Full JSON payload
`{value.json.field}`	Specific field from JSON
`{timestamp}`	Message timestamp
`{source_topic}`	Original MQTT topic
`{field}`	Field extracted from topic path

Writing Data

Store sensor readings with timestamps:

DEFINE ROUTE TemperatureDB WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.example.com"
        WITH PORT '3306'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT StoreTemperature
        WITH SOURCE_TOPIC "sensors/+/temperature"
        WITH QUERY "INSERT INTO temperature_readings (recorded_at, sensor_id, value) VALUES (NOW(), '{sensor_id}', {value.json.value})"

Secure connection with SSL:

DEFINE ROUTE SecureDB WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.cloud.com"
        WITH PORT '3306'
        WITH DATABASE "production"
        WITH USERNAME "app_user"
        WITH PASSWORD "secure_password"
        WITH USE_SSL "true"
    ADD EVENT StoreEvent
        WITH SOURCE_TOPIC "events/#"
        WITH QUERY "INSERT INTO events (ts, topic, payload) VALUES (NOW(), '{source_topic}', '{value.json}')"

Different tables for different data types:

DEFINE ROUTE MultiTableDB WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.example.com"
        WITH PORT '3306'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    
    ADD EVENT StoreSensorData
        WITH SOURCE_TOPIC "sensors/+/data"
        WITH QUERY "INSERT INTO sensor_data (ts, sensor_id, value) VALUES (NOW(), '{sensor_id}', '{value.json}')"
    
    ADD EVENT StoreAlerts
        WITH SOURCE_TOPIC "alerts/+"
        WITH QUERY "INSERT INTO alerts (ts, alert_type, message) VALUES (NOW(), '{alert_type}', '{value.json}')"

Insert a record or update it if a duplicate key exists. MySQL uses ON DUPLICATE KEY UPDATE:

DEFINE ROUTE DeviceRegistry WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.example.com"
        WITH PORT '3306'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT RegisterDevice
        WITH SOURCE_TOPIC "devices/+/registered"
        WITH QUERY "INSERT INTO devices (device_id, name, area, active, updated_at) VALUES ('{value.json.device_id}', '{value.json.name}', '{value.json.area}', 1, NOW()) ON DUPLICATE KEY UPDATE name = VALUES(name), area = VALUES(area), updated_at = NOW()"

ON DUPLICATE KEY UPDATE fires only when a row with the same unique key already exists — the existing row is updated instead of inserting a duplicate.

Modify existing records when a status or value change arrives:

DEFINE ROUTE DeviceRegistry WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.example.com"
        WITH PORT '3306'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT UpdateDeviceStatus
        WITH SOURCE_TOPIC "devices/+/status"
        WITH QUERY "UPDATE devices SET current_status = '{value.json.status}', last_seen = NOW() WHERE device_id = '{value.json.device_id}'"

Remove records on a decommission or cleanup event:

DEFINE ROUTE DeviceRegistry WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.example.com"
        WITH PORT '3306'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT DecommissionDevice
        WITH SOURCE_TOPIC "devices/+/decommissioned"
        WITH QUERY "DELETE FROM devices WHERE device_id = '{value.json.device_id}' AND active = 0"

Alternative: STORE IN with Models — Instead of writing EVENT queries, you can bind a model directly to this route. Every PUBLISH MODEL call automatically inserts a row — no query needed:

DEFINE MODEL SensorReading
    ADD STRING "sensor_id"
    ADD DOUBLE "value"
    STORE IN "TemperatureDB"
        WITH TABLE "temperature_readings"

See Data Storage Overview for the full STORE IN workflow.

Reading Data

EVENTs also support SELECT queries to read data back from the database. Publish a message to the event’s SOURCE_TOPIC, and the query result is published to DESTINATION_TOPIC:

DEFINE ROUTE SensorDB WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.example.com"
        WITH PORT '3306'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT GetLatestReading
        WITH SOURCE_TOPIC "db/query/latest"
        WITH DESTINATION_TOPIC "db/result/latest"
        WITH QUERY "SELECT sensor_id, value FROM readings WHERE sensor_id = '{payload}' ORDER BY ts DESC LIMIT 1"

To trigger this query, publish the sensor ID to the source topic:

Topic:   db/query/latest
Payload: temp001

The query result is published to db/result/latest, where your actions or external clients can consume it.

Timed Polling

EVENTs can also run SELECT queries on a fixed schedule using WITH EVERY:

DEFINE ROUTE DeviceRegistry WITH TYPE MYSQL
    ADD SQL_CONFIG
        WITH SERVER "mysql.example.com"
        WITH PORT '3306'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT PollActiveDevices
        WITH EVERY 10 SECONDS
        WITH DESTINATION_TOPIC "dashboard/devices/active"
        WITH QUERY "SELECT device_id, name, current_status, last_seen FROM devices WHERE active = 1 ORDER BY name"

Table Schema Examples

-- Basic sensor readings
CREATE TABLE sensor_readings (
    id INT AUTO_INCREMENT PRIMARY KEY,
    recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    sensor_id VARCHAR(50),
    value DOUBLE
);

-- Event log
CREATE TABLE events (
    id INT AUTO_INCREMENT PRIMARY KEY,
    ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    topic VARCHAR(255),
    payload JSON
);

Troubleshooting

Connection Refused

Verify SERVER and PORT are correct
Check MySQL is running and accepting connections
Verify firewall allows connections on port 3306
Check bind-address in MySQL configuration

Authentication Failed

Verify USERNAME and PASSWORD are correct
Check user has permissions on the DATABASE
Verify user can connect from your host

Query Syntax Errors

Check placeholder names match topic structure
Verify table and column names exist
Test query manually in MySQL client first

Next Steps

Data Storage Routes Overview

Compare all storage options.

MariaDB Route

Configure MariaDB storage (MySQL-compatible).

Quick Start

Developing and Managing Coreflux

Coreflux MQTT Broker

LoT Language

Product Updates

MySQL Overview

Basic Syntax

Connection Configuration

SQL_CONFIG Parameters

Event Configuration

EVENT Parameters

Query Placeholders

Writing Data

Reading Data

Timed Polling

Table Schema Examples

Troubleshooting

Next Steps

Data Storage Routes Overview

MariaDB Route

​MySQL Overview

​Basic Syntax

​Connection Configuration

​SQL_CONFIG Parameters

​Event Configuration

​EVENT Parameters

​Query Placeholders

​Writing Data

​Reading Data

​Timed Polling

​Table Schema Examples

​Troubleshooting

​Next Steps

Data Storage Routes Overview

MariaDB Route

MySQL Overview

Basic Syntax

Connection Configuration

SQL_CONFIG Parameters

Event Configuration

EVENT Parameters

Query Placeholders

Writing Data

Reading Data

Timed Polling

Table Schema Examples

Troubleshooting

Next Steps