eclipse-paho / eclipse-paho/paho.mqtt.java

Topic filter does not match shared subscription, listeners are not notified on actual topic

Open
#965 2 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
2.3k
Forks
919
PR merge metrics
No merged PRs in 30d

Description

Please fill out the form below before submitting, thank you!

- [X] Bug exists Release Version 1.2.5 ( Master Branch)
- [ ] Bug exists in MQTTv3 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)
- [X] Bug exists in MQTTv5 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)

**TLDR**:
TopicValidator does not match shared subscription topics with actual message topics.
i.e. `$share/GROUP-NAME/some-topic` is not matched when a message comes from `some-topic`.

https://github.com/eclipse/paho.mqtt.java/blob/f4e0db802a4433645ef011e711646a09ec9fae89/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/common/util/MqttTopicValidator.java#L151

**Long description:**
I've encountered an issue using the MQTTv5 feature - shared subscriptions. In order to do a broker-side load-balancing between running application instances I have to subscribe to a topic with a prefix: `$share/GROUP-NAME/`

That way I can assure that my messages are distributed between instances according to my broker settings - unfortunately even though messages are delivered to my Paho client (which I can see by running a generic callback for all messages), a listener registered while subscribing is not invoked.

This is due to the messages having `$share/GROUP-NAME/` prefix stripped - I've subscribed to` $share/GROUP-NAME/some-topic`, and incoming messages come from just `some-topic`.

As far as I can understand it is a part of the standard: https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901250

At the moment I've been forced to use a generic callback to manually invoke my listener - however this does feel like a hack.

*Minimal code to reproduce*

Expected standard out: `Message arrived in listener, topic: someTopic/AAA-BBB-CCC, message: `
Actual standard out: `Message arrived in callback, topic: someTopic/AAA-BBB-CCC, message: `

```

public class MyApplication {

public static void main(String[] args) throws MqttException, InterruptedException {
MqttClient client = new MqttClient("tcp://localhost:1883", "myClientId", new MemoryPersistence());
client.setCallback(new MqttCallbackImpl());
client.connect();

MqttSubscription subscription = new MqttSubscription("$share/GROUP-NAME/someTopic/#", 2);
client.subscribe(new MqttSubscription[]{subscription}, new IMqttMessageListener[]{new MqttListener()})
.waitForCompletion();

while (true) {
client.publish("someTopic/AAA-BBB-CCC", new MqttMessage());
Thread.sleep(1000);
}
}

public static class MqttListener implements IMqttMessageListener {
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
System.out.println("Message arrived in listener, topic: " + topic + " message: " + message);
}
}

public static class MqttCallbackImpl implements MqttCallback {
@Override
public void messageArrived(String topic, MqttMessage message) {
System.out.println("Message arrived in callback, topic: " + topic + " message: " + message);
}
@Override public void deliveryComplete(IMqttToken token) {}
@Override public void connectComplete(boolean reconnect, String serverURI) {}
@Override public void authPacketArrived(int reasonCode, MqttProperties properties) {}
@Override public void disconnected(MqttDisconnectResponse disconnectResponse) {}
@Override public void mqttErrorOccurred(MqttException exception) {}
}

}

```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.