MQTT Publish/Subscribe Script Manual

Scripts use Lua syntax. First read MQTT cloud-edge coordination for overall concepts and the roles of connection, publish, and subscribe configurations. This page details script write parameters, complete examples, and request/response message formats.

Overview

1. Customize MQTT publication formats and configure CRON tasks for data uploads.
2. Customize subscription-message parsing; the Edge-EMS Builder backend writes specified points based on parsed results.
3. Choose whether processing a subscription message requires a response and customize its topic and content.

Note: users need basic programming knowledge and familiarity with MQTT publish/subscribe. For script syntax, see the complete Lua script reference.

An unavailable referenced point is injected as the sentinel string __Null__. Explicitly check ${dataKey} == "__Null__"; do not assume an empty string or nil.

MQTT connection

Parameter	Value	Required	Type	Description
mqttConnectionConfigName	Connection configuration name	Yes	String	MQTT connection configuration name
status	Valid	Yes	String	Status: whether this connection is enabled
protocol	mqtts	Yes	String	MQTT protocol: mqtt/mqtts
protocolVersion	3.1.1	Yes	String	MQTT version: 3.1/3.1.1/5.0
serverHost	127.0.0.1	Yes	String	MQTT server address
serverPort	8883	Yes	String	MQTT server port
clientId	return "demo-edge"	Yes	String	Lua expression returning the client ID
username	username	Yes	String	Username
password	password	Yes	String	Password
cleanSession	true	Yes	Boolean	Clean session
sessionExpiryInterval	0	Yes	Number	Session expiry interval, specific to MQTT 5.0
keepAliveInterval	60	Yes	String	Keep-alive interval
connectTimeout	30	Yes	String	Connection timeout
sslEnable	true	Yes	Boolean	Whether to use SSL

MQTT data publishing

Parameter	Value	Required	Type	Description
mqttConnectionConfigId	100101	Yes	Number	MQTT connection configuration ID
mqttPublishConfigName	Publish configuration name	Yes	String	MQTT publish configuration name
topic	return "demo/telemetry"	Yes	String	Lua expression returning the topic
qos	“1”	Yes	String	QoS string enumeration
publishStrategy	0/10 * * * * ?	Yes	String	Publication schedule (CRON expression)
publishContent	{“name”: “dataKey”}	Yes	String	Publication content (Lua script)
note	note	Yes	String	Notes

The publication path serializes script results as JSON. To publish an object, return a Lua table directly. Calling toJsonString beforehand produces a quoted JSON string.

publishContent example 1: construct a Table

local map = {}
-- Time
map.time = OffsetDateTime.now():unixMilli()
-- Ordinary data value
map.name = "${数据8}"
-- String array
map.strList = { "${数据3}", "${数据4}" }
-- Numeric array
local d3 = ${数据3}
local d4 = ${数据4}
map.numberList = { d3, d4 }

return map

publishContent example 2: string concatenation

local json = '{"strList":["${数据3}","${数据4}"],"name":"${数据8}","time":' .. OffsetDateTime.now():unixMilli() .. ',"numberList":[' .. ${数据3} .. ',' .. ${数据4} .. ']}'
return jsonToMap(json)

Parsed result

{
    "strList": [
        "3.6",
        "0.6"
    ],
    "name": "2",
    "time": 1742180870064,
    "numberList": [
        3.6,
        0.6
    ]
}

MQTT message subscriptions

Parameter	Value	Required	Type	Description
mqttConnectionConfigId	2	Yes	Number	MQTT connection configuration ID
mqttSubscribeConfigName	Subscribe configuration name	Yes	String	MQTT subscribe configuration name
status	Valid	Yes	String	Status
topic	return "demo/commands"	Yes	String	Lua expression returning the topic
qos	“1”	Yes	String	QoS string enumeration
actionContent	{}	Yes	String	Action content (Lua script)
needResponse	true	Yes	Boolean	Whether a response is required
responseTopic	return "demo/replies"	When responses are enabled	String	Lua expression returning the response topic
response	return {ok=true}	When responses are enabled	String	Response script: strings are sent unchanged; tables are serialized as JSON
note	note	Yes	String	Notes

actionContent: parse the subscribed message

Write parameters

Parameter	Value	Required	Type	Description
dataType	Customization	Yes	String	Data type to operate on: Customization/Collection/Aggregation
dataKey	功率	Yes	String	Unique dataKey
dataValue	topic	Yes	String	Data value
availableDate	2023-01-15T07:00:00+08:00	No	String	Effective date (Customization)
expiredDate	2023-01-15T10:00:00+08:00	No	String	Expiration date (Customization)

Fixed parameter:

${subscribe_message}: message received from the subscribed topic.

Do not configure dataKey values that overlap with the fixed dataKey above when using Edge-EMS Builder.

actionContent parsing

local messageMap = jsonToMap(${subscribe_message})
local params = messageMap.params
local strategyList = jsonToListMap(toJsonString(params.strategy))
local date = params.date
local resultList = {}

for i, keyMap in ipairs(strategyList) do
    local availableDate = completeDateTime(date .. "T" .. keyMap.start)
    local expiredDate = completeDateTime(date .. "T" .. keyMap["end"])

    -- Data type to write: Collection (raw) / Customization (time-scheduled) / Aggregation (aggregated)
    local modeMap = buildCommonWriteMap("Customization", "mode", tostring(keyMap.mode), availableDate, expiredDate, false)
    local chargeMap = buildCommonWriteMap("Customization", "charge", tostring(keyMap.charge), availableDate, expiredDate, false)
    local valueMap = buildCommonWriteMap("Customization", "value", tostring(keyMap.value), availableDate, expiredDate, false)

    table.insert(resultList, modeMap)
    table.insert(resultList, chargeMap)
    table.insert(resultList, valueMap)
end

table.insert(resultList, buildSlaveWriteMap("DeviceB", "Share", "数据4", 777, nil, nil, false))

return resultList

Request example

/rpc/${iot_code}/${method}/request

Subscription topic: /rpc/yanhua001/ems/strategy/request

Note: iot_code is aggregation data.

Sent message

{
  "jsonrpc":"2.0",
   "id":"d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9",
   "method":"ems/strategy",
   "params":{
    "date":"2023-1-15",
    "strategy":[
     {
      "strategyDate":"2023-1-15",
      "start":"00:00:00",
      "end":"05:00:00",
      "mode":1,
      "charge":true,
      "value":100
      },
     {
      "start":"05:00:00",
      "end":"07:00:00",
      "mode":2,
      "charge":false,
      "value":100
     }
    ]
  }
 }

Parsed action result

Requirements for parsed results:

1. dataType must be Collection (raw data), Customization (time-scheduled data), or Aggregation (aggregated data).
2. dataKey must exist.
3. If a Customization dataKey has multiple values, their effective intervals must not overlap.
4. Updating Customization data completely removes existing data and replaces it with the newly supplied data.

{
    "actionType": "SUBSCRIBE",
    "actionResult": true,
    "mqttActionContent": {
        "Customization": {
            "mode": [
                {
                    "dataType": "Customization",
                    "dataKey": "mode",
                    "dataValue": "1",
                    "availableDate": "2023-01-15T00:00:00+08:00",
                    "expiredDate": "2023-01-15T05:00:00+08:00"
                },
                {
                    "dataType": "Customization",
                    "dataKey": "mode",
                    "dataValue": "2",
                    "availableDate": "2023-01-15T05:00:00+08:00",
                    "expiredDate": "2023-01-15T07:00:00+08:00"
                }
            ],
            "charge": [
                {
                    "dataType": "Customization",
                    "dataKey": "charge",
                    "dataValue": "true",
                    "availableDate": "2023-01-15T00:00:00+08:00",
                    "expiredDate": "2023-01-15T05:00:00+08:00"
                },
                {
                    "dataType": "Customization",
                    "dataKey": "charge",
                    "dataValue": "false",
                    "availableDate": "2023-01-15T05:00:00+08:00",
                    "expiredDate": "2023-01-15T07:00:00+08:00"
                }
            ],
            "value": [
                {
                    "dataType": "Customization",
                    "dataKey": "value",
                    "dataValue": "100",
                    "availableDate": "2023-01-15T00:00:00+08:00",
                    "expiredDate": "2023-01-15T05:00:00+08:00"
                },
                {
                    "dataType": "Customization",
                    "dataKey": "value",
                    "dataValue": "100",
                    "availableDate": "2023-01-15T05:00:00+08:00",
                    "expiredDate": "2023-01-15T07:00:00+08:00"
                }
            ]
        }
    },
    "mqttActionResult": {

    }
}

responseTopic: resolve the response topic

Fixed parameters:

${subscribe_message}: message received from the subscribed topic.

${subscribe_result}: whether actionContent executed successfully.

${subscribe_result_message}: actionContent execution result, such as the parsed action result.

Do not configure dataKey values that overlap with the fixed dataKeys above when using Edge-EMS Builder.

responseTopic parsing

local messageMap = jsonToMap(${subscribe_message})
local id = messageMap.id
local method = messageMap.method
return "/rpc/" .. "${iot_code}" .. "/" .. method .. "/response/" .. id

Request example

/rpc/${iot_code}/${method}/response/${id}

Target response topic: /rpc/yanhua001/ems/strategy/response/d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9

Sent message

{
  "jsonrpc":"2.0",
   "id":"d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9",
   "method":"ems/strategy",
   "params":{
    "date":"2023-1-15",
    "strategy":[
     {
      "strategyDate":"2023-1-15",
      "start":"00:00:00",
      "end":"05:00:00",
      "mode":1,
      "charge":true,
      "value":100
      },
     {
      "start":"05:00:00",
      "end":"07:00:00",
      "mode":2,
      "charge":false,
      "value":100
     }
    ]
  }
 }

Parsed topic result

/rpc/yanhua001/ems/strategy/response/d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9

response: construct the response

response parsing

local messageMap = jsonToMap(${subscribe_message})
local id = messageMap.id
local jsonrpc = messageMap.jsonrpc
local result = "${subscribe_result}"

local resultMap = {}
resultMap.jsonrpc = jsonrpc
resultMap.result = result
resultMap.id = id

return resultMap

Request example

/rpc/${iot_code}/${method}/response/${id}

Target response topic: /rpc/yanhua001/ems/strategy/response/d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9

Sent message

{
  "jsonrpc":"2.0",
   "id":"d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9",
   "method":"ems/strategy",
   "params":{
    "date":"2023-1-15",
    "strategy":[
     {
      "strategyDate":"2023-1-15",
      "start":"00:00:00",
      "end":"05:00:00",
      "mode":1,
      "charge":true,
      "value":100
      },
     {
      "start":"05:00:00",
      "end":"07:00:00",
      "mode":2,
      "charge":false,
      "value":100
     }
    ]
  }
 }

Parsed topic result

Note: the actual output is compact JSON.

{
    "result": "true",
    "id": "d1acc980-0e4e-11e8-98f0-ab5030b47df4:d1db7aa0-0e4e-11e8-b1d9-5f0ab230c0d9",
    "jsonrpc": "2.0"
}

Next steps

• Overall MQTT cloud-edge concepts: connection/publication/subscription configuration and transmission after network outages: MQTT cloud-edge coordination.
• Complete Lua built-in function and module reference: Complete Lua script reference.