This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 6044f38166f9 CAMEL-25944: camel-paho-mqtt5, camel-paho - connect the
consumer's client after its callback is set (#28462)
6044f38166f9 is described below
commit 6044f38166f950799468d2d1f862c5715b43b0cf
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 26 13:31:39 2026 +0530
CAMEL-25944: camel-paho-mqtt5, camel-paho - connect the consumer's client
after its callback is set (#28462)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/paho/mqtt5/PahoMqtt5Consumer.java | 10 +-
.../PahoMqtt5ConsumerSessionMessagesTest.java | 106 +++++++++++++++++++++
.../apache/camel/component/paho/PahoConsumer.java | 10 +-
.../paho/PahoConsumerSessionMessagesTest.java | 105 ++++++++++++++++++++
4 files changed, 227 insertions(+), 4 deletions(-)
diff --git
a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java
b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java
index 94d55b6b5f48..5c0ba1eec792 100644
---
a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java
+++
b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java
@@ -88,16 +88,22 @@ public class PahoMqtt5Consumer extends DefaultConsumer {
clientId = PahoMqtt5Endpoint.generateClientId();
}
client = createClient();
- LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
if (getEndpoint().getConfiguration().isManualAcksEnabled()) {
client.setManualAcks(true);
}
- client.connect(connectionOptions);
}
// the client can be shared by the consumers of several endpoints,
and it has a single callback
PahoMqtt5ClientCallback.register(client, this, stopClient);
+ if (stopClient) {
+ // connect once the callback is set: on a session that the
broker kept, it sends the messages of the
+ // session right after CONNACK, and Paho drops a message that
arrives while the client has no callback
+ // (and completes the QoS 2 flow of it all the same)
+ LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
+ client.connect(connectionOptions);
+ }
+
LOG.debug("Subscribing client: {} to topic: {}", clientId,
getEndpoint().getTopic());
client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
} catch (Exception startException) {
diff --git
a/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerSessionMessagesTest.java
b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerSessionMessagesTest.java
new file mode 100644
index 000000000000..6fc63a1c304c
--- /dev/null
+++
b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ConsumerSessionMessagesTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.eclipse.paho.mqttv5.client.MqttCallback;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+
+/**
+ * A consumer that connects to a session the broker kept (cleanStart=false and
a session expiry interval) gets the
+ * messages that the broker queued for the session right after CONNACK, before
it subscribes.
+ */
+class PahoMqtt5ConsumerSessionMessagesTest extends CamelTestSupport {
+
+ private final MqttClient client = mock(MqttClient.class);
+ private final AtomicReference<MqttCallback> callback = new
AtomicReference<>();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ PahoMqtt5Configuration configuration = new PahoMqtt5Configuration();
+ configuration.setBrokerUrl("tcp://localhost:1883");
+ configuration.setCleanStart(false);
+ configuration.setSessionExpiryInterval(3600);
+ context.addComponent("paho-mqtt5-session", new
PahoMqtt5Component(context) {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ return new PahoMqtt5Endpoint(uri, remaining, this,
configuration.copy()) {
+ @Override
+ public Consumer createConsumer(Processor processor) throws
Exception {
+ PahoMqtt5Consumer consumer = new
PahoMqtt5Consumer(this, processor) {
+ @Override
+ MqttClient createClient() {
+ return client;
+ }
+ };
+ configureConsumer(consumer);
+ return consumer;
+ }
+ };
+ }
+ });
+
+ doAnswer(invocation -> {
+ callback.set(invocation.getArgument(0));
+ return null;
+ }).when(client).setCallback(any(MqttCallback.class));
+ // the broker sends a message of the session right after CONNACK. Paho
1.2.5 (CommsCallback.messageArrived)
+ // gives a message only to the callback set at that time: without one
it drops the message, and for QoS 2
+ // ClientState sends PUBREC and then PUBCOMP all the same, so the
broker never sends it again
+ doAnswer(invocation -> {
+ MqttCallback current = callback.get();
+ if (current != null) {
+ MqttMessage message = new MqttMessage("queued while
offline".getBytes(StandardCharsets.UTF_8));
+ message.setQos(2);
+ current.messageArrived("test", message);
+ }
+ return null;
+ }).when(client).connect(any(MqttConnectionOptions.class));
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("paho-mqtt5-session:test").to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void messageSentRightAfterConnackIsRouted() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("queued while offline");
+ mock.setResultWaitTime(5000);
+
+ mock.assertIsSatisfied();
+ }
+}
diff --git
a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
index a63803f6f646..13eea01ca9bc 100644
---
a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
+++
b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConsumer.java
@@ -69,16 +69,22 @@ public class PahoConsumer extends DefaultConsumer {
clientId = "camel-" + MqttClient.generateClientId();
}
client = createClient();
- LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
if (getEndpoint().getConfiguration().isManualAcksEnabled()) {
client.setManualAcks(true);
}
- client.connect(connectOptions);
}
// the client can be shared by the consumers of several endpoints,
and it has a single callback
PahoClientCallback.register(client, this, stopClient);
+ if (stopClient) {
+ // connect once the callback is set: on a session that the
broker kept (cleanSession=false), it sends
+ // the messages of the session right after CONNACK, and Paho
drops a message that arrives while the
+ // client has no callback (without acknowledging it)
+ LOG.debug("Connecting client: {} to broker: {}", clientId,
getEndpoint().getConfiguration().getBrokerUrl());
+ client.connect(connectOptions);
+ }
+
LOG.debug("Subscribing client: {} to topic: {}", clientId,
getEndpoint().getTopic());
client.subscribe(getEndpoint().getTopic(),
getEndpoint().getConfiguration().getQos());
} catch (Exception startException) {
diff --git
a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerSessionMessagesTest.java
b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerSessionMessagesTest.java
new file mode 100644
index 000000000000..d55dec553d26
--- /dev/null
+++
b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoConsumerSessionMessagesTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+
+/**
+ * A consumer that connects to a session the broker kept (cleanSession=false)
gets the messages that the broker queued
+ * for the session right after CONNACK, before it subscribes.
+ */
+class PahoConsumerSessionMessagesTest extends CamelTestSupport {
+
+ private final MqttClient client = mock(MqttClient.class);
+ private final AtomicReference<MqttCallback> callback = new
AtomicReference<>();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ PahoConfiguration configuration = new PahoConfiguration();
+ configuration.setBrokerUrl("tcp://localhost:1883");
+ configuration.setCleanSession(false);
+ context.addComponent("paho-session", new PahoComponent(context) {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ return new PahoEndpoint(uri, remaining, this,
configuration.copy()) {
+ @Override
+ public Consumer createConsumer(Processor processor) throws
Exception {
+ PahoConsumer consumer = new PahoConsumer(this,
processor) {
+ @Override
+ MqttClient createClient() {
+ return client;
+ }
+ };
+ configureConsumer(consumer);
+ return consumer;
+ }
+ };
+ }
+ });
+
+ doAnswer(invocation -> {
+ callback.set(invocation.getArgument(0));
+ return null;
+ }).when(client).setCallback(any(MqttCallback.class));
+ // the broker sends a message of the session right after CONNACK. Paho
1.2.5 (CommsCallback.messageArrived)
+ // gives a message only to the callback set at that time: without one
it drops the message and does not
+ // acknowledge it, so the broker sends it again only after the next
connect
+ doAnswer(invocation -> {
+ MqttCallback current = callback.get();
+ if (current != null) {
+ MqttMessage message = new MqttMessage("queued while
offline".getBytes(StandardCharsets.UTF_8));
+ message.setQos(1);
+ current.messageArrived("test", message);
+ }
+ return null;
+ }).when(client).connect(any(MqttConnectOptions.class));
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("paho-session:test").to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void messageSentRightAfterConnackIsRouted() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("queued while offline");
+ mock.setResultWaitTime(5000);
+
+ mock.assertIsSatisfied();
+ }
+}
