¶ (c) Technocore - All Rights Reserved.
NOTICE: All information contained herein is, and remains
the property of TechnoCore.
The intellectual and technical concepts contained
herein are proprietary to TechnoCore and dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from TechnoCore.
¶ ObjMqtt
The ObjMqtt class provides a robust client for connecting to an MQTT broker, subscribing to topics, and processing incoming messages. Its primary function is to listen for event messages, dynamically parse their JSON payloads, and store the data in a relational database in a structured format.
A key feature of this class is its ability to automatically manage the database schema based on the content of the messages it receives.
¶ Dynamic Schema Generation
When an event message is received, the ObjMqtt class performs the following actions to store its data:
-
Table Name Determination: It constructs a table name based on the
event_typefield in the message payload. The format istrack_event_<event_type>. For example, anevent_typeofuser_loginwill result in a table namedtrack_event_user_login. -
Dynamic Column Creation: The class inspects the top-level keys within the
event_stateJSON object of the payload. For each key, it ensures a corresponding column exists in the target table. -
Data Type Inference: The system automatically determines the appropriate data type for each new column:
- If the value associated with a key is a dictionary (
dict) or a list (list), the column is created with theJSONdata type. - For all other data types (strings, numbers, booleans), the column is created with the
TEXTdata type.
- If the value associated with a key is a dictionary (
-
Data Insertion: The message data is then inserted as a new row in the table.
This process ensures that the database schema evolves automatically with the data, requiring no manual intervention when new fields are introduced in the MQTT messages.
¶ Standard Columns
Every table created by the ObjMqtt class includes the following standard columns:
Guid: A unique identifier for the event.Package: The software package or service that generated the event.SourceHost: The hostname or IP address of the machine where the event originated.Payload: AJSONcolumn used as a fallback to store theevent_stateif it is not a dictionary.created_at: A timestamp indicating when the record was inserted.
¶ Example Workflow
Consider an MQTT message published to the event/system topic with the following JSON payload:
{
"guid": "evt-12345-abcde",
"package": "auth_service",
"host": "prod-server-01",
"event_type": "user_login",
"event_state": {
"username": "jdoe",
"ip_address": "192.168.1.100",
"successful": true,
"metadata": {
"client": "WebApp",
"version": "2.5.1"
}
}
}
- The class subscribes to the topic and receives this message.
- It determines the target table name is
track_event_user_login. - It checks if this table exists. If not, it creates it with the standard columns.
- It inspects the keys in
event_state:username,ip_address,successful, andmetadata. - It ensures columns with these names exist in the
track_event_user_logintable.username,ip_address, andsuccessfulwill beTEXT.metadatawill beJSON, because its value is a dictionary.
- It inserts a new row with the corresponding values.
¶ Batch Processing
To optimize database performance, messages are processed in batches. The class collects messages in memory and inserts them into the database in bulk, either when the batch reaches a certain size (batch_size) or when a timeout (batch_timeout) is reached.
¶ MQTT Callback API
ObjMqtt uses the paho-mqtt VERSION2 callback API. The _on_connect
method signature is:
def _on_connect(self, client, userdata, connect_flags, reason_code, properties):
if reason_code == 0:
# connected — subscribe to topics
else:
# failed — reason_code gives a human-readable description
reason_code is a paho ReasonCode object; reason_code == 0 tests for
success and str(reason_code) gives a descriptive error string on failure.
¶ Command-Line Usage
The script can be run as a command-line tool to monitor MQTT topics directly.
python factory.core.ObjMqtt.py [TOPIC]
- TOPIC: The MQTT topic to subscribe to. It defaults to
#to subscribe to all topics.
Example:
# Monitor all topics
python factory.core.ObjMqtt.py
# Monitor a specific topic
python factory.core.ObjMqtt.py "axion/events/system"