davsclaus commented on code in PR #28324:
URL: https://github.com/apache/camel/pull/28324#discussion_r4357369455
##########
components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java:
##########
@@ -221,6 +194,40 @@ protected void doStop() throws Exception {
}
}
+ /**
+ * Subscribes again after the client reconnected.
+ */
+ void resubscribe() {
+ MqttClient current = client;
+ if (current == null || !isRunAllowed()) {
+ // stopping: do not subscribe the topic again, the stop
unsubscribes it
+ return;
+ }
+ try {
+ current.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
+ } catch (MqttException e) {
+ if (stopClient) {
+ LOG.warn("MQTT resubscribe failed on reconnect, restarting
route for recovery: {}",
+ e.getMessage(), e);
+ restartRouteAsync();
+ } else {
+ LOG.error(
+ "MQTT resubscribe failed on reconnect with externally
provided client,"
+ + " route will not be auto-restarted: {}",
+ e.getMessage(), e);
+ }
+ }
+ }
+
+ void onMessageArrived(String topic, MqttMessage message) {
+ LOG.debug("Message arrived on topic: {} -> {}", topic, message);
Review Comment:
Nit: `PahoClientCallback.messageArrived` already logs this line, so each
message is logged twice at DEBUG. One of the two can go:
```suggestion
```
##########
components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoClientCallback.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.paho;
+
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.MqttTopic;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The callback of a {@link MqttClient} used by consumers. A client has a
single callback, and the same client can be
+ * used by the consumers of several endpoints (the client option), so the
callback is shared by all the consumers of the
+ * client: a message goes to the consumers whose topic matches it, and every
consumer subscribes again after a
+ * reconnect. Only a client that the consumer created for itself gives every
message to that consumer. A message that no
+ * consumer of a shared client matches is dropped, and acknowledged when its
consumers use manualAcksEnabled.
+ */
+final class PahoClientCallback implements MqttCallbackExtended {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PahoClientCallback.class);
+
+ // the callbacks of the clients that have consumers, by client instance
+ private static final Map<MqttClient, PahoClientCallback> CALLBACKS = new
IdentityHashMap<>();
+
+ // the number of topics whose unmatched messages are logged at WARN (the
next ones at DEBUG)
+ private static final int MAX_WARNED_TOPICS = 100;
+
+ private final List<PahoConsumer> consumers = new CopyOnWriteArrayList<>();
+ private final MqttClient client;
+ // whether the client is the own client of its consumer, so that no other
consumer ever used it
+ private final boolean ownedClient;
+ // whether a consumer of the client acknowledges its messages itself
(manualAcksEnabled); kept after it stops, as
+ // the messages of its subscription can still arrive
+ private volatile boolean manualAcks;
+ private final Set<String> warnedTopics = ConcurrentHashMap.newKeySet();
+
+ private PahoClientCallback(MqttClient client, boolean ownedClient) {
+ this.client = client;
+ this.ownedClient = ownedClient;
+ }
+
+ /**
+ * Adds the consumer to the callback of the client, and sets that callback
on the client.
+ *
+ * @param ownedClient whether the consumer created the client for itself
(no other consumer uses it)
+ */
+ static void register(MqttClient client, PahoConsumer consumer, boolean
ownedClient) {
+ synchronized (CALLBACKS) {
+ PahoClientCallback callback
+ = CALLBACKS.computeIfAbsent(client, c -> new
PahoClientCallback(c, ownedClient));
+ if
(consumer.getEndpoint().getConfiguration().isManualAcksEnabled()) {
+ callback.manualAcks = true;
+ }
+ if (!callback.consumers.contains(consumer)) {
+ callback.consumers.add(consumer);
+ }
+ client.setCallback(callback);
+ }
+ }
+
+ /**
+ * Removes the consumer from the callback of its client, if any.
+ *
+ * @return whether another consumer of the client still uses the topic
filter of the consumer, so that the topic
+ * must stay subscribed
+ */
+ static boolean unregister(PahoConsumer consumer) {
+ String topic = consumer.getEndpoint().getTopic();
+ boolean topicStillUsed = false;
+ synchronized (CALLBACKS) {
+ Iterator<PahoClientCallback> it = CALLBACKS.values().iterator();
+ while (it.hasNext()) {
+ PahoClientCallback callback = it.next();
+ if (callback.consumers.remove(consumer)) {
+ for (PahoConsumer other : callback.consumers) {
+ if (topic.equals(other.getEndpoint().getTopic())) {
+ topicStillUsed = true;
+ }
+ }
+ }
+ if (callback.consumers.isEmpty()) {
+ it.remove();
+ }
+ }
+ }
+ return topicStillUsed;
+ }
+
+ @Override
+ public void connectComplete(boolean reconnect, String serverURI) {
+ if (reconnect) {
+ for (PahoConsumer consumer : consumers) {
+ consumer.resubscribe();
+ }
+ }
+ }
+
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws
Exception {
+ LOG.debug("Message arrived on topic: {} -> {}", topic, message);
+ // a snapshot: a consumer can unregister on another thread
+ PahoConsumer[] current = consumers.toArray(new PahoConsumer[0]);
+ if (ownedClient && current.length == 1) {
+ // the own client of a consumer: it is the only one that
subscribed, so it gets every message, as before
+ current[0].onMessageArrived(topic, message);
+ return;
+ }
+ // a shared client: a stopped consumer can leave its subscription on
the client (cleanStart / cleanSession
+ // false), so a message goes only to the consumers whose topic filter
matches it, even when one consumer is left
+ boolean delivered = false;
+ for (PahoConsumer consumer : current) {
+ if (matches(consumer.getEndpoint().getTopic(), topic)) {
+ consumer.onMessageArrived(topic, message);
+ delivered = true;
+ }
+ }
+ if (!delivered) {
+ // with manual acks the client sends no PUBACK (QoS 1) or PUBCOMP
(QoS 2) until the message is acknowledged,
+ // and the broker sends a limited number of unacknowledged
messages, so a dropped message that stays
+ // unacknowledged would stop every consumer of the client
+ // (the own client of a consumer is not acknowledged: it is being
stopped, and a session that the broker
+ // keeps sends the message again on the next connect)
+ boolean acknowledge = !ownedClient && manualAcks &&
message.getQos() > 0;
+ String dropped = acknowledge ? "dropped and acknowledged" :
"dropped";
+ if (warnedTopics.size() < MAX_WARNED_TOPICS &&
warnedTopics.add(topic)) {
+ LOG.warn("Message arrived on topic: {} that no running
consumer of the client subscribed to; it is {}"
+ + " (further messages of this topic are logged at
DEBUG)",
+ topic, dropped);
+ } else {
+ LOG.debug("Message arrived on topic: {} that no running
consumer of the client subscribed to; it is {}",
+ topic, dropped);
+ }
+ if (acknowledge) {
+ acknowledge(topic, message);
+ }
+ }
+ }
+
+ private void acknowledge(String topic, MqttMessage message) {
+ try {
+ client.messageArrivedComplete(message.getId(), message.getQos());
+ } catch (MqttException e) {
+ LOG.warn("Failed to acknowledge the dropped MQTT message with ID:
{} of topic: {}: {}",
+ message.getId(), topic, e.toString(), e);
+ }
+ }
+
+ static boolean matches(String topicFilter, String topic) {
+ String filter = topicFilter;
+ // a shared subscription ($share/<group>/<filter>) delivers the
messages of <filter>
+ if (filter.startsWith("$share/")) {
+ int index = filter.indexOf('/', "$share/".length());
+ if (index > 0) {
+ filter = filter.substring(index + 1);
+ }
+ }
+ try {
+ return MqttTopic.isMatched(filter, topic);
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
Review Comment:
Same as in camel-paho-mqtt5: `MqttTopic.isMatched` (v3, 1.2.5) gives the
same wrong results for empty levels and `$` topics. Suggested replacement (the
`MqttTopic` import is then unused):
```suggestion
// MQTT 4.7.2: a filter that starts with a wildcard does not match a
topic that starts with $
if (topic.startsWith("$") && (filter.startsWith("+") ||
filter.startsWith("#"))) {
return false;
}
// split with -1 keeps the empty levels, which + matches (MQTT
4.7.1.3)
String[] filterLevels = filter.split("/", -1);
String[] topicLevels = topic.split("/", -1);
for (int i = 0; i < filterLevels.length; i++) {
String level = filterLevels[i];
if ("#".equals(level)) {
// # matches the parent level and any number of child levels
return i == filterLevels.length - 1;
}
if (i >= topicLevels.length || !("+".equals(level) ||
level.equals(topicLevels[i]))) {
return false;
}
}
return filterLevels.length == topicLevels.length;
```
##########
components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ClientCallback.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.paho.mqtt5;
+
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.eclipse.paho.mqttv5.client.IMqttToken;
+import org.eclipse.paho.mqttv5.client.MqttCallback;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
+import org.eclipse.paho.mqttv5.common.MqttException;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
+import org.eclipse.paho.mqttv5.common.util.MqttTopicValidator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The callback of a {@link MqttClient} used by consumers. A client has a
single callback, and the same client can be
+ * used by the consumers of several endpoints (the client option), so the
callback is shared by all the consumers of the
+ * client: a message goes to the consumers whose topic matches it, and every
consumer subscribes again after a
+ * reconnect. Only a client that the consumer created for itself gives every
message to that consumer. A message that no
+ * consumer of a shared client matches is dropped, and acknowledged when its
consumers use manualAcksEnabled.
+ */
+final class PahoMqtt5ClientCallback implements MqttCallback {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PahoMqtt5ClientCallback.class);
+
+ // the callbacks of the clients that have consumers, by client instance
+ private static final Map<MqttClient, PahoMqtt5ClientCallback> CALLBACKS =
new IdentityHashMap<>();
+
+ // the number of topics whose unmatched messages are logged at WARN (the
next ones at DEBUG)
+ private static final int MAX_WARNED_TOPICS = 100;
+
+ private final List<PahoMqtt5Consumer> consumers = new
CopyOnWriteArrayList<>();
+ private final MqttClient client;
+ // whether the client is the own client of its consumer, so that no other
consumer ever used it
+ private final boolean ownedClient;
+ // whether a consumer of the client acknowledges its messages itself
(manualAcksEnabled); kept after it stops, as
+ // the messages of its subscription can still arrive
+ private volatile boolean manualAcks;
+ private final Set<String> warnedTopics = ConcurrentHashMap.newKeySet();
+
+ private PahoMqtt5ClientCallback(MqttClient client, boolean ownedClient) {
+ this.client = client;
+ this.ownedClient = ownedClient;
+ }
+
+ /**
+ * Adds the consumer to the callback of the client, and sets that callback
on the client.
+ *
+ * @param ownedClient whether the consumer created the client for itself
(no other consumer uses it)
+ */
+ static void register(MqttClient client, PahoMqtt5Consumer consumer,
boolean ownedClient) {
+ synchronized (CALLBACKS) {
+ PahoMqtt5ClientCallback callback
+ = CALLBACKS.computeIfAbsent(client, c -> new
PahoMqtt5ClientCallback(c, ownedClient));
+ if
(consumer.getEndpoint().getConfiguration().isManualAcksEnabled()) {
+ callback.manualAcks = true;
+ }
+ if (!callback.consumers.contains(consumer)) {
+ callback.consumers.add(consumer);
+ }
+ client.setCallback(callback);
+ }
+ }
+
+ /**
+ * Removes the consumer from the callback of its client, if any.
+ *
+ * @return whether another consumer of the client still uses the topic
filter of the consumer, so that the topic
+ * must stay subscribed
+ */
+ static boolean unregister(PahoMqtt5Consumer consumer) {
+ String topic = consumer.getEndpoint().getTopic();
+ boolean topicStillUsed = false;
+ synchronized (CALLBACKS) {
+ Iterator<PahoMqtt5ClientCallback> it =
CALLBACKS.values().iterator();
+ while (it.hasNext()) {
+ PahoMqtt5ClientCallback callback = it.next();
+ if (callback.consumers.remove(consumer)) {
+ for (PahoMqtt5Consumer other : callback.consumers) {
+ if (topic.equals(other.getEndpoint().getTopic())) {
+ topicStillUsed = true;
+ }
+ }
+ }
+ if (callback.consumers.isEmpty()) {
+ it.remove();
+ }
+ }
+ }
+ return topicStillUsed;
+ }
+
+ @Override
+ public void connectComplete(boolean reconnect, String serverURI) {
+ for (PahoMqtt5Consumer consumer : consumers) {
+ consumer.connectComplete(reconnect);
+ }
+ }
+
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws
Exception {
+ LOG.debug("Message arrived on topic: {} -> {}", topic, message);
+ // a snapshot: a consumer can unregister on another thread
+ PahoMqtt5Consumer[] current = consumers.toArray(new
PahoMqtt5Consumer[0]);
+ if (ownedClient && current.length == 1) {
+ // the own client of a consumer: it is the only one that
subscribed, so it gets every message, as before
+ current[0].onMessageArrived(topic, message);
+ return;
+ }
+ // a shared client: a stopped consumer can leave its subscription on
the client (cleanStart / cleanSession
+ // false), so a message goes only to the consumers whose topic filter
matches it, even when one consumer is left
+ boolean delivered = false;
+ for (PahoMqtt5Consumer consumer : current) {
+ if (matches(consumer.getEndpoint().getTopic(), topic)) {
+ consumer.onMessageArrived(topic, message);
+ delivered = true;
+ }
+ }
+ if (!delivered) {
+ // with manual acks the client sends no PUBACK for a QoS 1 message
until it is acknowledged, and the broker
+ // sends no more than receiveMaximum unacknowledged messages, so a
dropped message that stays unacknowledged
+ // would stop every consumer of the client; the client completes
the QoS 2 flow on its own
+ // (the own client of a consumer is not acknowledged: it is being
stopped, and a session that the broker
+ // keeps sends the message again on the next connect)
+ boolean acknowledge = !ownedClient && manualAcks &&
message.getQos() == 1;
+ String dropped = acknowledge ? "dropped and acknowledged" :
"dropped";
+ if (warnedTopics.size() < MAX_WARNED_TOPICS &&
warnedTopics.add(topic)) {
+ LOG.warn("Message arrived on topic: {} that no running
consumer of the client subscribed to; it is {}"
+ + " (further messages of this topic are logged at
DEBUG)",
+ topic, dropped);
+ } else {
+ LOG.debug("Message arrived on topic: {} that no running
consumer of the client subscribed to; it is {}",
+ topic, dropped);
+ }
+ if (acknowledge) {
+ acknowledge(topic, message);
+ }
+ }
+ }
+
+ private void acknowledge(String topic, MqttMessage message) {
+ try {
+ client.messageArrivedComplete(message.getId(), message.getQos());
+ } catch (MqttException e) {
+ LOG.warn("Failed to acknowledge the dropped MQTT message with ID:
{} of topic: {}: {}",
+ message.getId(), topic, e.toString(), e);
+ }
+ }
+
+ static boolean matches(String topicFilter, String topic) {
+ String filter = topicFilter;
+ // a shared subscription ($share/<group>/<filter>) delivers the
messages of <filter>
+ if (filter.startsWith("$share/")) {
+ int index = filter.indexOf('/', "$share/".length());
+ if (index > 0) {
+ filter = filter.substring(index + 1);
+ }
+ }
+ try {
+ return MqttTopicValidator.isMatched(filter, topic);
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
Review Comment:
`MqttTopicValidator.isMatched` does not let `+` match an empty level (`+/+`
vs `/finance`, `a/+/b` vs `a//b` return false) and lets a leading wildcard
match a `$` topic (`#` vs `$SYS/x` returns true), see finding 1. A matcher on
the split levels follows MQTT 4.7 (the `MqttTopicValidator` import is then
unused):
```suggestion
// MQTT 4.7.2: a filter that starts with a wildcard does not match a
topic that starts with $
if (topic.startsWith("$") && (filter.startsWith("+") ||
filter.startsWith("#"))) {
return false;
}
// split with -1 keeps the empty levels, which + matches (MQTT
4.7.1.3)
String[] filterLevels = filter.split("/", -1);
String[] topicLevels = topic.split("/", -1);
for (int i = 0; i < filterLevels.length; i++) {
String level = filterLevels[i];
if ("#".equals(level)) {
// # matches the parent level and any number of child levels
return i == filterLevels.length - 1;
}
if (i >= topicLevels.length || !("+".equals(level) ||
level.equals(topicLevels[i]))) {
return false;
}
}
return filterLevels.length == topicLevels.length;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
