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 978947b98ff0 camel-smpp: do not process a message the listener is not
initialized for (#29677)
978947b98ff0 is described below
commit 978947b98ff0a6b8857680959647bce754a1d8cc
Author: dhairya05 <[email protected]>
AuthorDate: Fri Nov 13 03:14:42 2026 +0530
camel-smpp: do not process a message the listener is not initialized for
(#29677)
MessageReceiverListenerImpl dropped the result of the bounded wait for its
initialization (and swallowed the interrupt), so a message that arrived
before
the StartupListener ran went on with a null consumer/processor and the SMSC
got
a NullPointerException.
The wait is now awaitInitialization(), which returns the result of the await
and restores the interrupt flag. When it fails, onAcceptDeliverSm and
onAcceptDataSm throw ProcessRequestException (error code 255, as the
existing
fallback), and onAcceptAlertNotification logs a warning and returns.
---
.../smpp/MessageReceiverListenerImpl.java | 41 ++++++++++++++--------
.../smpp/MessageReceiverListenerImplTest.java | 30 ++++++++++++++++
2 files changed, 56 insertions(+), 15 deletions(-)
diff --git
a/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/MessageReceiverListenerImpl.java
b/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/MessageReceiverListenerImpl.java
index 82cbae15aa34..9e39ff3d221d 100644
---
a/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/MessageReceiverListenerImpl.java
+++
b/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/MessageReceiverListenerImpl.java
@@ -42,6 +42,7 @@ import org.slf4j.LoggerFactory;
public class MessageReceiverListenerImpl implements MessageReceiverListener {
private static final Logger LOG =
LoggerFactory.getLogger(MessageReceiverListenerImpl.class);
+ private static final long INITIALIZATION_TIMEOUT_SECONDS = 10;
private MessageIDGenerator messageIDGenerator = new
RandomMessageIDGenerator();
private Consumer consumer;
@@ -76,11 +77,10 @@ public class MessageReceiverListenerImpl implements
MessageReceiverListener {
@Override
public void onAcceptAlertNotification(AlertNotification alertNotification)
{
- try {
- // wait for initialization in case consumer is active before
startup listener above has been executed
- initialized.await(10, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- // ignore
+ if (!awaitInitialization()) {
+ LOG.warn("Skipping alertNotification {}: the consumer was not
initialized within {} seconds",
+ alertNotification, INITIALIZATION_TIMEOUT_SECONDS);
+ return;
}
LOG.debug("Received an alertNotification {}", alertNotification);
@@ -104,11 +104,9 @@ public class MessageReceiverListenerImpl implements
MessageReceiverListener {
@Override
public void onAcceptDeliverSm(DeliverSm deliverSm) throws
ProcessRequestException {
- try {
- // wait for initialization in case consumer is active before
startup listener above has been executed
- initialized.await(10, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- // ignore
+ if (!awaitInitialization()) {
+ throw new ProcessRequestException(
+ "The consumer was not initialized within " +
INITIALIZATION_TIMEOUT_SECONDS + " seconds", 255);
}
LOG.debug("Received a deliverSm {}", deliverSm);
@@ -141,11 +139,9 @@ public class MessageReceiverListenerImpl implements
MessageReceiverListener {
@Override
public DataSmResult onAcceptDataSm(DataSm dataSm, Session session) throws
ProcessRequestException {
- try {
- // wait for initialization in case consumer is active before
startup listener above has been executed
- initialized.await(10, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- // ignore
+ if (!awaitInitialization()) {
+ throw new ProcessRequestException(
+ "The consumer was not initialized within " +
INITIALIZATION_TIMEOUT_SECONDS + " seconds", 255);
}
LOG.debug("Received a dataSm {}", dataSm);
@@ -172,6 +168,21 @@ public class MessageReceiverListenerImpl implements
MessageReceiverListener {
}
}
+ /**
+ * Waits for the startup listener to publish the consumer, the processor
and the exception handler, which a message
+ * arriving before the route is fully started would otherwise read as null.
+ *
+ * @return {@code true} once they are set, {@code false} if the wait timed
out or was interrupted
+ */
+ private boolean awaitInitialization() {
+ try {
+ return initialized.await(INITIALIZATION_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
public void setMessageIDGenerator(MessageIDGenerator messageIDGenerator) {
this.messageIDGenerator = messageIDGenerator;
}
diff --git
a/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/MessageReceiverListenerImplTest.java
b/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/MessageReceiverListenerImplTest.java
index 0de1acecc7f7..558c572afaae 100644
---
a/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/MessageReceiverListenerImplTest.java
+++
b/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/MessageReceiverListenerImplTest.java
@@ -37,6 +37,7 @@ import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
@@ -166,4 +167,33 @@ class MessageReceiverListenerImplTest {
trxContext.stop();
}
}
+
+ @Test
+ void aMessageArrivingBeforeInitializationIsRejected() throws Exception {
+ DefaultCamelContext trxContext = new DefaultCamelContext();
+ trxContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:messageReceiver").routeId("messageReceiver").process(e -> {
+ });
+ }
+ });
+ when(endpoint.getCamelContext()).thenReturn(trxContext);
+
+ // the context is never started, so the StartupListener never runs and
consumer/processor stay null
+ MessageReceiverListenerImpl listener = new
MessageReceiverListenerImpl(endpoint, "messageReceiver");
+
+ try {
+ // makes the bounded wait return at once instead of taking its
full timeout
+ Thread.currentThread().interrupt();
+
+ ProcessRequestException e
+ = assertThrows(ProcessRequestException.class, () ->
listener.onAcceptDeliverSm(deliverSm()));
+ assertTrue(e.getMessage().contains("not initialized"),
e.getMessage());
+ assertTrue(Thread.currentThread().isInterrupted(), "the interrupt
flag has to be restored");
+ } finally {
+ // clear the flag so it cannot leak into another test on this
thread
+ Thread.interrupted();
+ }
+ }
}
