Pass MQTT message metadata to/from application plugin

[feymartynov]

MQTT allows passing properties along with message payload. This is somewhat like HTTP headers. There is a number of standard properties such as response_topic (a topic where the response message has to be sent) and correlation_data (some binary to match the response message with its original request). Also MQTT 5 introduces user properties – custom key-value string pairs. All of these may be significant for the application plugin logic but currently we don't pass this information into it. Also currently MQTT transport can only send messages to the topic from the config making usage response_topic property on purpose impossible.

The solution is to merge all this metadata under __metadata key in the request payload so the application plugin can get it. And vice versa: when the application plugin responds if the payload contains __metadata key then the transport plugin parses this metadata, sets properties from there and possibly overrides the destination topic from config + QoS and retain flags.

All of this is optional so back compatibility is preserved. Also this will not affect other transports: we still can call the application plugin over HTTP or anything else.

[lminiero]

Sorry but I can't merge this. This breaks the opaque nature of transports and plugins, which should never have to worry (or care) which protocol is used to carry messages and events. The moment we open the door to one protocol, people will ask for more, and soon we'll end up with tons of conditions for different messages for different targets, that I have no intention to facilitate.

[feymartynov]

OK, got it. Actually I could do my task by only proxying some properties from request to response. Is there any proper way to do it on the transport level and avoid passing them through the application plugin?

[lminiero]

In the transport API, you can pass a pointer to something you own to incoming_request:

https://github.com/meetecho/janus-gateway/blob/master/transports/transport.h#L240

The same value will be passed back to you when you get a send_message that is a response to that request:

https://github.com/meetecho/janus-gateway/blob/master/transports/transport.h#L210

That's what we use for passing the correlation ID back and forth in the RabbitMQ transport, for instance. That should allow you to do what you need with metadata: closing this PR in the meanwhile.

[feymartynov]

I would also like to measure what time it takes for the request to be processed in the application plugin and set it as a user property in the response. Is it OK if I implement this logic in the transport?

[feymartynov]

@lminiero I have utilized request_id argument to pass properties from the request to the response like RabbitMQ transport does for correlation id but there's a problem. In case of "janus": "message" request Janus passes request_id to send_message transport callback only for ack response. The actual response from the application plugin goes as event and request_id is NULL in that case. How can I get properties from the request on the transport level?

[lminiero]

There's no way to correlate asynchronous events at the transport level, and neither should there be. The only way to do correlation in that case is at the Janus API level, where the transaction ID may or may not be part of the event (that's entirely up to plugins).

[feymartynov]

@lminiero Transaction ID is up to the client so we can't pass anything with it. Would it be OK if I add a key-value state in the context where keys are transaction IDs and values are request metadata structs?

[lminiero]

Not sure why you need that metadata. As I said, most events will not include a transaction ID at all, very few plugins add one: when they do, that's what applications can use to relate an event to a request they made before. I see no reason why the transport should be aware of that.

[feymartynov]

I need it for the following:

• Overriding default publish topic from config by the value from response_topic property if present which is MQTT 5 standard.
• Setting correlation_data from the request if present which is also a standard property.
• Optionally proxying custom user properties from the request.
• Optionally setting processing time property => I have to remember the time when the processing started somewhere.

[feymartynov]

@lminiero If transaction ID is not set by the application plugin then it means that the event is not a part of the transaction i.e. it's not a response for a request so neither of the above is applicable in that case.

By the way all of these features when done on the transport level extend only MQTT transport and do not affect other transports or application plugins.

[feymartynov]

@lminiero Could you confirm/comment on this so I wouldn't come up with another worthless PR? :)

[lminiero]

An event containing a transaction ID is not, strictly speaking, a response either. The only responses are either a direct response (synchronous request) or the ack (asynchronous request). Events containing a transaction ID can be considered updates on that request, not responses, but even then they don't work the same way in all plugins: in the VideoRoom, for instance, you typically only receive a single update to a request; in the SIP plugin, you'll receive the same transaction ID on multiple events, even those not related to the original request.

So again, I really don't understand why the transport protocol, MQTT in this case, should know anything about that. All the correlation information is already available in the Janus API messages themselves. We have a RabbitMQ-based application that uses three plugins at the same time, and request/response/event management works just fine without extra functionality: we only need the correlation ID to know when we get a response, and after that events can be handled without that.

If you override the publish_topic won't this cause backwards compatibility issues to applications using the Janus MQTT stack as it exists today?

[feymartynov]

Let's consider an application plugin that handles requests but can't use direct response because processing time may be significant so that blocking the incoming message processing thread for such a time is not a good idea.

So the plugin sends direct ack response immediately and does all the work asynchronously in another thread. When it's finished it pushes a single event which contains the actual response data.

For correlating this event with the initial request on the application plugin level it copies the transaction ID like you said.

However on the transport level we also need to correlate because MQTT 5 standard tells us that correlation_data property must be copied. So we have to pass this value somehow from the request to the event message.

We don't want the application plugin to know anything about transport-specific stuff to keep application plugins isolated from transports so passing it through the application plugin like I originally proposed in this PR is not an option.

The new proposal for passing correlation_data along with other metadata is to introduce a key-value state in the transport's context. When the transport receives a request it saves its metadata into the state under transaction ID key. When an event with this transaction ID is being pushed by the application plugin the transport looks into the state, gets metadata and copies correlation_data as well as other stuff from it into the event's message properties.

@lminiero What do you think?