Connectors Overview

What are Connectors?

Connectors define connections between your Coreflux MQTT broker and external systems. They enable broker-to-broker communication, database storage, industrial protocol integration, REST API connectivity, email notifications, AI integration, and media processing — all without writing custom middleware.

Write DEFINE CONNECTOR and deploy it with -addConnector. DEFINE ROUTE and -addRoute remain supported aliases, as do the other Route command spellings. The broker stores each definition exactly as you typed it. Overload settings use ADD CONNECTOR_CONFIG; ADD ROUTE_CONFIG remains a supported alias. Older brokers accept only DEFINE ROUTE and -addRoute. See Connector command spellings.

Like plugging a cable into a device. A Connector is the cable between your broker and an external system — a database, a PLC, or another broker.

Architecture diagram showing the Coreflux MQTT broker at the center connected to databases, PLCs, REST APIs, cloud brokers, email, and AI through Connectors — Connectors connect your Coreflux MQTT broker to databases, industrial equipment, APIs, cloud services, and more

Connector Categories

System Connectors

Configure MQTT bridges and Sparkplug roles.

Streaming Connectors

Connect MQTT topics to Apache Kafka.

gRPC Connectors

Call an external gRPC service from MQTT (TYPE GRPC), with one bidirectional call per device topic by default.

Notification Connectors

Send alerts and reports to people from MQTT events.

Data Storage Connectors

Store MQTT data in PostgreSQL, MySQL, MongoDB, OpenSearch, Snowflake, and more.

REST API Connectors

HTTP client/server for external API integration and webhook endpoints.

Industrial Connectors

Connect to PLCs via Modbus, Siemens S7, OPC UA, ADS, Allen-Bradley, and FINS.

MCP Connectors

Connect out to external MCP servers (TYPE MCP / CALL MCP). To expose this broker as an MCP server, see MCP Server.

AI Connectors and Agents

Call a model with a TYPE LLM connector, or define an agent that uses it.

Media Connectors

Decode barcodes, capture camera frames, and publish microphone audio.

Overload configuration

Choose wait or drop when a connector cannot keep up with incoming MQTT.

This page covers the fundamentals of connector configuration. Choose a specific connector category above for detailed syntax and examples.

Basic Syntax

DEFINE CONNECTOR <connector_name> WITH TYPE <connector_type>
    ADD <CONFIG_SECTION>
        WITH <parameter> "<value>"
    ADD MAPPING <mapping_name>
        WITH SOURCE_TOPIC "<source>"
        WITH DESTINATION_TOPIC "<destination>"

Components

string

required

Unique identifier for the connector. Use descriptive names like CloudBridge, SensorDataIndexer, AlertEmailer.

type

required

The type of connector (see Available Connector Types below).

block

Configuration blocks specific to the connector type (e.g., MQTT_CONFIG, MONGODB_CONFIG, REST_API_CONFIG).

block

Optional. Controls wait-or-drop behavior when this connector cannot keep up with incoming MQTT. Works on every connector type. ADD ROUTE_CONFIG remains a supported alias. See Store and Forward Configuration.

block

Defines topic mappings for data flow between source and destination.

Disable a connector

Add DISABLE true on its own line to keep a connector defined but stopped:

DEFINE CONNECTOR PlantHistorian WITH TYPE POSTGRESQL
    DISABLE true
    ADD SQL_CONFIG
        WITH SERVER "historian.plant.local"

A disabled connector stays listed and its definition is still published, but it isn’t started. It opens no connection, maps no topics, and uses no action slots. DISABLE false, or no DISABLE line, means enabled. The state survives a restart. To switch it without editing the code, use -disableConnector <name> and -enableConnector <name> (aliases -disableRoute / -enableRoute). -disableConnector refuses a TYPE LLM connector that agents still use: disable those agents first. See Disable and enable.

Available Connector Types

Connector Type	Description
`MQTT_BRIDGE`	Broker-to-broker topic synchronization
`SPARKPLUG_B`	Sparkplug decoding and edge-node publishing
`SPARKPLUG_HOST`	Sparkplug Primary Host Application

Connector Type	Description
`KAFKA`	Stream data both ways with Apache Kafka
`GRPC`	Call an external gRPC service from MQTT topics (unary and streaming)

Connector Type	Description
`EMAIL`	Send templated emails from MQTT messages

Connector Type	Description
`POSTGRESQL`	PostgreSQL database storage
`MYSQL`	MySQL database storage
`MARIADB`	MariaDB database storage
`SQLSERVER`	SQL Server database storage
`MONGODB`	MongoDB document database
`OPENSEARCH`	OpenSearch indexing and search
`CRATEDB`	CrateDB time-series database
`SNOWFLAKE`	Snowflake cloud data warehouse
`FILE_STORAGE`	CSV/JSON file storage

Connector Type	Description
`MODBUS_TCP`	Modbus TCP protocol
`MODBUS_SERIAL`	Modbus RTU over RS-232/485
`SIEMENS_S7`	Siemens S7 PLCs
`OPCUA`	OPC UA client
`ADS`	Beckhoff TwinCAT ADS
`ALLEN_BRADLEY`	Allen-Bradley/Rockwell PLCs
`ETHERNETIP`	EtherNet/IP CIP protocol
`FINS`	Omron FINS protocol

Connector Type	Description
`REST_API`	REST API client/server
`MQTT_TO_REST`	Forward MQTT messages to REST endpoints
`MCP`	Model Context Protocol for external tools
`AGENT`	Autonomous AI assistant with tool access
`BARCODE`	Barcode encoding and decoding
`VIDEO_INPUT`	JPEG capture from a camera, stream, or file
`VIDEO_OUTPUT`	HTTP MJPEG from MQTT frames
`AUDIO_INPUT`	Raw PCM capture from a microphone, tone, silence, or file

Quick Examples

Email Alerts
Modbus TCP
REST API
PostgreSQL Storage
MQTT Bridge

Send email notifications from MQTT events:

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

Read data from PLCs via Modbus:

DEFINE CONNECTOR PlcReader WITH TYPE MODBUS_TCP
    ADD MODBUS_CONFIG
        WITH HOST "192.168.1.100"
        WITH PORT '502'
        WITH SLAVE_ID '1'
    ADD MAPPING Sensors
        WITH EVERY 500 MILLISECONDS
        ADD TAG Temperature
            WITH ADDRESS "100"
            WITH ADDRESS_TYPE "HOLDING_REGISTER"
            WITH DATA_TYPE "FLOAT"
            WITH SOURCE_TOPIC "plc/temperature"

Connect to external REST APIs:

DEFINE CONNECTOR WeatherService WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH ENABLE_CLIENT "true"
        WITH BASE_ADDRESS "https://api.weather.com"
        WITH USE_SSL "true"
    ADD EVENT GetWeather
        WITH SOURCE_TOPIC "weather/request"
        WITH DESTINATION_TOPIC "weather/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/v1/current?city={value.json.city}"

Store MQTT data in PostgreSQL:

DEFINE CONNECTOR SensorDB WITH TYPE POSTGRESQL
    ADD SQL_CONFIG
        WITH SERVER "postgres.example.com"
        WITH PORT '5432'
        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, value) VALUES (NOW(), '{sensor_id}', '{value.json}')"

Connect two brokers and sync topics:

DEFINE CONNECTOR 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 "EdgeBridge"
        WITH USE_TLS "true"
    ADD MAPPING sensorSync
        WITH SOURCE_TOPIC "sensors/+/data"
        WITH DESTINATION_TOPIC "edge/sensors/+/data"
        WITH DIRECTION "out"

Understanding Connector Configuration

Connectors share two common building blocks: MQTT configuration and topic mappings. You’ll see these across most connector types, so understanding them here saves time when setting up any specific connector.

Connecting to a Broker

Connectors that communicate with MQTT brokers — like bridges — need to know which broker to talk to. The simplest option is BROKER SELF, which tells the connector to use your local Coreflux broker:

ADD MQTT_CONFIG
    WITH BROKER SELF

When you need to reach a remote or authenticated broker (for example, a cloud broker or a second Coreflux instance on your network), provide the connection details explicitly:

ADD MQTT_CONFIG
    WITH BROKER_ADDRESS "192.168.1.50"
    WITH BROKER_PORT '1883'
    WITH CLIENT_ID "MyRouteClient"
    WITH USERNAME "user"
    WITH PASSWORD "password"

Not every connector type requires an MQTT_CONFIG block. Database connectors, email connectors, and REST connectors have their own configuration sections. Check the specific connector page for details.

Controlling Data Flow with Mappings

Mappings define which MQTT topics flow through a connector and in which direction. For example, to push local sensor readings to a cloud broker:

ADD MAPPING sensorSync
    WITH SOURCE_TOPIC "sensors/+/temperature"
    WITH DESTINATION_TOPIC "cloud/sensors/+/temperature"
    WITH DIRECTION "out"

The DIRECTION parameter controls where data goes:

Direction	What it does	Example scenario
`out`	Local broker to destination	Push edge data to the cloud
`in`	Destination to local broker	Pull remote commands locally
`both`	Bidirectional sync	Keep two brokers in sync

When a connector cannot keep up

Incoming MQTT can outpace the destination. Add ADD CONNECTOR_CONFIG on any connector to choose whether that connector waits or drops the newest message (ADD ROUTE_CONFIG remains a supported alias). Connectors without the block use the broker default (balanced). See Store and Forward Configuration.

Connector Status Monitoring

Every connector automatically publishes real-time health and status information to a dedicated MQTT topic. No additional configuration is required — the status topic is created as soon as the connector is defined.

Status updates are published to a $SYS topic and are available to any client subscribed to the broker.

Status Topic

The topic follows this pattern:

$SYS/Coreflux/Routes/<connector_name>/status

For example, a connector named MyS7Route publishes its status to $SYS/Coreflux/Routes/MyS7Route/status.

Status Payload

The status topic publishes a JSON object with connection health, message statistics, and timing information:

{
  "name": "MyS7Route",
  "type": "SIEMENS_S7",
  "connection": "Connected",
  "health": "Green",
  "errors": {
    "total": 0,
    "consecutive": 0,
    "rate": 0
  },
  "messages": {
    "total": 4749,
    "succeeded": 4749,
    "failed": 0
  },
  "lastActivity": "2026-03-03T15:03:14.563Z",
  "lastSuccess": "2026-03-03T15:03:14.563Z",
  "lastError": null,
  "uptime": "00:08:01",
  "connectedSince": "2026-03-03T14:55:13.299Z",
  "timestamp": "2026-03-03T15:03:14.590Z"
}

Field Reference

Field	Type	Description
`name`	string	The connector name as defined in LoT
`type`	string	The connector type (e.g., `SIEMENS_S7`, `MQTT_BRIDGE`, `POSTGRESQL`)
`connection`	string	Current connection state (`Connected`, `Disconnected`, etc.)
`health`	string	Overall health indicator (`Green`, `Yellow`, `Red`)
`errors.total`	integer	Total number of errors since the connector started
`errors.consecutive`	integer	Number of consecutive errors without a successful message
`errors.rate`	number	Current error rate
`messages.total`	integer	Total messages processed
`messages.succeeded`	integer	Messages processed successfully
`messages.failed`	integer	Messages that failed processing
`lastActivity`	string	ISO 8601 timestamp of the last message (success or failure)
`lastSuccess`	string	ISO 8601 timestamp of the last successful message
`lastError`	string \| null	ISO 8601 timestamp of the last error, or `null` if no errors occurred
`uptime`	string	Time the connector has been running (`HH:MM:SS`)
`connectedSince`	string	ISO 8601 timestamp of when the connector established its current connection
`timestamp`	string	ISO 8601 timestamp of when this status update was generated

Checking Connector Status

You can monitor any connector’s health by subscribing to its status topic. Use MQTT Explorer or any MQTT client to subscribe:

$SYS/Coreflux/Routes/+/status

This wildcard subscription shows status updates for all active connectors at once.

Use Cases

Edge-to-Cloud Sync
Industrial Integration
Real-Time Analytics
Alert Notifications

Bridge local edge data to cloud brokers:

DEFINE CONNECTOR 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 USE_TLS "true"
    ADD MAPPING allSensors
        WITH SOURCE_TOPIC "local/sensors/#"
        WITH DESTINATION_TOPIC "edge-01/sensors/#"
        WITH DIRECTION "out"

Read data from PLCs via multiple protocols:

DEFINE CONNECTOR SiemensPLC WITH TYPE SIEMENS_S7
    ADD S7_CONFIG
        WITH IP "192.168.1.100"
        WITH CPU_TYPE "S71500"
        WITH RACK '0'
        WITH SLOT '1'
    ADD MAPPING ProcessData
        WITH EVERY 500 MILLISECONDS
        ADD TAG Temperature
            WITH ADDRESS "DB1.DBD0"
            WITH DATA_TYPE "REAL"
            WITH SOURCE_TOPIC "plc/temperature"

Store sensor data for analysis:

DEFINE CONNECTOR AnalyticsStore WITH TYPE OPENSEARCH
    ADD OPENSEARCH_CONFIG
        WITH BASE_URL "https://analytics.example.com:9200"
        WITH USERNAME "admin"
        WITH PASSWORD "password"
        WITH USE_SSL "true"
    ADD EVENT IndexReading
        WITH SOURCE_TOPIC "sensors/#"
        WITH QUERY "CLEAN:{index: sensor-data, body: {timestamp: {timestamp}, value: {value.json}}}"

Email stakeholders when critical events occur:

DEFINE CONNECTOR CriticalAlerts WITH TYPE EMAIL
    ADD SMTP_CONFIG
        WITH HOST "smtp.office365.com"
        WITH PORT '587'
        WITH USE_TLS "true"
        WITH USERNAME "alerts@company.com"
        WITH PASSWORD "password"
    ADD EVENT notify
        WITH SOURCE_TOPIC "alarms/critical/+"
        WITH SUBJECT "CRITICAL: {value.json.alarm_type}"
        WITH RECIPIENT "ops-team@company.com"

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

What are Connectors?

Connector Categories

System Connectors

Streaming Connectors

gRPC Connectors

Notification Connectors

Data Storage Connectors

REST API Connectors

Industrial Connectors

MCP Connectors

AI Connectors and Agents

Media Connectors

Overload configuration

Basic Syntax

Components

Disable a connector

Available Connector Types

Quick Examples

Understanding Connector Configuration

Connecting to a Broker

Controlling Data Flow with Mappings

When a connector cannot keep up

Connector Status Monitoring

Status Topic

Status Payload

Field Reference

Checking Connector Status

Use Cases

Next Steps

Industrial Connectors

Data Storage

​What are Connectors?

​Connector Categories

System Connectors

Streaming Connectors

gRPC Connectors

Notification Connectors

Data Storage Connectors

REST API Connectors

Industrial Connectors

MCP Connectors

AI Connectors and Agents

Media Connectors

Overload configuration

​Basic Syntax

​Components

​Disable a connector

​Available Connector Types

​Quick Examples

​Understanding Connector Configuration

​Connecting to a Broker

​Controlling Data Flow with Mappings

​When a connector cannot keep up

​Connector Status Monitoring

​Status Topic

​Status Payload

​Field Reference

​Checking Connector Status

​Use Cases

​Next Steps

Industrial Connectors

Data Storage

What are Connectors?

Connector Categories

Basic Syntax

Components

Disable a connector

Available Connector Types

Quick Examples

Understanding Connector Configuration

Connecting to a Broker

Controlling Data Flow with Mappings

When a connector cannot keep up

Connector Status Monitoring

Status Topic

Status Payload

Field Reference

Checking Connector Status

Use Cases

Next Steps