Implement Request/Response flow in MQTT

[dilanSachi]

Description:
$subject

[dilanSachi]

In this feature, a mqtt subscriber has the option of responding to any message sent from a publisher. The publisher sends a metadata as response_topic in its message which the subscriber will respond to. Publisher listens to the response_topic and will get the response sent by the subscriber. In order to specifically identify the response messages, theres a correlation_id that can be set in the message which the subscriber will return.

There is also a concept of sending response_topic_info in the CONNACK packet. CONNACK is sent by the broker when a client is connected to it. From this, response_topic_info, publisher can build a response_topic such that there will be no auth issues when publishing to topics (assume response_topic_info -> rtopic. Publisher can build a response topic as /rtopic/123. Broker provides write access to all /rtopic/#).

[dilanSachi]

Image taken from https://www.emqx.com/en/blog/mqtt5-request-response

[dilanSachi]

In this,
Publisher -> send and listen
Subscriber -> listen and reply

Publisher side
In java, there’s an api to publish. And a separate callback to subscribe and listen.

Option - 1
We can add a return type to the publish api so that publisher->publish will block until response is received.
However user cannot do anything until a response is received.

Nats also have a similar structure like this. However the java api of nats (send) returns the response directly rather than to a callback. It doesn't have a concept of correlation id where it's user's responsibility to handle the appropriate response to specific publish.

If we try to implement it similar to this, it will block users' operation flow.

Option - 2
Have a separate api to receive messages.
publisher->receive
This will return a stream(stream<mqtt:Message, error>)(like grpc bidirectional streaming).
Publisher can process messages one by one in the receiving order (same as java client).

Subscriber side
In the subscriber side, we can provide an api for the caller to send the response to the response topic.

    remote function onMessage(mqtt:Message message, mqtt:Caller caller) returns error? {
        log:printInfo(check string:fromBytes(message.payload));
        check caller->publish("This is the response");
    }

Caller will automatically grab the response topic and publish to that topic.
The onComplete method is there to get notified when the receiver receives the message and acknowledges it.

@shafreenAnfar WDYT?