Add MQTT Entity Sink Feature: Flat Topic Streaming feat: #535

[virajdere]

Adds feature #535

This pull request addresses #535 and introduces the MQTT Entity Sink feature to the cppagent, enabling observations to MQTT brokers using a flat, per-data-item topic structure.

Key changes include:

• New MqttEntitySink class and integration with agent configuration.
• Support for all relevant MQTT connection, authentication, and topic options.
• Documentation and configuration examples for agent setup.

Checklist

• Existing sinks unchanged
• All tests passing/No failures
• Builds cleanly
• Added additional tests for the feature
• Existing features remain unchanged/unaffected

Future development directions:

• Identify and fix any performance bottlenecks for the Entity format sink (currently no performance issues identified for brokers emqx/moquette as a part of initial testing)
• Add support for persistent buffer for all the data during broker disconnections/restarts.
• All persisted data should be relayed once network is restored
• Add support for durableRelay which allows users to configure whether they want to relay the persisted data
• This feature will be improved further with additional asset and device publishing

Test Config
Sinks {
  MqttEntitySink {
    # MQTT Broker Connection
    MqttHost = localhost
    MqttPort = 1883
    
    # MQTT Authentication (optional)
    # MqttUserName = mtconnect
    # MqttPassword = password123
    
    # MQTT Client ID (optional, auto-generated if not specified)
    # MqttClientId = client_id
    
    # QoS Levels:
    # 0 = At Most Once (fire and forget)
    # 1 = At Least Once (acknowledged delivery) - DEFAULT
    # 2 = Exactly Once (guaranteed delivery)
    # MqttQOS = at_least_once
    
    # Retain Messages
    # When true, broker retains last message for new subscribers
    # MqttRetain = true
    
    # Topic Structure Configuration
    # These use [device] placeholder which gets replaced with device UUID
    ObservationTopicPrefix = MTConnect/Devices/[device]/Observations
    DeviceTopicPrefix = MTConnect/Probe/[device]
    AssetTopicPrefix = MTConnect/Asset/[device]
    MqttLastWillTopic = MTConnect/Probe/[device]/Availability
    
    # TLS Configuration (if needed)
    # MqttTls = true
    # MqttCaCert = GroupCACertificate.pem
    # MqttCert = certificate.pem
    # MqttPrivateKey = privateKey.pem
  }
}

@wsobel

[wsobel]

I'll review the code and test locally.

[wsobel]

Reviewing issues with build in CI/CD process. Creating new branch on cppagent for testing.

[virajdere]

Reviewing issues with build in CI/CD process. Creating new branch on cppagent for testing.

issues probably with the newly added test file mqtt_entity_sink_test.cpp or the underlying logic in mqtt_entity_sink. a bad use of shared pointers? I am currently not able to figure out the issue. a robust review and some code update will be helpful. Feel free to make any changes necessary.

[wsobel]

There were a few issues I can help you fix. I’ll send some more comments soon.

[wsobel]

I have created a patch file that should fix the issue with the tests. The issue was the client was being closed after the test had exited and the local objects were not there. I just added a method to close the client before the function exits.

I also ran it through clangformat, so some of the changes are just whitespace related.

PR_571.patch

[wsobel]

There are still some issues. Using a heap sanitizer and allocation debugger to test.

[wsobel]

Fixed another timing issue when checking for unavailable to exclude conditions and the agent device.

PR_571_2.patch

[wsobel]

I created a local branch: feature/mqtt-entity-sink if you want to merge against a branch.

[wsobel]

Looks good. If it passes, I'll merge and update the version.

[wsobel]

Did you try the json entity printer that generates a similar representation for the observations. It also supports all the extended types like data sets and tables. I didn't see that code in the formatter.

What are your thoughts?

[wsobel]

You need to shut the agent down first. Otherwise there are issues. I can send patch.

[wsobel]

The callback is called during cleanup of the client. You need to either shut down the client or remove the handler before exiting this test. Otherwise the json object has been destructed and is no longer viable.