Data Pipeline Routes

Data Pipeline Overview

Data pipeline routes enable data flow between your Coreflux broker and external systems. They connect your broker to other MQTT brokers and email services.

Like a postal forwarding service. Mail (messages) sent to your old address (local broker) automatically gets forwarded to your new address (remote broker). You set up the forwarding rules once, and it just works.

The pipeline route types are:

MQTT Bridge - Broker-to-broker communication for edge-to-cloud sync and multi-site connectivity
Email - Send notifications and alerts triggered by MQTT messages

MQTT Bridge

The MQTT_BRIDGE route enables seamless data transfer between two MQTT brokers. Use it for edge-to-cloud synchronization, multi-site connectivity, or backup systems.

Basic Syntax

DEFINE ROUTE CloudBridge WITH TYPE MQTT_BRIDGE
    ADD SOURCE_CONFIG
        WITH BROKER SELF
    ADD DESTINATION_CONFIG
        WITH BROKER_ADDRESS "cloud.example.com"
        WITH BROKER_PORT '8883'
        WITH CLIENT_ID "EdgeDevice001"
        WITH USE_TLS "true"
    ADD MAPPING sensorSync
        WITH SOURCE_TOPIC "sensors/+/data"
        WITH DESTINATION_TOPIC "edge/sensors/+/data"
        WITH DIRECTION "out"

Connection Configuration

Broker Settings

BROKER_ADDRESS

string

required

IP address or hostname of the broker. Use SELF for the local broker.

BROKER_PORT

integer

required

MQTT port (typically 1883 for unencrypted, 8883 for TLS).

CLIENT_ID

string

required

Unique client identifier for the bridge connection.

Authentication

USERNAME

string

Authentication username.

PASSWORD

string

Authentication password.

TLS/Security

USE_TLS

boolean

Enable TLS encryption. Default: false.

ALLOW_UNTRUSTED_CERTS

boolean

Allow untrusted certificates. Default: false.

SERVER_CA_CERT_PATH

string

Path to CA certificate file.

CLIENT_CERT_PATH

string

Path to client certificate for mTLS.

CLIENT_CERT_PASS

string

Client certificate password.

Connection Handling

RECONNECTION_RETRIES

integer

Number of reconnection attempts. Default: 5.

Mapping Configuration

Topic Mapping

SOURCE_TOPIC

string

required

Topic pattern to subscribe to on the source broker. Supports wildcards.

DESTINATION_TOPIC

string

required

Topic to publish to on the destination broker.

DIRECTION

string

required

Data flow direction: out (source→dest), in (dest→source), or both.

Bridge Examples

Edge to Cloud
Cloud to Edge
Bidirectional
Secure mTLS Bridge

Send local sensor data to a cloud broker:

DEFINE ROUTE EdgeToCloud WITH TYPE MQTT_BRIDGE
    ADD SOURCE_CONFIG
        WITH BROKER SELF
    ADD DESTINATION_CONFIG
        WITH BROKER_ADDRESS "iot.cloud-provider.com"
        WITH BROKER_PORT '8883'
        WITH CLIENT_ID "EdgeDevice001"
        WITH USERNAME "edge_user"
        WITH PASSWORD "secure_password"
        WITH USE_TLS "true"
    ADD MAPPING sensorData
        WITH SOURCE_TOPIC "sensors/+/data"
        WITH DESTINATION_TOPIC "edge-001/sensors/+/data"
        WITH DIRECTION "out"

Receive commands from cloud to local broker:

DEFINE ROUTE CloudCommands WITH TYPE MQTT_BRIDGE
    ADD SOURCE_CONFIG
        WITH BROKER SELF
    ADD DESTINATION_CONFIG
        WITH BROKER_ADDRESS "iot.cloud-provider.com"
        WITH BROKER_PORT '8883'
        WITH CLIENT_ID "EdgeDevice001-Commands"
        WITH USE_TLS "true"
    ADD MAPPING commands
        WITH SOURCE_TOPIC "local/commands/#"
        WITH DESTINATION_TOPIC "edge-001/commands/#"
        WITH DIRECTION "in"

Full two-way synchronization:

DEFINE ROUTE FullSync WITH TYPE MQTT_BRIDGE
    ADD SOURCE_CONFIG
        WITH BROKER SELF
    ADD DESTINATION_CONFIG
        WITH BROKER_ADDRESS "partner-broker.example.com"
        WITH BROKER_PORT '1883'
        WITH CLIENT_ID "SyncClient"
    ADD MAPPING bidirectionalSync
        WITH SOURCE_TOPIC "shared/data/#"
        WITH DESTINATION_TOPIC "partner/data/#"
        WITH DIRECTION "both"

TLS-encrypted bridge with mutual authentication:

DEFINE ROUTE SecureBridge WITH TYPE MQTT_BRIDGE
    ADD SOURCE_CONFIG
        WITH BROKER_ADDRESS "192.168.1.10"
        WITH BROKER_PORT '8883'
        WITH CLIENT_ID "SecureSource"
        WITH USE_TLS "true"
        WITH SERVER_CA_CERT_PATH "/certs/ca.pem"
        WITH CLIENT_CERT_PATH "/certs/client.pem"
        WITH CLIENT_CERT_PASS "cert_password"
    ADD DESTINATION_CONFIG
        WITH BROKER_ADDRESS "secure.cloud.com"
        WITH BROKER_PORT '8883'
        WITH CLIENT_ID "SecureDest"
        WITH USE_TLS "true"
        WITH ALLOW_UNTRUSTED_CERTS "false"
    ADD MAPPING secureMapping
        WITH SOURCE_TOPIC "secure/+/data"
        WITH DESTINATION_TOPIC "cloud/+/data"
        WITH DIRECTION "out"

Email Route

The EMAIL route sends templated emails based on MQTT messages. Ideal for alerts, notifications, and reports.

Basic Syntax

DEFINE ROUTE AlertEmail WITH TYPE EMAIL
    ADD SMTP_CONFIG
        WITH HOST "smtp.gmail.com"
        WITH PORT '587'
        WITH USERNAME "alerts@company.com"
        WITH PASSWORD "app-password-here"
        WITH USE_TLS "true"
    ADD EVENT criticalAlert
        WITH SOURCE_TOPIC "alerts/critical/+"
        WITH SUBJECT "CRITICAL: {value.json.alert_type}"
        WITH RECIPIENT "ops-team@company.com"

SMTP Configuration

Server Settings

HOST

string

required

SMTP server address (e.g., smtp.gmail.com, smtp.office365.com).

PORT

integer

required

SMTP port (typically 587 for TLS, 465 for SSL).

USE_TLS

boolean

Enable TLS encryption. Recommended: true.

Authentication

USERNAME

string

required

Email account to send from.

PASSWORD

string

required

Email password or app-specific password.

Event Configuration

Trigger & Routing

SOURCE_TOPIC

string

MQTT topic to trigger email sending.

DESTINATION_TOPIC

string

MQTT topic to publish send status.

Email Content

TEMPLATE_PATH

string

Path to HTML email template file.

SUBJECT

string

Email subject line (supports placeholders).

RECIPIENT

string

Email recipient address (supports placeholders).

Template Placeholders

Placeholder	Description
`{value}`	Full JSON payload as string
`{value.json.field}`	Single field from JSON
`{value.json.nested.field}`	Nested field access

Email Examples

Critical Alerts
Dynamic Recipient
With Embedded Resources

Send immediate alerts for critical events:

DEFINE ROUTE CriticalAlerts WITH TYPE EMAIL
    ADD SMTP_CONFIG
        WITH HOST "smtp.gmail.com"
        WITH PORT '587'
        WITH USERNAME "alerts@company.com"
        WITH PASSWORD "app-password-here"
        WITH USE_TLS "true"
    ADD EVENT criticalNotify
        WITH SOURCE_TOPIC "alerts/critical/+"
        WITH TEMPLATE_PATH "/templates/critical_alert.html"
        WITH SUBJECT "CRITICAL: {value.json.alert_type}"
        WITH RECIPIENT "ops-team@company.com"

Send to email address from payload:

DEFINE ROUTE UserNotifications WITH TYPE EMAIL
    ADD SMTP_CONFIG
        WITH HOST "smtp.office365.com"
        WITH PORT '587'
        WITH USERNAME "noreply@company.com"
        WITH PASSWORD "secure-password"
        WITH USE_TLS "true"
    ADD EVENT userNotify
        WITH SOURCE_TOPIC "users/+/notifications"
        WITH TEMPLATE_PATH "/templates/user_notification.html"
        WITH SUBJECT "Hello {value.json.username}"
        WITH RECIPIENT "{value.json.email}"

Include images in email:

DEFINE ROUTE ReportEmail WITH TYPE EMAIL
    ADD SMTP_CONFIG
        WITH HOST "smtp.company.com"
        WITH PORT '587'
        WITH USERNAME "reports@company.com"
        WITH PASSWORD "password"
        WITH USE_TLS "true"
    ADD EVENT sendReport
        WITH SOURCE_TOPIC "reports/daily"
        WITH TEMPLATE_PATH "/templates/daily_report.html"
        WITH SUBJECT "Daily Report - {value.json.date}"
        WITH RECIPIENT "management@company.com"
        WITH EMBED_RESOURCES
            ADD "/images/logo.png"
            ADD "/images/chart.png"

For Gmail, use an App Password instead of your regular password. Enable 2FA on your Google account first.

Troubleshooting

Bridge Connection Timeout

Verify IP address and port are correct
Check firewall allows outbound connections on the configured port
Ensure remote broker is running and accessible
Try increasing RECONNECTION_RETRIES

TLS/Certificate Errors

Verify certificate paths are correct
Ensure certificates are valid and not expired
Check certificate chain is complete
For testing, use ALLOW_UNTRUSTED_CERTS "true" (not recommended for production)

Messages Not Bridging

Verify SOURCE_TOPIC and DESTINATION_TOPIC are correct
Check DIRECTION setting matches your expected data flow
Ensure CLIENT_ID is unique across all connections
Check broker logs for subscription errors

Email Not Sending

Verify SMTP credentials are correct
For Gmail, ensure you’re using an App Password
Check firewall allows outbound SMTP connections
Verify recipient email address is valid

Email Authentication Failed

Double-check USERNAME and PASSWORD
Ensure 2FA is enabled if using app passwords
Some providers require enabling “less secure apps” access
Check for account lockouts due to failed attempts

Best Practices

Use TLS for External Connections

Always enable TLS when connecting to external brokers:

WITH USE_TLS "true"

Unique Client IDs

Each bridge connection needs a unique CLIENT_ID to avoid conflicts:

WITH CLIENT_ID "EdgeDevice001-SensorBridge"

Test with Local Broker First

Use WITH BROKER SELF during development:

ADD SOURCE_CONFIG
    WITH BROKER SELF

Secure Credentials

Store passwords securely and use app-specific passwords where available.

Handle Reconnection

Configure appropriate retry settings:

WITH RECONNECTION_RETRIES '10'

Quick Start

Developing and Managing Coreflux

Coreflux MQTT Broker

LoT Language

Product Updates

Data Pipeline Overview

MQTT Bridge

Basic Syntax

Connection Configuration

Mapping Configuration

Bridge Examples

Email Route

Basic Syntax

SMTP Configuration

Event Configuration

Template Placeholders

Email Examples

Troubleshooting

Best Practices

Next Steps

Data Storage Routes

REST API Routes

​Data Pipeline Overview

​MQTT Bridge

​Basic Syntax

​Connection Configuration

​Mapping Configuration

​Bridge Examples

​Email Route

​Basic Syntax

​SMTP Configuration

​Event Configuration

​Template Placeholders

​Email Examples

​Troubleshooting

​Best Practices

​Next Steps

Data Storage Routes

REST API Routes

Data Pipeline Overview

MQTT Bridge

Basic Syntax

Connection Configuration

Mapping Configuration

Bridge Examples

Email Route

Basic Syntax

SMTP Configuration

Event Configuration

Template Placeholders

Email Examples

Troubleshooting

Best Practices

Next Steps