Is this the correct DefaultFunctionsPipeline behavior

[majidalikhan8baader]

We are writing a custom application service and in this application service we want to process a data and afterwards send it to the MQTT.
For this we have the following code

func (app *mqttApp) createTopicSpecificPipeline(functionsLibrary functions.FunctionsLibrary, mqttConfig transforms.MQTTSecretConfig, pipelineConfig map[string]string) error {
	log.Printf("CreateTopicSpecificPipelineLoop for resource '%s'", mqttConfig.Topic)
	err := app.service.SetDefaultFunctionsPipeline(
		functionsLibrary.TransformEvent(mqttConfig),
		transforms.NewMQTTSecretSender(mqttConfig, false).MQTTSend,
	)
	if err != nil {
		return err
	}
	return nil
}

As per my understanding, the result of functionsLibrary.TransformEvent(mqttConfig), is processed and sent to transforms.NewMQTTSecretSender(mqttConfig, false).MQTTSend which then sends the information to the Broker.

And if that's the case, I can change the mqttConfig on the fly in the TransformEvent function such as updating the topic name based on the data I have received from the event. But, it just doesn't update and while debugging I found out that my pipeline first goes into this part transforms.NewMQTTSecretSender(mqttConfig, false).MQTTSend and afterward when the connection is established it goes to the above method and processes it and forwards the result to the MQTTSend.

I tried achieving this via PipeLinePerTopic but there, each time I send an event a new connection to the MQTT is created. How do I avoid doing that, how do I use the default application to update my mqttConfig variable in the TransformEvent and later just use it and then avoid using PipelinePerTopic for the separate topic publication on MQTT.

[ejlee3]

It is possible to specify a placeholder in the publish topic format in the MQTT Config. This topic can then be filled in by adding a key-value pair to the context where the key is the custom placeholder and the value is the custom value to add to the topic. Information for trigger topic placeholders is here.

For a more advanced use case, you should be able to use a custom topic formatter and call transforms.NewMQTTSecretSenderWithTopicFormatter. The documentation for the builtin function is here.

[majidalikhan8baader]

@ejlee3 This is wonderful, are there any examples available? I have tried the material you pointed to but am still unable to get into the topic. From the looks of it, once I NewMQTTSecretSenderWithTopicFormatter I should be able to solve the problem where my mqtt connection drops. But, more examples of its usage would be great. Thanks

[jiekechoo]

@majidalikhan8baader @ejlee3 I have the same issue in my app service. edgexfoundry/app-functions-sdk-go#1558
Any suggestion for publishing multiple topics in single mqtt connection?