MQTTExport with dynamic topics

[majidalikhan8baader]

I have a pipeline that generates specific topics based on a key-value pair and sends data to that format on MQTT Broker. But, connection to the broker is lost and connected after each call. This causes data loss. Following is the implementation.

functionsLibrary := functions.NewFunctionsLibrary()
	mqttConfig := app.serviceConfig.AppServiceMQTT.MqttExportConfig
	eventMapping := app.serviceConfig.AppServiceMQTT.EventMapping

	err := app.createTopicSpecificPipelineParallel(functionsLibrary, mqttConfig, eventMapping)

// Default and sequential implementation of the pipelines
func (app *mqttApp) createTopicSpecificPipeline(functionsLibrary functions.FunctionsLibrary, mqttConfig transforms.MQTTSecretConfig, eventMapping map[string]string) error {
	log.Printf("CreateTopicSpecificPipelineLoop for resource '%s'", mqttConfig.Topic)
	for key, value := range eventMapping {
		mqttConfig.Topic = key
		err := app.service.AddFunctionsPipelineForTopics(
			key,
			[]string{value},
			functionsLibrary.TransformEvent(),
			transforms.NewMQTTSecretSender(mqttConfig, false).MQTTSend,
		)
		if err != nil {
			return err
		}
	}
	return nil
}

Any examples or ideas on how can I achieve the same task in different ways? I have asked this question before and got an answer that can be resolved with NewMQTTSecretSenderWithTopicFormatter but I haven't found any examples or proper guide anywhere for that. It might be possible that I didn't grasp the whole idea behind it because of my lack of knowledge and being a beginner with EdgeX.

Any help? Appreciated.

[ngjaying]

You can try rule engine eKuiper, it supports dynamic property for MQTT sink.

Getting started: https://ekuiper.org/docs/en/latest/edgex/edgex_rule_engine_tutorial.html#overview
Sample rule to send data to dynamic topic:

curl -X POST \
  http://$kuiper_server:59720/rules \
  -H 'Content-Type: application/json' \
  -d '{
  "id": "rule1",
  "sql": "SELECT *, meta(id) as eventId FROM demo",
  "actions": [
    {
      "mqtt": {
        "server": "tcp://broker.emqx.io:1883",
        "topic": "{{.eventId }}",
        "clientId": "demo_001"
      }
    }
  ]
}'

[majidalikhan8baader]

eKuiper isn't gonna be of any help as we need to process data before pushing it out to the MQTT, functionsLibrary.TransformEvent() is the one where we process and transform the event.

[ngjaying]

Indeed, it is used to process data before pushing to any sink like MQTT.

[majidalikhan8baader]

And I want the data to be pushed to MQTT but each data that is processed from the eKuiper is first pushed to MessageBus (Redis) where I read it and perform some operations and now, after that I want to push the extra processed data to MQTT on different topics, this is where I am having problems. As, if I change the topic names on the fly like my example, I am unable to maintain connection to the broker thus, losing the data as well. I believe it's because the modification in the config object forces the functionspipeline to create a new connection each time and closing the old one.

Now, I am wondering if there's a way to do so without losing the connection. As of now, doesn't seem like there's. Maybe, I will have to create my own implementation which I want to avoid as it's extra work and doesn't seem like that it's some special case that I have to. I believe there must be a way, I just haven't found it yet, so asking here.

@ngjaying but I am really thankful that you took your time to answer.

[lenny-goodell]

The MQTT Export supports placeholders in the topic that are replaced with data from the Context Storage as the data is published.

The Event's profilename, devicename and sourcename are pre=populated in the Context Storage for every Event received and paced the the functions pipeline. You can also add custom data to the Context Storage in your transform function and use the key(s) for that data as placeholders in you topic.

See the example here for default topic format: https://docs.edgexfoundry.org/2.3/microservices/application/Triggers/#example

Simply uses this same default topic format for your MQTT Export topic.

[majidalikhan8baader]

I have seen the examples and documentations and I find myself stuck as there is no concrete example for the whole process of setting up the placeholders or topicFormatter.

Anything that gives the idea where we can use the use the context to store placeholders to be replaced later, and how do we use them would be a great help.

[majidalikhan8baader]

I achieved it with the help of a try-and-error approach. I suggest having examples in the example repo or providing the proper documentation with examples. That would be really helpful. I can also do this myself and provide an example that then can be merged into the repository.

[lenny-goodell]

I achieved it with the help of a try-and-error approach

Yes, MQTT Export docs need a clear example of setting the export topic using the place holders and using the context storage for custom placeholders
Yes, please submit a docs PR. we welcome contributions. TY

[lenny-goodell]

BTW, the custom Topic Formatter only allows you to change how the topic is format for the placeholders rather then the simple default formater of {placholder}