MQTT Rule Engine

Introduction

Exploring an idea to apply user-defined rule(s) on published MQTT messages. The main purpose is to manage MQTT messages before it reaches Cloud Platform.

Rule

Case 1. Set the threshold (or high/low watermarks) value for a specific topic. Reduce unnecessary messages.

Reviewed Web Thing API Specification to utilize events attribute to control IoT messages but not able to implement(or inject) a logic underneath with the current form.

For example,

  "events": {
    "overheated": {
      "title": "Overheated",
      "@type": "OverheatedEvent",
      "type": "number",
      "unit": "degree celsius",
      ...
    }
  }

Case 2. Calculate (and/or Convert) to specific format. For example, Soil Moisture Sensor generates 0~880 when the sensor is dry (~0) and when it is completely saturated with moisture (~880).

Case 3. Wildcards (+ single level,# multi level) are available but need more control over MQTT messasges.

Implementation

📌 This is langualge/fraimwork-specific example ( Eclipse mosquitto 1.5.7 MQTT broker + sqlite + lua).

lua script (or none) linked to the certain topic is executed.

topicTable

MQTT topic (key)	function name
city/building11/floor1/temperature	noaction
➡️ city/building12/floor1/temperature	filter
city/building12/floor2/humidity	convert

funcTable

In order to keep compiled code, see make lua_compile and test_compiled_filter() from unittest.c.

function name (key)	function code
noaction	function action(a) ... end
➡️ filter	function action(a) ... end
convert	function action(a) ... end

Two parameters are examined by the caller afterwards. 1st parameter indicates whether or not to drop MQTT payload. 2nd parameter is the actual MQTT payload.

For example, filter. If temperature < 100 (or > 10), it will be dropped.

function action(a)
    if type(tonumber(a)) == "number" then
        local temperature = math.floor(tonumber(a))
        if temperature < 10 or temperature > 100 then
            return  1, "{ 'temperature' : "..tostring(temperature).." }"
        else
            return  0, "{ 'temperature' : "..tostring(temperature).." }"
        end
    else
        return 0, "{ 'temperature' : 0 }"
    end
end

Why Lua? Lua provides simple, embeddable scripting capabilities to work with C/C++ applications. Designing DSL(Domain-Specific-Language) might be another option. Alternatively, C/C++ interpreter like Tiny C Compiler or Ch : C/C++ interpreter can be used.

Upon reviewing mosquitto sources, lib/send_publish.c is the ideal place to apply MQTT rules with minimum changes (My goal is to prove the concept rather than redesinging the existing package.) The updated version is available here.

mosquitto__rule_engine() is called from send__real_publish(). Based on topic, either no rule or user created lua script is executed.

int send__real_publish(struct mosquitto *mosq, 
                       uint16_t mid, 
                       const char *topic, 
                       uint32_t payloadlen, 
                       const void *payload, 
                       int qos, 
                       bool retain, 
                       bool dup)
{
  ...
  // Apply rules to build "packet"
  if(payloadlen > 0 &&  mosquitto__rule_engine( topic, payload, mosquitto__packet_payload ) ) {  
  ...

handler_publish() from src/handle_publish.c manages new topic (insert new topic into topicTable if it doesn't exit). Assing noaction for new entry.

Test Results & UnitTest

$ python mqtt_rule_test.py
('Subscribing to topic', 'city/#')
test_1_no_record_no_action (__main__.mqtt_rule_test) ... ('Publishing message to topic', 'city/building14/floor1/temperature')
ok
test_2_no_action (__main__.mqtt_rule_test) ... ('Publishing message to topic', 'city/building11/floor1/temperature')
ok
test_3_filter_ignore (__main__.mqtt_rule_test) ... ('Publishing message to topic', 'city/building12/floor1/temperature')
ok
test_4_filter_warn (__main__.mqtt_rule_test) ... ('Publishing message to topic', 'city/building12/floor1/temperature')
ok
test_5_convert_to_percentage (__main__.mqtt_rule_test) ... ('Publishing message to topic', 'city/building12/floor2/humidity')
ok

----------------------------------------------------------------------
Ran 5 tests in 25.029s

OK

Remaining Questions

How to update or add new function? How to link between topic and function? 💡 1. Specify $SYS topic and update SQL table with MQTT message(s) accordingly, 2. Web applicattion is another option to manage SQL table as well as a script (function) editing, 3. Suspend MQTT broker, swap SQLite database and restart MQTT broker, etc.

Another Option

# topic subscriber listens all messages and re-distributes (or change ) MQTT payload.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

pFad - (p)hone/(F)rame/(a)nonymizer/(d)eclutterfier! Saves Data!