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 dfa135ebf0f1 CAMEL-25308: camel-smpp - build received exchanges
through the consumer, not the endpoint (#27329)
dfa135ebf0f1 is described below
commit dfa135ebf0f178dbb8851014352b61fd90f642ae
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Oct 5 19:04:36 2026 +0200
CAMEL-25308: camel-smpp - build received exchanges through the consumer,
not the endpoint (#27329)
Co-Authored-By: Claude Opus 5 <[email protected]>
---
.../smpp/MessageReceiverListenerImpl.java | 133 ++++++++++++----
.../apache/camel/component/smpp/SmppEndpoint.java | 43 ++++--
.../smpp/MessageReceiverListenerImplTest.java | 169 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 16 ++
4 files changed, 316 insertions(+), 45 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 484a503f1c39..82cbae15aa34 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
@@ -86,16 +86,20 @@ public class MessageReceiverListenerImpl implements
MessageReceiverListener {
Exchange exchange =
createOnAcceptAlertNotificationExchange(alertNotification);
try {
- processor.process(exchange);
- } catch (Exception e) {
- exchange.setException(e);
- }
+ try {
+ processor.process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
- if (exchange.getException() != null) {
- exceptionHandler.handleException("Cannot process exchange. This
exception will be ignored.", exchange,
- exchange.getException());
+ if (exchange.getException() != null) {
+ exceptionHandler.handleException("Cannot process exchange.
This exception will be ignored.", exchange,
+ exchange.getException());
+ }
+ } finally {
+ // in a finally so that a custom exception handler that throws
cannot cost the release
+ consumer.releaseExchange(exchange, false);
}
- consumer.releaseExchange(exchange, false);
}
@Override
@@ -110,24 +114,28 @@ public class MessageReceiverListenerImpl implements
MessageReceiverListener {
Exchange exchange;
try {
- exchange = endpoint.createOnAcceptDeliverSmExchange(deliverSm);
+ exchange = createOnAcceptDeliverSmExchange(deliverSm);
} catch (Exception e) {
exceptionHandler.handleException("Cannot create exchange. This
exception will be ignored.", e);
return;
}
try {
- processor.process(exchange);
- } catch (Exception e) {
- exchange.setException(e);
- }
+ try {
+ processor.process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
- if (exchange.getException() != null) {
- ProcessRequestException pre =
exchange.getException(ProcessRequestException.class);
- if (pre == null) {
- pre = new
ProcessRequestException(exchange.getException().getMessage(), 255,
exchange.getException());
+ if (exchange.getException() != null) {
+ ProcessRequestException pre =
exchange.getException(ProcessRequestException.class);
+ if (pre == null) {
+ pre = new
ProcessRequestException(exchange.getException().getMessage(), 255,
exchange.getException());
+ }
+ throw pre;
}
- throw pre;
+ } finally {
+ consumer.releaseExchange(exchange, false);
}
}
@@ -142,28 +150,82 @@ public class MessageReceiverListenerImpl implements
MessageReceiverListener {
LOG.debug("Received a dataSm {}", dataSm);
MessageId newMessageId = messageIDGenerator.newMessageId();
- Exchange exchange = endpoint.createOnAcceptDataSm(dataSm,
newMessageId.getValue());
+ Exchange exchange = createOnAcceptDataSmExchange(dataSm,
newMessageId.getValue());
try {
- processor.process(exchange);
- } catch (Exception e) {
- exchange.setException(e);
- }
+ try {
+ processor.process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
- if (exchange.getException() != null) {
- ProcessRequestException pre =
exchange.getException(ProcessRequestException.class);
- if (pre == null) {
- pre = new
ProcessRequestException(exchange.getException().getMessage(), 255,
exchange.getException());
+ if (exchange.getException() != null) {
+ ProcessRequestException pre =
exchange.getException(ProcessRequestException.class);
+ if (pre == null) {
+ pre = new
ProcessRequestException(exchange.getException().getMessage(), 255,
exchange.getException());
+ }
+ throw pre;
}
- throw pre;
- }
- return new DataSmResult(newMessageId, dataSm.getOptionalParameters());
+ return new DataSmResult(newMessageId,
dataSm.getOptionalParameters());
+ } finally {
+ consumer.releaseExchange(exchange, false);
+ }
}
public void setMessageIDGenerator(MessageIDGenerator messageIDGenerator) {
this.messageIDGenerator = messageIDGenerator;
}
+ /**
+ * Creates the exchange for a received {@code deliver_sm}.
+ * <p/>
+ * Built through {@link Consumer#createExchange(boolean)} rather than on
the endpoint, as
+ * {@link org.apache.camel.Endpoint#createExchange()} instructs: an
exchange a consumer received a message for has
+ * to come from the configured {@code ExchangeFactory}, or pooling never
applies to it and its {@code fromRouteId}
+ * is left unset.
+ *
+ * @param deliverSm the received message from the SMSC
+ * @return a new exchange
+ */
+ private Exchange createOnAcceptDeliverSmExchange(DeliverSm deliverSm)
throws Exception {
+ Exchange exchange = consumer.createExchange(false);
+ try {
+ // the pattern has to be set explicitly: in transceiver mode the
consumer belongs to the receiver route,
+ // not to this SMPP endpoint, so the factory would otherwise stamp
that endpoint's pattern on the exchange
+ exchange.setPattern(endpoint.getExchangePattern());
+ exchange.setProperty(Exchange.BINDING, endpoint.getBinding());
+
exchange.setIn(endpoint.getBinding().createSmppMessage(endpoint.getCamelContext(),
deliverSm));
+ return exchange;
+ } catch (Exception e) {
+ // the exchange is already out of the ExchangeFactory, and the
caller's variable is still unassigned,
+ // so its finally cannot release it - a failed decode would
otherwise cost a pooled exchange
+ consumer.releaseExchange(exchange, false);
+ throw e;
+ }
+ }
+
+ /**
+ * Creates the exchange for a received {@code data_sm}, through the
consumer for the same reason as
+ * {@link #createOnAcceptDeliverSmExchange(DeliverSm)}.
+ *
+ * @param dataSm the received message from the SMSC
+ * @param smppMessageId the smpp message id which will be used in the
response
+ * @return a new exchange
+ */
+ private Exchange createOnAcceptDataSmExchange(DataSm dataSm, String
smppMessageId) {
+ Exchange exchange = consumer.createExchange(false);
+ try {
+ // see createOnAcceptDeliverSmExchange: the pattern cannot be left
to the factory in transceiver mode
+ exchange.setPattern(endpoint.getExchangePattern());
+ exchange.setProperty(Exchange.BINDING, endpoint.getBinding());
+
exchange.setIn(endpoint.getBinding().createSmppMessage(endpoint.getCamelContext(),
dataSm, smppMessageId));
+ return exchange;
+ } catch (Exception e) {
+ consumer.releaseExchange(exchange, false);
+ throw e;
+ }
+ }
+
/**
* Create a new exchange for communicating with this endpoint from a SMSC
with the specified {@link ExchangePattern}
* such as whether its going to be an {@link ExchangePattern#InOnly} or
{@link ExchangePattern#InOut} exchange
@@ -173,9 +235,14 @@ public class MessageReceiverListenerImpl implements
MessageReceiverListener {
*/
public Exchange createOnAcceptAlertNotificationExchange(AlertNotification
alertNotification) {
Exchange exchange = consumer.createExchange(false);
- exchange.setProperty(Exchange.BINDING, endpoint.getBinding());
-
exchange.setIn(endpoint.getBinding().createSmppMessage(endpoint.getCamelContext(),
alertNotification));
- return exchange;
+ try {
+ exchange.setProperty(Exchange.BINDING, endpoint.getBinding());
+
exchange.setIn(endpoint.getBinding().createSmppMessage(endpoint.getCamelContext(),
alertNotification));
+ return exchange;
+ } catch (Exception e) {
+ consumer.releaseExchange(exchange, false);
+ throw e;
+ }
}
}
diff --git
a/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/SmppEndpoint.java
b/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/SmppEndpoint.java
index 62dc5fcfeb6f..84771fb69c4a 100644
---
a/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/SmppEndpoint.java
+++
b/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/SmppEndpoint.java
@@ -81,9 +81,13 @@ public class SmppEndpoint extends DefaultEndpoint implements
EndpointServiceLoca
/**
* Create a new exchange for communicating with this endpoint from a SMSC
*
- * @param deliverSm the received message from the SMSC
- * @return a new exchange
+ * @param deliverSm the received message from the SMSC
+ * @return a new exchange
+ * @deprecated an exchange a consumer received a message for has
to come from
+ * {@link
org.apache.camel.Consumer#createExchange(boolean)}, so that the configured
+ * {@code ExchangeFactory} sees it; {@code
MessageReceiverListenerImpl} does that itself now
*/
+ @Deprecated
public Exchange createOnAcceptDeliverSmExchange(DeliverSm deliverSm)
throws Exception {
return createOnAcceptDeliverSmExchange(getExchangePattern(),
deliverSm);
}
@@ -92,10 +96,15 @@ public class SmppEndpoint extends DefaultEndpoint
implements EndpointServiceLoca
* Create a new exchange for communicating with this endpoint from a SMSC
with the specified {@link ExchangePattern}
* such as whether its going to be an {@link ExchangePattern#InOnly} or
{@link ExchangePattern#InOut} exchange
*
- * @param exchangePattern the message exchange pattern for the exchange
- * @param deliverSm the received message from the SMSC
- * @return a new exchange
+ * @param exchangePattern the message exchange pattern for the
exchange
+ * @param deliverSm the received message from the SMSC
+ * @return a new exchange
+ * @deprecated an exchange a consumer received a message
for has to come from
+ * {@link
org.apache.camel.Consumer#createExchange(boolean)}, so that the configured
+ * {@code ExchangeFactory} sees it; {@code
MessageReceiverListenerImpl} does that itself
+ * now
*/
+ @Deprecated
public Exchange createOnAcceptDeliverSmExchange(
ExchangePattern exchangePattern,
DeliverSm deliverSm)
@@ -109,10 +118,15 @@ public class SmppEndpoint extends DefaultEndpoint
implements EndpointServiceLoca
/**
* Create a new exchange for communicating with this endpoint from a SMSC
*
- * @param dataSm the received message from the SMSC
- * @param smppMessageId the smpp message id which will be used in the
response
- * @return a new exchange
+ * @param dataSm the received message from the SMSC
+ * @param smppMessageId the smpp message id which will be used in the
response
+ * @return a new exchange
+ * @deprecated an exchange a consumer received a message for
has to come from
+ * {@link
org.apache.camel.Consumer#createExchange(boolean)}, so that the configured
+ * {@code ExchangeFactory} sees it; {@code
MessageReceiverListenerImpl} does that itself
+ * now
*/
+ @Deprecated
public Exchange createOnAcceptDataSm(DataSm dataSm, String smppMessageId) {
return createOnAcceptDataSm(getExchangePattern(), dataSm,
smppMessageId);
}
@@ -121,11 +135,16 @@ public class SmppEndpoint extends DefaultEndpoint
implements EndpointServiceLoca
* Create a new exchange for communicating with this endpoint from a SMSC
with the specified {@link ExchangePattern}
* such as whether its going to be an {@link ExchangePattern#InOnly} or
{@link ExchangePattern#InOut} exchange
*
- * @param exchangePattern the message exchange pattern for the exchange
- * @param dataSm the received message from the SMSC
- * @param smppMessageId the smpp message id which will be used in the
response
- * @return a new exchange
+ * @param exchangePattern the message exchange pattern for the
exchange
+ * @param dataSm the received message from the SMSC
+ * @param smppMessageId the smpp message id which will be used in
the response
+ * @return a new exchange
+ * @deprecated an exchange a consumer received a message
for has to come from
+ * {@link
org.apache.camel.Consumer#createExchange(boolean)}, so that the configured
+ * {@code ExchangeFactory} sees it; {@code
MessageReceiverListenerImpl} does that itself
+ * now
*/
+ @Deprecated
public Exchange createOnAcceptDataSm(ExchangePattern exchangePattern,
DataSm dataSm, String smppMessageId) {
Exchange exchange = createExchange(exchangePattern);
exchange.setProperty(Exchange.BINDING, getBinding());
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
new file mode 100644
index 000000000000..0de1acecc7f7
--- /dev/null
+++
b/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/MessageReceiverListenerImplTest.java
@@ -0,0 +1,169 @@
+/*
+ * 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.smpp;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.support.DefaultExchange;
+import org.jsmpp.bean.DataSm;
+import org.jsmpp.bean.DeliverSm;
+import org.jsmpp.extra.ProcessRequestException;
+import org.jsmpp.session.Session;
+import org.junit.jupiter.api.BeforeEach;
+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.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * An exchange a consumer received a message for has to be built through
{@link org.apache.camel.Consumer} rather than
+ * on the endpoint, so that the configured {@code ExchangeFactory} sees it,
and it has to be released afterwards.
+ */
+class MessageReceiverListenerImplTest {
+
+ private SmppConsumer consumer;
+ private SmppEndpoint endpoint;
+ private CamelContext camelContext;
+ private Exchange exchange;
+
+ @BeforeEach
+ void setUp() {
+ camelContext = new DefaultCamelContext();
+ exchange = new DefaultExchange(camelContext);
+ consumer = mock(SmppConsumer.class);
+ endpoint = mock(SmppEndpoint.class);
+
+ when(consumer.createExchange(false)).thenReturn(exchange);
+ when(endpoint.getCamelContext()).thenReturn(camelContext);
+ when(endpoint.getBinding()).thenReturn(new SmppBinding());
+ when(endpoint.getExchangePattern()).thenReturn(ExchangePattern.InOnly);
+ }
+
+ private MessageReceiverListenerImpl listener(Processor processor) {
+ return new MessageReceiverListenerImpl(consumer, endpoint, processor,
mock(ExceptionHandler.class));
+ }
+
+ private static DeliverSm deliverSm() {
+ DeliverSm deliverSm = new DeliverSm();
+ deliverSm.setShortMessage("Hello SMPP world!".getBytes());
+ return deliverSm;
+ }
+
+ @Test
+ void deliverSmTakesItsExchangeFromTheConsumerAndReleasesIt() throws
Exception {
+ listener(received -> assertEquals("Hello SMPP world!",
received.getIn().getBody(String.class)))
+ .onAcceptDeliverSm(deliverSm());
+
+ verify(consumer).createExchange(false);
+ verify(consumer).releaseExchange(exchange, false);
+ }
+
+ /**
+ * The failure path throws a {@link ProcessRequestException} so the SMSC
gets a NACK, and must still release.
+ */
+ @Test
+ void deliverSmReleasesItsExchangeWhenTheRouteFailed() {
+ MessageReceiverListenerImpl listener
+ = listener(received -> received.setException(new
IllegalStateException("the route blew up")));
+
+ ProcessRequestException thrown
+ = assertThrows(ProcessRequestException.class, () ->
listener.onAcceptDeliverSm(deliverSm()));
+ assertEquals("the route blew up", thrown.getMessage());
+
+ verify(consumer).releaseExchange(exchange, false);
+ }
+
+ @Test
+ void dataSmTakesItsExchangeFromTheConsumerAndReleasesIt() throws Exception
{
+ DataSm dataSm = new DataSm();
+
+ listener(received -> {
+ }).onAcceptDataSm(dataSm, mock(Session.class));
+
+ verify(consumer).createExchange(false);
+ verify(consumer).releaseExchange(exchange, false);
+ }
+
+ /**
+ * The exchange is taken from the factory before the message is decoded,
and the caller's own {@code finally} cannot
+ * release it because its variable is still unassigned when creation
throws. A malformed delivery receipt would
+ * otherwise cost a pooled exchange on every occurrence.
+ */
+ @Test
+ void deliverSmReleasesItsExchangeWhenTheMessageCannotBeDecoded() throws
Exception {
+ SmppBinding failing = mock(SmppBinding.class);
+ when(failing.createSmppMessage(any(CamelContext.class),
any(DeliverSm.class)))
+ .thenThrow(new IOException("malformed delivery receipt"));
+ when(endpoint.getBinding()).thenReturn(failing);
+
+ ExceptionHandler handler = mock(ExceptionHandler.class);
+ new MessageReceiverListenerImpl(consumer, endpoint, received -> {
+ }, handler).onAcceptDeliverSm(deliverSm());
+
+ verify(consumer).releaseExchange(exchange, false);
+ verify(handler).handleException(anyString(), any(Exception.class));
+ }
+
+ /**
+ * In transceiver mode - {@code SmppProducer} with {@code
messageReceiverRouteId} - the consumer belongs to the
+ * receiver route rather than to the SMPP endpoint, so a received message
now reports that route's endpoint as
+ * {@code fromEndpoint}, the way alert notifications already did. The SMPP
endpoint's exchange pattern still has to
+ * win over the receiver endpoint's, which is why the pattern is set
explicitly.
+ */
+ @Test
+ void transceiverModeUsesTheReceiverRouteConsumerAndKeepsTheSmppPattern()
throws Exception {
+ AtomicReference<Exchange> received = new AtomicReference<>();
+ DefaultCamelContext trxContext = new DefaultCamelContext();
+ trxContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:messageReceiver").routeId("messageReceiver").process(received::set);
+ }
+ });
+
+ when(endpoint.getCamelContext()).thenReturn(trxContext);
+ when(endpoint.getExchangePattern()).thenReturn(ExchangePattern.InOut);
+
+ // constructed before the context starts, so its StartupListener
resolves the receiver route's consumer
+ MessageReceiverListenerImpl listener = new
MessageReceiverListenerImpl(endpoint, "messageReceiver");
+ trxContext.start();
+ try {
+ listener.onAcceptDeliverSm(deliverSm());
+
+ assertNotNull(received.get(), "the receiver route should have been
given the message");
+ assertEquals("direct://messageReceiver",
received.get().getFromEndpoint().getEndpointUri());
+ assertEquals(ExchangePattern.InOut, received.get().getPattern(),
+ "the SMPP endpoint's pattern must win over the receiver
endpoint's");
+ } finally {
+ trxContext.stop();
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 8313a4b93338..46a3d152e3c9 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -591,6 +591,22 @@ falls back to its logging handler for such an exchange
rather than bridging it a
the consumer itself no longer stays silent, and that with
`consumerProcessedStrategy=delete` the loss is now visible:
the document is removed during the poll, before the route runs.
+=== camel-smpp - received exchanges now come from the consumer
+
+Exchanges for received `deliver_sm` and `data_sm` messages are now created
through
+`Consumer#createExchange(boolean)` instead of on the endpoint, so that the
configured `ExchangeFactory` sees them.
+Alert notifications already worked this way. Two consequences:
+
+* `Exchange#getFromRouteId()` is now set on received messages, where it was
previously `null`, and exchange pooling
+ applies to them.
+* In transceiver mode - an `smpp` producer with `messageReceiverRouteId` - the
consumer belongs to the receiver
+ route rather than to the SMPP endpoint, so `Exchange#getFromEndpoint()` now
reports that route's endpoint (for
+ example `direct://messageReceiver`) instead of the SMPP endpoint. This
matches what alert notifications already
+ did. The exchange pattern is unchanged: the SMPP endpoint's pattern is still
applied explicitly.
+
+`SmppEndpoint#createOnAcceptDeliverSmExchange` and
`SmppEndpoint#createOnAcceptDataSm` are deprecated. They are no
+longer used by the component; build such an exchange through the consumer
instead.
+
=== camel-console
The `context` developer console no longer counts routes created by Kamelets in
its `routesTotal` and `routesStarted`
