Business Plugin Development

This document uses "business plugin" as a collective term for data forwarding targets derived from DataForwardBase, protocol servers, database writers, cloud integrations, and synchronization bridges. Business plugins do not collect raw data from field devices. They consume collected variables, device states, alarms, or plugin events and write them to external systems or expose server-side protocol capabilities.

To extend drivers for PLCs, instruments, OPC, MQTT collection, or other devices, see Collection Plugin Development.

Source Code Entry

Entry	Function
ThingsGatewayRuntime.Application/Driver/DataForward/DataForwardBase.cs	runtime base class for all business plugins.
ThingsGatewayRuntime.Application/Driver/DataForward/DataForwardPropertyBase.cs	Root-level plugin property class.
ThingsGatewayRuntime.Application/Driver/DataForward/DataForwardVariablePropertyBase.cs	Root class of the target variable attribute.
ThingsGatewayRuntime.Application/Driver/DataForward/DataForwardChannelPropertyBase.cs	requires TCP, serial port, DTU, SSL target attribute base class.
ThingsGatewayRuntime.Application/Driver/DataForward/Cache/*	Cache queue, offline file cache, topic template, script model, batch upload base class.
ThingsGatewayRuntime.Application/Task/DataForward/DataForwardMange/DataForwardMange.cs	Forwards groups, target launches, event distribution, group-level data production , redundant switching scheduling entry.
ThingsGatewayRuntime.Application/Model/DataForwardGroupRuntime.cs	forwards the range-parsed runtime variables and device indexes of the group.
ThingsGatewayRuntime.Plugin/Plugin/*	MQTT, Kafka, RabbitMQ, Webhook, ModbusSlave, SyncBridge, etc. implementations.
ThingsGatewayRuntime.NOAOTPlugin/Plugin/OpcUa/OpcUaServer	OPC UA Server business plugin implementation.
ThingsGatewayRuntimePRO/src/ThingsGatewayRuntime.NOAOTPROPlugin/Plugin/*	Implemented in professional versions such as IEC104, IEC61850 Server, ZeroMQ, etc.

Boundaries between Business Plugins and Collection Plugins

Project	Collection Plugins	Business Plugins
Base Classes	CollectBase	DataForwardBase
Configuration Location	Collection Devices	Data forwarding objectives
Data sources	PLC, instruments, external reporting, virtual computing	Runtime variables, device status, alarms, plugin events
Main tasks	Read and write on-site locations	Upload, storage, server-side protocols, bridging, write-back
variable range	Variable enabled under the device	Range of variables parsed by the forwarding group
Write entry	InvokeWriteAsync, InvokeMethodAsync	After writing the target protocol, the acquisition side RPC is usually called again.

Do not write the business plugin as "then scan all variables globally." The variable range is determined by the forwarding group. Business plugins should read the currently visible data from the target via GetVariables(), IdVariableRuntimes, CollectDevices or the group production entry.

Plugin Discovery Rules

Rules	Description
Inherit DataForwardBase	PluginService Only recognize non-abstract derived classes of DataForwardBase as data forwarding plugins.
Publicly disclose the parameterless constructor	Created at runtime via Activator.CreateInstance, with each target having an independent instance.
The full name of the plugin is the configuration key	Data forwarding target storage type FullName. Renaming the namespace or class name will affect the old configuration.
Inject the target property first, then initialize	DataForwardMange.StartTargetAsync Call SetDriverProperties, then InitTarget, InitAsync, StartAsync.
The target variable attribute does not determine the variable range	The target variable attribute only stores plugin-specific configurations such as external mappings, permissions, and data types. Whether a variable enters the target depends on the range of the forwarded group and the relationship between group variables.
UsesGroupDataProducer determines the data entry point	cached targets, usually true, consuming data produced by the unified forwarding group; The server-side protocol class is usually false, which directly handles variable changes.

Base Class Selection

Base Class	Suitable Scenario	Focus must be placed on	existing references
DataForwardBase	protocol server, synchronization bridge, and targets requiring self-managed memory mapping or connection loops.	TargetProperties,TargetPropertyType, optional VariablePropertyType, ProtectedInitAsync, AfterVariablesChangedAsync, OnVariableChanged, ProtectedExecuteAsync, IsConnected.	ModbusSlave, OpcUaServer, IEC104Slave, IEC61850Server, SyncBridge.
DataForwardBaseWithCache	requires offline caching, but the trigger model does not fully correspond to the target uploaded by periodic variables.	DataForwardPropertyWithCacheDataForwardPropertyWithCache, enabled models, Update*Model or AcceptProduced*.	HisAlarmForwardTarget.
DataForwardBaseWithCacheInterval	Consumes data produced by the forwarding group uniformly, supporting periodic/variable/batch processing and offline caching.	DataForwardPropertyWithCacheInterval, Model Switch, AcceptProduced* or Update*Model.	HisDataForwardTarget, RealDataForwardTarget, ThingsBoardClientProducer.
DataForwardBaseWithCacheIntervalScript	Requires topic templates, entity scripts, and custom upload templates, but only wants to reuse conversion capabilities.	After inheriting, call methods such as GetVariableBasicDataTopicArray by model.	Parent class of the script upload class target.
DataForwardBaseWithCacheIntervalScriptAll	MQTT, Kafka, RabbitMQ, Webhook, ZeroMQ and similar "Topic/Payload Generated and Uploaded" targets.	Implement Upload(TopicArray, CancellationToken), initialize client connections, and override IsConnected.	MqttClientProducer, KafkaProducer, RabbitMQProducer, Webhook, ZeroMQProducer.
DataForwardChannelPropertyBase	target requires TCP client/server, serial port, DTU, and SSL parameters.	property class inherits it and calls ProtectedInitAsync in InitChannelAsync.	ModbusSlaveProperty, IEC104SlaveProperty.

If the goal is simply to "convert variables to JSON and send to a certain system," prioritize using DataForwardBaseWithCacheIntervalScriptAll. If the goal is to simulate a protocol server externally and allow external systems to actively read and write memory point tables, prioritize using DataForwardBase.

Target Attributes and Variable Attributes

Types	Functions
DataForwardPropertyBase	Goal-level property root class. Connection address, authentication, table name, cache, and template are all target attributes.
DataForwardVariablePropertyBase	Root class of the target variable-level property. The external address, data type, write permissions, etc., of a single variable under a certain target fall under this category.
DataForwardVariableProperty	General reserved field Data 1 to Data 10, suitable for simple external mapping.
DataForwardPropertyWithCache	Offline caching, memory queue limit, upload sharding, offline data filtering, upload concurrency.
DataForwardPropertyWithCacheInterval	Inherits cache properties; the trigger mode and cycle are uniformly determined by the forwarding group.
DataForwardPropertyWithCacheIntervalScript	Adds detailed logs, JSON format, list/dictionary upload, topic template, entity script, and upload template.
DataForwardChannelPropertyBase	Adds channel type, remote address, local binding, SSL, serial port, heartbeat, DTU, concurrency, connection timeout, etc.

When implementing the target attribute, ensure that both the instance and type of the target attribute are consistent:

private readonly MyTargetProperty _properties = new();

public override DataForwardPropertyBase TargetProperties => _properties;

public override Type TargetPropertyType { get; } = typeof(MyTargetProperty);

If the plugin has variable configurations, then provide:

private readonly MyVariableProperty _variableProperties = new();

public override DataForwardVariablePropertyBase VariablePropertys => _variableProperties;

public override Type? VariablePropertyType { get; } = typeof(MyVariableProperty);

DataForwardVariablePropertyBase.Enable is not a dynamic property, so the page will not have a second enable switch. Whether to forward a variable is determined by the range of group variables, the enabled status of variables within the group, the enabled status of the target, and the trigger conditions.

DynamicProperty Contract

business plugins and collection plugins use the same set of dynamic property rules.

Project	Instructions
Only [DynamicProperty] will be displayed and saved	Do not tag runtime connection objects, client objects, cache fields, or temporary indexes.
The default value will enter the new form	Default port, Topic, table name, and cache off/enable policies should be conservative.
CertificatePurpose for certificate dropdown	Certificate fields such as MQTT, OPC UA, IEC61850 TLS should indicate Client, Server, or CA.
Remark Specify the unit and format clearly	such as milliseconds, seconds, topic templates, SQL table names, and concatenation string formats.
Complex objects can also be used as properties	such as upload template configuration or SOE configuration, but JSON serialization and import/export must be restored.

Lifecycle Hooks

Phase	Runtime Actions	Plugin Development Notes
Constructor	Create the target plugin instance.	Only initialize lightweight fields, do not connect to external systems.
Property injection	SetDriverProperties(forwarder.TargetProperties, forwarder.TargetPropertyType, target.TargetPropertys).	ProtectedInitAsync can read the full target property.
InitTarget	mounts CurrentGroup,CurrentTarget, logs, target.Forwarder, calls ProtectedInitTarget.	You can cache basic information of groups/targets, but do not start a connection.
InitAsync	call ProtectedInitAsync, then AfterVariablesChangedAsync.	Parsing properties, initializing caches, creating client configurations, and building variable mappings.
StartAsync	calls ProtectedStartAsync, controlled by StartTimeout. After success, set IsStarted.	When connecting to an external system or starting a server listener, if a failure fails, throw an exception or return the failure state.
GetTasks	Create a target scheduling loop. By default, ProtectedExecuteAsync is called by group cycle.	Most cached targets reuse parent class tasks; The server goal can be to refresh memory or check connections during execution.
Initial Snapshot	After the target is launched, the forwarding manager pushes the current snapshot to the group production target.	Don't assume you have to wait for the next variable change before data comes in.
Variable/Device/Alarm/Event Changes	Based on UsesGroupDataProducer AcceptProduced* or On*Changed.	Cached targets should not repeatedly subscribe to global events themselves.
StopAsync/SafetyDisposeAsync	Stop scheduling loops to release channels, logs, caches, clients, and servers.	Connection must be closed, cache released, and subscription canceled to avoid repeated resending after restarting.

The forwarding group is the only entry point to the variable range

DataForwardGroupRuntime.RebuildVariables() generates two runtime indexes based on the group configuration.

Index	Description
IdVariableRuntimes	The variable after parsing within the current group range, with the key being variable Id.
CollectDevices	Collection devices that are derived from variables within the current group range.

Business plugins should read data through these entry points:

entry	suitable scenario
GetVariables()	traverses the visible variable of the current target, which is the most commonly used.
IdVariableRuntimes	Quickly search by variable ID.
CollectDevices	Generate device snapshots, protocol server nodes, or device connection status.
GetVariableProperty<TProperty>(variable)	Read the strong-type target variable attribute of a variable under the current target.
GetVariablePropertyValue(variable, propertyName)	Read a single field of the target variable property; if not configured, it falls back to the default property instance.
TryGetVariablePropertyValue(...)	Used when you need to distinguish between "not configured" and "configured as null".

The target variable attribute cannot "pull" the variable into the forwarding range. When "variable has value but target hasn't been sent" appears on site, first check the forwarding group range, variable enabled within the group, trigger mode, target enabled, and target connection.

Data Entry: Direct Events and Group Production

Pattern	UsesGroupDataProducer	Data Entry	Plugin Fit
Direct Event	false	OnVariableChanged, OnDeviceChanged, OnAlarmChanged, OnPluginEventChanged	ModbusSlave, IEC104Slave, OPC UA Server, IEC61850Server, SyncBridge.
Group Production	true	AcceptProducedVariableChange, AcceptProducedVariableSnapshot, AcceptProducedDeviceSnapshot, AcceptProducedAlarm, AcceptProducedPluginEvent	MQTT, Kafka, RabbitMQ, Webhook, ZeroMQ, historical data, real-time data, historical alarms.

cached base classes fix UsesGroupDataProducer to true. The forwarding manager unifies production data based on group trigger mode, batch mode, and maximum batch size, then calls AcceptProduced*. This way, multiple objectives won't have to perform scope parsing and batch processing separately.

Caching and Failure Semantics

DataForwardBaseWithCache Maintain independent memory queues and file caches for each model.

Properties	Description
CacheEnable	Whether offline file caching is enabled. When closing the shutdown, the failed data is discarded semantics using at-most-once.
CacheFileMaxLength	maximum row length per cache file; if exceeded, old data is deleted.
SplitSize	batch split size when uploading or reposting.
QueueMaxCount	Memory queue limit; if exceeded, orders are placed first; if no order is possible, old data is discarded.
OnlineFilter	whether to filter offline variables.
Concurrency	Topic: The number of concurrent uploads is implemented by the specific plugin. When the

sending method fails, the cache base class will decide whether to place the disk based on CacheEnable. The script upload target also writes the generated but unconfirmed TopicArray into the Topic outbox. After connection recovery, it ReplayTopicUploadCache first reissue the old data and then process the new data.

Topic, Script, and Upload Template

DataForwardBaseWithCacheIntervalScriptAll has handled these tasks:

Capability	Description
Topic Template	ThingsGateway/Variable/${DeviceName} These templates are grouped and replaced by entity attributes.
Entity scripts	variables, devices, alarms, and plugin events can all be reshaped through dynamic model scripts.
Upload Template	After configuring the content template, use ${property-name} to generate a custom payload.
List/Item by Item	IsVariableList, IsDeviceList, etc. to control whether to merge or send each item.
Dictionary upload	variable/alarm can be converted into the dictionary structure of DeviceName -> Name -> entity.
JSON format	Indentation and ignoring null are controlled by properties.

These plugins usually only need to implement Upload(TopicArray topicArray, CancellationToken cancellationToken).

Script Upload Target Template

public sealed class MyProducer : DataForwardBaseWithCacheIntervalScriptAll
{
    private readonly MyProducerProperty _properties = new();
    private readonly DataForwardVariableProperty _variableProperties = new();
    private MyClient? _client;
    private bool _success = true;

    protected override DataForwardPropertyWithCacheIntervalScript DataForwardPropertyWithCacheIntervalScript => _properties;

    public override Type TargetPropertyType { get; } = typeof(MyProducerProperty);

    public override Type VariablePropertyType { get; } = typeof(DataForwardVariableProperty);

    public override DataForwardVariablePropertyBase VariablePropertys => _variableProperties;

    public override bool IsConnected() => _success && _client?.Connected == true;

    protected override async Task ProtectedInitAsync(CancellationToken cancellationToken)
    {
        _client = new MyClient(_properties.Endpoint, _properties.Token);
        await base.ProtectedInitAsync(cancellationToken).ConfigureAwait(false);
    }

    protected override async Task ProtectedStartAsync(CancellationToken cancellationToken)
    {
        await _client!.ConnectAsync(cancellationToken).ConfigureAwait(false);
        await base.ProtectedStartAsync(cancellationToken).ConfigureAwait(false);
    }

    protected override async ValueTask<OperResult> Upload(TopicArray topicArray, CancellationToken cancellationToken)
    {
        try
        {
            await _client!.PublishAsync(topicArray.Topic, topicArray.Payload.Memory, cancellationToken).ConfigureAwait(false);
            _success = true;
            return OperResult.Success;
        }
        catch (Exception ex)
        {
            _success = false;
            return new OperResult(ex);
        }
    }

    protected override async Task SafetyDisposeAsync(bool disposing)
    {
        if (_client != null)
            await _client.DisposeAsync().ConfigureAwait(false);
        await base.SafetyDisposeAsync(disposing).ConfigureAwait(false);
    }
}

public sealed class MyProducerProperty : DataForwardPropertyWithCacheIntervalScript
{
    [DynamicProperty("Service address")]
    public string Endpoint { get; set; } = "http://127.0.0.1:8080";

    [DynamicProperty("Access token")]
    public string Token { get; set; } = string.Empty;
}

Protocol Server Target Template

Protocol server targets usually do not use script upload base classes, but instead maintain an external protocol memory mapping.

public sealed class MyServerTarget : DataForwardBase
{
    private readonly MyServerProperty _properties = new();
    private readonly MyServerVariableProperty _variableProperties = new();
    private readonly ConcurrentQueue<VariableRuntime> _changedVariables = new();
    private MyServer? _server;

    public override DataForwardPropertyBase TargetProperties => _properties;

    public override Type TargetPropertyType { get; } = typeof(MyServerProperty);

    public override DataForwardVariablePropertyBase VariablePropertys => _variableProperties;

    public override Type VariablePropertyType { get; } = typeof(MyServerVariableProperty);

    public override bool IsConnected() => _server?.IsRunning == true;

    protected override async Task ProtectedInitAsync(CancellationToken cancellationToken)
    {
        _server = new MyServer(_properties.BindUrl);
        await base.ProtectedInitAsync(cancellationToken).ConfigureAwait(false);
    }

    public override async Task AfterVariablesChangedAsync(CancellationToken cancellationToken)
    {
        await base.AfterVariablesChangedAsync(cancellationToken).ConfigureAwait(false);
        foreach (var variable in GetVariables())
        {
            var map = GetVariableProperty<MyServerVariableProperty>(variable);
            if (map != null)
                _server!.Register(map.Address, variable.DataType);
        }
    }

    protected override async Task ProtectedStartAsync(CancellationToken cancellationToken)
    {
        await _server!.StartAsync(cancellationToken).ConfigureAwait(false);
        await base.ProtectedStartAsync(cancellationToken).ConfigureAwait(false);
    }

    public override void OnVariableChanged(VariableRuntime variableRuntime, VariableBasicData variableData, CancellationToken cancellationToken)
    {
        if (cancellationToken.IsCancellationRequested)
            return;
        _changedVariables.Enqueue(variableRuntime);
    }

    protected override Task ProtectedExecuteAsync(object? state, CancellationToken cancellationToken)
    {
        while (_changedVariables.TryDequeue(out var variable))
        {
            var map = GetVariableProperty<MyServerVariableProperty>(variable);
            if (map != null)
                _server!.SetValue(map.Address, variable.Value);
        }
        return Task.CompletedTask;
    }
}

ModbusSlave, IEC104Slave, OpcUaServer, and IEC61850Server all follow this model: variable changes first enter a mapping or queue maintained by the plugin, then the protocol server exposes read/write services.

Current Source Code Business Plugin Reference Matrix

The table below checks 15 business plugins based on the current source code. Before developing new goals, prioritize looking for the type closest to your reference.

Plugin	Base Class	Target Attributes	Variable Attributes	Key Points
HisDataForwardTarget	DataForwardBaseWithCacheInterval	HisDataForwardProperty	HisDataForwardVariableProperty	Historical data entry, sampling strategies, conditional expressions, tabs, retention days.
RealDataForwardTarget	DataForwardBaseWithCacheInterval	RealDataForwardProperty	None	Real-time table overwrite, period/change trigger, database writer.
HisAlarmForwardTarget	DataForwardBaseWithCache	HisAlarmForwardProperty	None	Historical alarm entry, alarm level filtering, cache pruning.
ModbusSlave	DataForwardBase	ModbusSlaveProperty	ModbusSlaveVariableProperty	Modbus slave station memory mapping, channel properties, variable addresses, RPC write permissions.
MqttClientProducer	DataForwardBaseWithCacheIntervalScriptAll	MqttClientProducerProperty	MqttClientProducerVariableProperty	MQTT client upload, topic/payload template, subscription RPC write.
MqttServerProducer	DataForwardBaseWithCacheIntervalScriptAll	MqttServerProducerProperty	MqttServerProducerVariableProperty	MQTT Server upload, client connection management, topic/payload templates.
ThingsBoardClientProducer	DataForwardBaseWithCacheInterval	ThingsBoardClientProducerProperty	ThingsBoardClientProducerVariableProperty	ThingsBoard Gateway protocol, telemetry upload, device connection/disconnection, RPC Write.
KafkaProducer	DataForwardBaseWithCacheIntervalScriptAll	KafkaProducerProperty	DataForwardVariableProperty	Kafka Producer initialization, TopicArray upload, concurrency, and caching.
RabbitMQProducer	DataForwardBaseWithCacheIntervalScriptAll	RabbitMQProducerProperty	DataForwardVariableProperty	RabbitMQ connection, Exchange/RoutingKey, TopicArray upload.
Webhook	DataForwardBaseWithCacheIntervalScriptAll	WebhookProperty	DataForwardVariableProperty	HTTP Webhook, request method, header, payload template, failure buffer.
SyncBridge	DataForwardBase, IRpcDriver	SyncBridgeProperty	SyncBridgeVariableProperty	Gateway synchronization bridge, variable change queue, RPC invert proxy.
OpcUaServer	DataForwardBase	OpcUaServerProperty	OpcUaServerVariableProperty	OPC UA server nodes, certificates, security policies, variable write permissions.
IEC104Slave	DataForwardBase	IEC104SlaveProperty	IEC104SlaveVariableProperty	IEC104 Slave Station, Telesignal/Telemetry/Bit Array Mapping, Channel Properties.
IEC61850Server	DataForwardBase	IEC61850ServerProperty	IEC61850ServerVariableProperty	IEC61850 Server modeling, dataset, write index, object type conversion.
ZeroMQProducer	DataForwardBaseWithCacheIntervalScriptAll	ZeroMQProducerProperty	DataForwardVariableProperty	ZeroMQ Push/Pub/Dealer upload, binding mode, TopicArray, and caching.

Plugin Type References

Types	Recommended References
External Message System Uploads	MqttClientProducer,KafkaProducer,RabbitMQProducer, ZeroMQProducer.
HTTP/REST push	Webhook.
Industrial protocol server	ModbusSlave,IEC104Slave,OpcUaServer, IEC61850Server.
Database storage	HisDataForwardTarget,RealDataForwardTarget, HisAlarmForwardTarget.
Platform-specific protocol	ThingsBoardClientProducer.
Gateway bridging and write-back proxy	SyncBridge.

Common Development Errors

Errors	Consequences	Correct Practices
When connecting to external systems with constructors,	attributes have not yet been injected, so logs and cancel tokens are not available.	Connect to ProtectedStartAsync, and client configuration to ProtectedInitAsync.
TargetProperties and TargetPropertyType	Misalignment of form fields and runtime objects, import and export exceptions.	Both always correspond to the same strong type attribute.
Treat the target variable attributes as range filtering	Variable changes will not enter the target or troubleshoot direction errors.	The range of variables is determined only by the forwarding group, and the target variable attributes are only mapped externally.
Cached targets subscribe to global variable events	Repeated sending, batch processing, and trigger mode failure.	After inheriting the cache base class, use AcceptProduced* and Update*Model.
Send failed but still returns success	Offline cache cannot take over, data is lost and logs are misleading.	Send failed Return failure OperResult.
IsConnected always returns true	frontend state and cache reissue distortion judgments.	Returns based on the actual state of the client, socket, server, or channel.
Ignore CancellationToken	Freezes when stopping targets, switching redundancy, or refreshing configurations.	All connections, sending, waiting, and resending pass cancel tokens.
Do not close client/server	port occupancy, duplicate connections, or file cache not released when releasing.	SafetyDisposeAsync, then call the base.
Uploading templates does not release TopicArray	High-frequency uploads cause memory pressure.	Use the parent class UpdateTopicArrays and do not bypass its release logic.

Validation Checklist

Scenario	Validation Point
Plugin Discovery	Plugin Management and Data Forwarding Target dropdown shows plugins of the data forwarding type.
Attribute Forms	Target Properties, Variable Properties, Default Values, Certificates, Enumerations, and Import/Export are all correct.
Range Analysis	In Manual, All, CollectDevice, CollectGroup modes, the GetVariables() quantity meets expectations.
Change triggers	Single point change can enter the target, group variable participates group triggers, update mode takes effect.
Cycle trigger	cycle snapshots are executed by group interval, batch mode, and maximum batch execution.
External connection	When the target connection fails, LastErrorMessage is clear; after recovery, reconnection or restart can be successful.
Offline Caching	When the external system is disconnected, failed data enters the cache and is reissued batch after recovery.
Write-Back/RPC	Server protocol or platform RPC write-back can be applied to the collected variable and is rejected when permission is disabled.
Restart the target	Modify property saving, no duplicate subscriptions after multiple start-stops, port occupancy, or repeated sending.
Redundancy Targets	When redundancy is enabled, master-standby switching does not lose runtime references, and failed targets can be cleaned up.