You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2021/02/22 09:48:16 UTC
[camel] branch exchange-factory updated (966be0e -> c46d265)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch exchange-factory
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 966be0e CAMEL-16222: PooledExchangeFactory experiment
new 2dbbf05 CAMEL-16222: PooledExchangeFactory experiment
new 6824daa CAMEL-16222: PooledExchangeFactory experiment
new c46d265 CAMEL-16222: PooledExchangeFactory experiment
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/camel/component/nats/NatsConsumer.java | 2 +-
.../camel/oaipmh/handler/AbstractHandler.java | 1 -
.../org/apache/camel/oaipmh/handler/Harvester.java | 6 --
.../reactive/streams/ReactiveStreamsConsumer.java | 4 +-
.../sip/listener/SipSubscriptionListener.java | 8 +-
.../camel/component/slack/SlackConsumer.java | 22 +++-
.../camel/component/slack/SlackEndpoint.java | 26 -----
.../smpp/MessageReceiverListenerImpl.java | 24 ++++-
.../apache/camel/component/smpp/SmppConsumer.java | 3 +-
.../apache/camel/component/smpp/SmppEndpoint.java | 28 -----
.../smpp/MessageReceiverListenerImplTest.java | 117 ---------------------
.../camel/component/smpp/SmppConsumerTest.java | 12 +++
.../apache/camel/component/snmp/SnmpEndpoint.java | 14 ---
.../camel/component/snmp/SnmpTrapConsumer.java | 17 ++-
.../component/SoroushBotAbstractConsumer.java | 38 +++----
.../component/SoroushBotMultiThreadConsumer.java | 1 +
.../component/SoroushBotSingleThreadConsumer.java | 1 +
.../camel/component/splunk/SplunkConsumer.java | 28 ++---
.../camel/component/cron/SpringCronConsumer.java | 2 +-
.../apache/camel/component/sql/SqlConsumer.java | 9 +-
.../apache/camel/component/ssh/SshConsumer.java | 26 ++---
.../camel/component/stomp/StompEndpoint.java | 10 +-
.../camel/component/stream/StreamConsumer.java | 12 ++-
.../camel/component/stream/StreamEndpoint.java | 9 --
24 files changed, 139 insertions(+), 281 deletions(-)
delete mode 100644 components/camel-smpp/src/test/java/org/apache/camel/component/smpp/MessageReceiverListenerImplTest.java
[camel] 02/03: CAMEL-16222: PooledExchangeFactory experiment
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch exchange-factory
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 6824daa4316007724f6a139a120b4007278b2ee6
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Feb 22 10:45:18 2021 +0100
CAMEL-16222: PooledExchangeFactory experiment
---
.../sip/listener/SipSubscriptionListener.java | 7 +-
.../camel/component/slack/SlackConsumer.java | 23 +++-
.../camel/component/slack/SlackEndpoint.java | 26 -----
.../smpp/MessageReceiverListenerImpl.java | 24 ++++-
.../apache/camel/component/smpp/SmppConsumer.java | 3 +-
.../apache/camel/component/smpp/SmppEndpoint.java | 28 -----
.../smpp/MessageReceiverListenerImplTest.java | 117 ---------------------
.../camel/component/smpp/SmppConsumerTest.java | 12 +++
.../apache/camel/component/snmp/SnmpEndpoint.java | 14 ---
.../camel/component/snmp/SnmpTrapConsumer.java | 17 ++-
.../component/SoroushBotAbstractConsumer.java | 38 +++----
.../component/SoroushBotMultiThreadConsumer.java | 1 +
.../component/SoroushBotSingleThreadConsumer.java | 1 +
.../camel/component/splunk/SplunkConsumer.java | 28 ++---
.../camel/component/cron/SpringCronConsumer.java | 2 +-
.../apache/camel/component/sql/SqlConsumer.java | 8 +-
.../apache/camel/component/ssh/SshConsumer.java | 26 ++---
.../camel/component/stomp/StompEndpoint.java | 10 +-
.../camel/component/stream/StreamConsumer.java | 12 ++-
.../camel/component/stream/StreamEndpoint.java | 9 --
20 files changed, 137 insertions(+), 269 deletions(-)
diff --git a/components/camel-sip/src/main/java/org/apache/camel/component/sip/listener/SipSubscriptionListener.java b/components/camel-sip/src/main/java/org/apache/camel/component/sip/listener/SipSubscriptionListener.java
index 7951940..6738ed8 100644
--- a/components/camel-sip/src/main/java/org/apache/camel/component/sip/listener/SipSubscriptionListener.java
+++ b/components/camel-sip/src/main/java/org/apache/camel/component/sip/listener/SipSubscriptionListener.java
@@ -48,14 +48,15 @@ public class SipSubscriptionListener implements SipListener {
this.setSipSubscriber(sipSubscriber);
}
- private void dispatchExchange(Object response) throws CamelException {
+ private void dispatchExchange(Object response) {
LOG.debug("Consumer Dispatching the received notification along the route");
- Exchange exchange = sipSubscriber.getEndpoint().createExchange(ExchangePattern.InOnly);
+ Exchange exchange = sipSubscriber.createExchange(true);
+ exchange.setPattern(ExchangePattern.InOnly);
exchange.getIn().setBody(response);
try {
sipSubscriber.getProcessor().process(exchange);
} catch (Exception e) {
- throw new CamelException("Error in consumer while dispatching exchange", e);
+ sipSubscriber.getExceptionHandler().handleException("Error in consumer while dispatching exchange", e);
}
}
diff --git a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
index 396e66e..8ea304c 100644
--- a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
+++ b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
@@ -25,8 +25,11 @@ import java.util.List;
import java.util.Queue;
import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.slack.helper.SlackMessage;
import org.apache.camel.support.ScheduledBatchPollingConsumer;
import org.apache.camel.util.CastUtils;
import org.apache.camel.util.ObjectHelper;
@@ -109,7 +112,7 @@ public class SlackConsumer extends ScheduledBatchPollingConsumer {
timestamp = (String) singleMess.get("ts");
}
i++;
- Exchange exchange = slackEndpoint.createExchange(singleMess);
+ Exchange exchange = createExchange(singleMess);
answer.add(exchange);
}
}
@@ -185,4 +188,22 @@ public class SlackConsumer extends ScheduledBatchPollingConsumer {
}
}
+ public Exchange createExchange(JsonObject object) {
+ Exchange exchange = createExchange(true);
+ SlackMessage slackMessage = new SlackMessage();
+ String text = object.getString(SlackConstants.SLACK_TEXT_FIELD);
+ String user = object.getString("user");
+ slackMessage.setText(text);
+ slackMessage.setUser(user);
+ if (ObjectHelper.isNotEmpty(object.get("icons"))) {
+ JsonObject icons = object.getMap("icons");
+ if (ObjectHelper.isNotEmpty(icons.get("emoji"))) {
+ slackMessage.setIconEmoji(icons.getString("emoji"));
+ }
+ }
+ Message message = exchange.getIn();
+ message.setBody(slackMessage);
+ return exchange;
+ }
+
}
diff --git a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackEndpoint.java b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackEndpoint.java
index c5bf928..9893b21 100644
--- a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackEndpoint.java
+++ b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackEndpoint.java
@@ -18,20 +18,15 @@ package org.apache.camel.component.slack;
import org.apache.camel.Category;
import org.apache.camel.Consumer;
-import org.apache.camel.Exchange;
-import org.apache.camel.ExchangePattern;
-import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.component.slack.helper.SlackMessage;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.support.ScheduledPollEndpoint;
import org.apache.camel.util.ObjectHelper;
-import org.apache.camel.util.json.JsonObject;
/**
* Send and receive messages to/from Slack.
@@ -183,25 +178,4 @@ public class SlackEndpoint extends ScheduledPollEndpoint {
this.serverUrl = serverUrl;
}
- public Exchange createExchange(JsonObject object) {
- return createExchange(getExchangePattern(), object);
- }
-
- public Exchange createExchange(ExchangePattern pattern, JsonObject object) {
- Exchange exchange = super.createExchange(pattern);
- SlackMessage slackMessage = new SlackMessage();
- String text = object.getString(SlackConstants.SLACK_TEXT_FIELD);
- String user = object.getString("user");
- slackMessage.setText(text);
- slackMessage.setUser(user);
- if (ObjectHelper.isNotEmpty(object.get("icons"))) {
- JsonObject icons = object.getMap("icons");
- if (ObjectHelper.isNotEmpty(icons.get("emoji"))) {
- slackMessage.setIconEmoji(icons.getString("emoji"));
- }
- }
- Message message = exchange.getIn();
- message.setBody(slackMessage);
- return exchange;
- }
}
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 c076335..9eb4e2e 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
@@ -17,6 +17,7 @@
package org.apache.camel.component.smpp;
import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
import org.apache.camel.Processor;
import org.apache.camel.spi.ExceptionHandler;
import org.jsmpp.bean.AlertNotification;
@@ -37,11 +38,14 @@ public class MessageReceiverListenerImpl implements MessageReceiverListener {
private static final Logger LOG = LoggerFactory.getLogger(MessageReceiverListenerImpl.class);
private MessageIDGenerator messageIDGenerator = new RandomMessageIDGenerator();
+ private SmppConsumer consumer;
private SmppEndpoint endpoint;
private Processor processor;
private ExceptionHandler exceptionHandler;
- public MessageReceiverListenerImpl(SmppEndpoint endpoint, Processor processor, ExceptionHandler exceptionHandler) {
+ public MessageReceiverListenerImpl(SmppConsumer consumer, SmppEndpoint endpoint, Processor processor,
+ ExceptionHandler exceptionHandler) {
+ this.consumer = consumer;
this.endpoint = endpoint;
this.processor = processor;
this.exceptionHandler = exceptionHandler;
@@ -51,7 +55,7 @@ public class MessageReceiverListenerImpl implements MessageReceiverListener {
public void onAcceptAlertNotification(AlertNotification alertNotification) {
LOG.debug("Received an alertNotification {}", alertNotification);
- Exchange exchange = endpoint.createOnAcceptAlertNotificationExchange(alertNotification);
+ Exchange exchange = createOnAcceptAlertNotificationExchange(alertNotification);
try {
processor.process(exchange);
} catch (Exception e) {
@@ -62,6 +66,7 @@ public class MessageReceiverListenerImpl implements MessageReceiverListener {
exceptionHandler.handleException("Cannot process exchange. This exception will be ignored.", exchange,
exchange.getException());
}
+ consumer.releaseExchange(exchange, false);
}
@Override
@@ -117,4 +122,19 @@ public class MessageReceiverListenerImpl implements MessageReceiverListener {
public void setMessageIDGenerator(MessageIDGenerator messageIDGenerator) {
this.messageIDGenerator = messageIDGenerator;
}
+
+ /**
+ * 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 alertNotification the received message from the SMSC
+ * @return a new exchange
+ */
+ 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;
+ }
+
}
diff --git a/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/SmppConsumer.java b/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/SmppConsumer.java
index dee6ec3..c04c7f1 100644
--- a/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/SmppConsumer.java
+++ b/components/camel-smpp/src/main/java/org/apache/camel/component/smpp/SmppConsumer.java
@@ -72,7 +72,8 @@ public class SmppConsumer extends DefaultConsumer {
}
}
};
- this.messageReceiverListener = new MessageReceiverListenerImpl(getEndpoint(), getProcessor(), getExceptionHandler());
+ this.messageReceiverListener
+ = new MessageReceiverListenerImpl(this, getEndpoint(), getProcessor(), getExceptionHandler());
}
@Override
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 f9751e0..6f3b6d2 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
@@ -26,7 +26,6 @@ import org.apache.camel.Producer;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.support.DefaultEndpoint;
-import org.jsmpp.bean.AlertNotification;
import org.jsmpp.bean.DataSm;
import org.jsmpp.bean.DeliverSm;
@@ -71,33 +70,6 @@ public class SmppEndpoint extends DefaultEndpoint {
/**
* Create a new exchange for communicating with this endpoint from a SMSC
*
- * @param alertNotification the received message from the SMSC
- * @return a new exchange
- */
- public Exchange createOnAcceptAlertNotificationExchange(AlertNotification alertNotification) {
- return createOnAcceptAlertNotificationExchange(getExchangePattern(), alertNotification);
- }
-
- /**
- * 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 alertNotification the received message from the SMSC
- * @return a new exchange
- */
- public Exchange createOnAcceptAlertNotificationExchange(
- ExchangePattern exchangePattern,
- AlertNotification alertNotification) {
- Exchange exchange = createExchange(exchangePattern);
- exchange.setProperty(Exchange.BINDING, getBinding());
- exchange.setIn(getBinding().createSmppMessage(getCamelContext(), alertNotification));
- return exchange;
- }
-
- /**
- * Create a new exchange for communicating with this endpoint from a SMSC
- *
* @param deliverSm the received message from the SMSC
* @return a new exchange
*/
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
deleted file mode 100644
index a34f6c8..0000000
--- a/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/MessageReceiverListenerImplTest.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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 org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.spi.ExceptionHandler;
-import org.jsmpp.PDUStringException;
-import org.jsmpp.bean.AlertNotification;
-import org.jsmpp.bean.DataSm;
-import org.jsmpp.bean.DeliverSm;
-import org.jsmpp.bean.OptionalParameter;
-import org.jsmpp.session.DataSmResult;
-import org.jsmpp.session.SMPPSession;
-import org.jsmpp.util.MessageIDGenerator;
-import org.jsmpp.util.MessageId;
-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.assertSame;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-public class MessageReceiverListenerImplTest {
-
- private MessageReceiverListenerImpl listener;
- private SmppEndpoint endpoint;
- private Processor processor;
- private ExceptionHandler exceptionHandler;
-
- @BeforeEach
- public void setUp() {
- endpoint = mock(SmppEndpoint.class);
- processor = mock(Processor.class);
- exceptionHandler = mock(ExceptionHandler.class);
-
- listener = new MessageReceiverListenerImpl(endpoint, processor, exceptionHandler);
- listener.setMessageIDGenerator(new MessageIDGenerator() {
- public MessageId newMessageId() {
- try {
- return new MessageId("1");
- } catch (PDUStringException e) {
- throw new RuntimeException(e);
- }
- }
- });
- }
-
- @Test
- public void onAcceptAlertNotificationSuccess() throws Exception {
- AlertNotification alertNotification = mock(AlertNotification.class);
- Exchange exchange = mock(Exchange.class);
-
- when(endpoint.createOnAcceptAlertNotificationExchange(alertNotification))
- .thenReturn(exchange);
- when(exchange.getException()).thenReturn(null);
-
- listener.onAcceptAlertNotification(alertNotification);
-
- verify(endpoint).createOnAcceptAlertNotificationExchange(alertNotification);
- verify(processor).process(exchange);
- }
-
- @Test
- public void onAcceptDeliverSmException() throws Exception {
- DeliverSm deliverSm = mock(DeliverSm.class);
- Exchange exchange = mock(Exchange.class);
-
- when(endpoint.createOnAcceptDeliverSmExchange(deliverSm))
- .thenReturn(exchange);
- when(exchange.getException()).thenReturn(null);
-
- listener.onAcceptDeliverSm(deliverSm);
-
- verify(endpoint).createOnAcceptDeliverSmExchange(deliverSm);
- verify(processor).process(exchange);
- }
-
- @Test
- public void onAcceptDataSmSuccess() throws Exception {
- SMPPSession session = mock(SMPPSession.class);
- DataSm dataSm = mock(DataSm.class);
- Exchange exchange = mock(Exchange.class);
- OptionalParameter[] optionalParameters = new OptionalParameter[] {};
-
- when(endpoint.createOnAcceptDataSm(dataSm, "1"))
- .thenReturn(exchange);
- when(exchange.getException()).thenReturn(null);
- when(dataSm.getOptionalParameters())
- .thenReturn(optionalParameters);
-
- DataSmResult result = listener.onAcceptDataSm(dataSm, session);
-
- verify(endpoint).createOnAcceptDataSm(dataSm, "1");
- verify(processor).process(exchange);
-
- assertEquals("1", result.getMessageId());
- assertSame(optionalParameters, result.getOptionalParameters());
- }
-
-}
diff --git a/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/SmppConsumerTest.java b/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/SmppConsumerTest.java
index 6404991..2274dbd 100644
--- a/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/SmppConsumerTest.java
+++ b/components/camel-smpp/src/test/java/org/apache/camel/component/smpp/SmppConsumerTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.component.smpp;
+import org.apache.camel.ExtendedCamelContext;
import org.apache.camel.Processor;
+import org.apache.camel.spi.ExchangeFactory;
import org.jsmpp.bean.BindType;
import org.jsmpp.bean.NumberingPlanIndicator;
import org.jsmpp.bean.TypeOfNumber;
@@ -28,6 +30,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.reset;
@@ -39,6 +42,8 @@ import static org.mockito.Mockito.when;
*/
public class SmppConsumerTest {
+ private ExchangeFactory exchangeFactory;
+ private ExtendedCamelContext context;
private SmppConsumer consumer;
private SmppEndpoint endpoint;
private SmppConfiguration configuration;
@@ -51,10 +56,17 @@ public class SmppConsumerTest {
configuration.setServiceType("CMT");
configuration.setSystemType("cp");
configuration.setPassword("password");
+ context = mock(ExtendedCamelContext.class);
+ exchangeFactory = mock(ExchangeFactory.class);
endpoint = mock(SmppEndpoint.class);
processor = mock(Processor.class);
session = mock(SMPPSession.class);
+ when(endpoint.getCamelContext()).thenReturn(context);
+ when(context.adapt(ExtendedCamelContext.class)).thenReturn(context);
+ when(context.getExchangeFactory()).thenReturn(exchangeFactory);
+ when(exchangeFactory.newExchangeFactory(any())).thenReturn(exchangeFactory);
+
// the construction of SmppConsumer will trigger the getCamelContext call
consumer = new SmppConsumer(
endpoint,
diff --git a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java
index 3cf11b7..69303b4 100644
--- a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java
+++ b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java
@@ -30,7 +30,6 @@ import org.apache.camel.spi.UriPath;
import org.apache.camel.support.DefaultPollingEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.snmp4j.CommandResponderEvent;
import org.snmp4j.PDU;
import org.snmp4j.mp.SnmpConstants;
import org.snmp4j.security.SecurityLevel;
@@ -140,19 +139,6 @@ public class SnmpEndpoint extends DefaultPollingEndpoint {
}
/**
- * creates an exchange for the given message
- *
- * @param pdu the pdu
- * @param event a snmp4j CommandResponderEvent
- * @return an exchange
- */
- public Exchange createExchange(PDU pdu, CommandResponderEvent event) {
- Exchange exchange = super.createExchange();
- exchange.setIn(new SnmpMessage(getCamelContext(), pdu, event));
- return exchange;
- }
-
- /**
* creates and configures the endpoint
*
* @throws Exception if unable to setup connection
diff --git a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpTrapConsumer.java b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpTrapConsumer.java
index 4639e8c..0de0f65 100644
--- a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpTrapConsumer.java
+++ b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpTrapConsumer.java
@@ -137,7 +137,7 @@ public class SnmpTrapConsumer extends DefaultConsumer implements CommandResponde
if (LOG.isDebugEnabled()) {
LOG.debug("Received trap event for {} : {}", this.endpoint.getAddress(), pdu);
}
- Exchange exchange = endpoint.createExchange(pdu, event);
+ Exchange exchange = createExchange(pdu, event);
try {
getProcessor().process(exchange);
} catch (Exception e) {
@@ -146,5 +146,20 @@ public class SnmpTrapConsumer extends DefaultConsumer implements CommandResponde
if (exchange.getException() != null) {
getExceptionHandler().handleException(exchange.getException());
}
+ releaseExchange(exchange, false);
}
+
+ /**
+ * creates an exchange for the given message
+ *
+ * @param pdu the pdu
+ * @param event a snmp4j CommandResponderEvent
+ * @return an exchange
+ */
+ public Exchange createExchange(PDU pdu, CommandResponderEvent event) {
+ Exchange exchange = createExchange(false);
+ exchange.setIn(new SnmpMessage(getEndpoint().getCamelContext(), pdu, event));
+ return exchange;
+ }
+
}
diff --git a/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotAbstractConsumer.java b/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotAbstractConsumer.java
index fbff371..c9340e7 100644
--- a/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotAbstractConsumer.java
+++ b/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotAbstractConsumer.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.soroushbot.component;
-import java.io.IOException;
import java.util.concurrent.TimeUnit;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -65,15 +64,6 @@ public abstract class SoroushBotAbstractConsumer extends DefaultConsumer impleme
run();
}
- protected final void handleExceptionThrownWhileCreatingOrProcessingExchange(
- Exchange exchange, SoroushMessage soroushMessage, Exception ex) {
- //set originalMessage property to the created soroushMessage to let Error Handler access the message
- exchange.setProperty("OriginalMessage", soroushMessage);
- //use this instead of handleException() to manually set the exchange.
- getExceptionHandler().handleException("message can not be processed due to :" + ex.getMessage(), exchange, ex);
-
- }
-
/**
* handle how processing of the exchange should be started
*
@@ -142,25 +132,21 @@ public abstract class SoroushBotAbstractConsumer extends DefaultConsumer impleme
@Override
public void onEvent(EventSource eventSource, String id, String type, String data) {
- Exchange exchange = endpoint.createExchange();
+ Exchange exchange = createExchange(false);
try {
SoroushMessage soroushMessage = objectMapper.readValue(data, SoroushMessage.class);
- try {
- exchange.getIn().setBody(soroushMessage);
- if (LOG.isDebugEnabled()) {
- LOG.debug("event data is: " + data);
- }
- // if autoDownload is true, download the resource if provided in the message
- if (endpoint.isAutoDownload()) {
- endpoint.handleDownloadFiles(soroushMessage);
- }
- //let each subclass decide how to start processing of each exchange
- sendExchange(exchange);
- } catch (Exception ex) {
- handleExceptionThrownWhileCreatingOrProcessingExchange(exchange, soroushMessage, ex);
+ exchange.getIn().setBody(soroushMessage);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("event data is: " + data);
+ }
+ // if autoDownload is true, download the resource if provided in the message
+ if (endpoint.isAutoDownload()) {
+ endpoint.handleDownloadFiles(soroushMessage);
}
- } catch (IOException ex) {
- LOG.error("can not parse data due to following error", ex);
+ //let each subclass decide how to start processing of each exchange
+ sendExchange(exchange);
+ } catch (Exception ex) {
+ getExceptionHandler().handleException(ex);
}
}
diff --git a/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotMultiThreadConsumer.java b/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotMultiThreadConsumer.java
index 96cecac..4badfee 100644
--- a/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotMultiThreadConsumer.java
+++ b/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotMultiThreadConsumer.java
@@ -57,6 +57,7 @@ public class SoroushBotMultiThreadConsumer extends SoroushBotAbstractConsumer {
getExceptionHandler().handleException("Error processing exchange",
exchange, exchange.getException());
}
+ releaseExchange(exchange, false);
});
} catch (IllegalStateException ex) {
throw new CongestionException(ex, exchange.getIn().getBody(SoroushMessage.class));
diff --git a/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotSingleThreadConsumer.java b/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotSingleThreadConsumer.java
index 5926b3b..1ed907c 100644
--- a/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotSingleThreadConsumer.java
+++ b/components/camel-soroush/src/main/java/org/apache/camel/component/soroushbot/component/SoroushBotSingleThreadConsumer.java
@@ -42,5 +42,6 @@ public class SoroushBotSingleThreadConsumer extends SoroushBotAbstractConsumer {
getExceptionHandler().handleException("Error processing exchange",
exchange, exchange.getException());
}
+ releaseExchange(exchange, false);
}
}
diff --git a/components/camel-splunk/src/main/java/org/apache/camel/component/splunk/SplunkConsumer.java b/components/camel-splunk/src/main/java/org/apache/camel/component/splunk/SplunkConsumer.java
index dca8f97..72834d2 100644
--- a/components/camel-splunk/src/main/java/org/apache/camel/component/splunk/SplunkConsumer.java
+++ b/components/camel-splunk/src/main/java/org/apache/camel/component/splunk/SplunkConsumer.java
@@ -66,28 +66,18 @@ public class SplunkConsumer extends ScheduledBatchPollingConsumer {
@Override
public void process(SplunkEvent splunkEvent) {
- final Exchange exchange = getEndpoint().createExchange();
+ final Exchange exchange = createExchange(true);
Message message = exchange.getIn();
message.setBody(splunkEvent);
- try {
- LOG.trace("Processing exchange [{}]...", exchange);
- getAsyncProcessor().process(exchange, new AsyncCallback() {
- @Override
- public void done(boolean doneSync) {
- LOG.trace("Done processing exchange [{}]...", exchange);
- }
- });
- } catch (Exception e) {
- exchange.setException(e);
- }
- if (exchange.getException() != null) {
- getExceptionHandler().handleException("Error processing exchange", exchange,
- exchange.getException());
- }
-
+ LOG.trace("Processing exchange [{}]...", exchange);
+ getAsyncProcessor().process(exchange, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ LOG.trace("Done processing exchange [{}]...", exchange);
+ }
+ });
}
-
});
// Return 0: no exchanges returned by poll, as exchanges have been returned asynchronously
return 0;
@@ -107,7 +97,7 @@ public class SplunkConsumer extends ScheduledBatchPollingConsumer {
LOG.trace("Received {} messages in this poll", splunkEvents.size());
Queue<Exchange> answer = new LinkedList<>();
for (SplunkEvent splunkEvent : splunkEvents) {
- Exchange exchange = getEndpoint().createExchange();
+ Exchange exchange = createExchange(true);
Message message = exchange.getIn();
message.setBody(splunkEvent);
answer.add(exchange);
diff --git a/components/camel-spring/src/main/java/org/apache/camel/component/cron/SpringCronConsumer.java b/components/camel-spring/src/main/java/org/apache/camel/component/cron/SpringCronConsumer.java
index 217d9e7..e60ad73 100644
--- a/components/camel-spring/src/main/java/org/apache/camel/component/cron/SpringCronConsumer.java
+++ b/components/camel-spring/src/main/java/org/apache/camel/component/cron/SpringCronConsumer.java
@@ -29,7 +29,7 @@ public class SpringCronConsumer extends ScheduledPollConsumer {
@Override
protected int poll() throws Exception {
- Exchange exchange = getEndpoint().createExchange();
+ Exchange exchange = createExchange(true);
getProcessor().process(exchange);
return 1;
}
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
index 036e9b7..145b7f1 100644
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
@@ -231,7 +231,7 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
}
protected Exchange createExchange(Object data) {
- final Exchange exchange = getEndpoint().createExchange(ExchangePattern.InOnly);
+ final Exchange exchange = createExchange(false);
Message msg = exchange.getIn();
if (getEndpoint().getOutputHeader() != null) {
msg.setHeader(getEndpoint().getOutputHeader(), data);
@@ -274,10 +274,12 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
if (getEndpoint().isTransacted() && exchange.isFailed()) {
// break out as we are transacted and should rollback
Exception cause = exchange.getException();
+ // must release exchange
+ releaseExchange(exchange, false);
if (cause != null) {
throw cause;
} else {
- throw new RollbackExchangeException("Rollback transaction due error processing exchange", exchange);
+ throw new RollbackExchangeException("Rollback transaction due error processing exchange", null);
}
}
@@ -306,6 +308,8 @@ public class SqlConsumer extends ScheduledBatchPollingConsumer {
} else {
handleException("Error executing onConsume/onConsumeFailed query " + sql, e);
}
+ } finally {
+ releaseExchange(exchange, false);
}
}
diff --git a/components/camel-ssh/src/main/java/org/apache/camel/component/ssh/SshConsumer.java b/components/camel-ssh/src/main/java/org/apache/camel/component/ssh/SshConsumer.java
index 8f7c229..a4b29f1 100644
--- a/components/camel-ssh/src/main/java/org/apache/camel/component/ssh/SshConsumer.java
+++ b/components/camel-ssh/src/main/java/org/apache/camel/component/ssh/SshConsumer.java
@@ -56,22 +56,21 @@ public class SshConsumer extends ScheduledPollConsumer {
}
String command = endpoint.getPollCommand();
- Exchange exchange = endpoint.createExchange();
-
- String knownHostResource = endpoint.getKnownHostsResource();
- if (knownHostResource != null) {
- client.setServerKeyVerifier(new ResourceBasedSSHKeyVerifier(
- exchange.getContext(), knownHostResource,
- endpoint.isFailOnUnknownHost()));
- }
+ Exchange exchange = createExchange(false);
+ try {
+ String knownHostResource = endpoint.getKnownHostsResource();
+ if (knownHostResource != null) {
+ client.setServerKeyVerifier(new ResourceBasedSSHKeyVerifier(
+ exchange.getContext(), knownHostResource,
+ endpoint.isFailOnUnknownHost()));
+ }
- SshResult result = SshHelper.sendExecCommand(exchange.getIn().getHeaders(), command, endpoint, client);
+ SshResult result = SshHelper.sendExecCommand(exchange.getIn().getHeaders(), command, endpoint, client);
- exchange.getIn().setBody(result.getStdout());
- exchange.getIn().setHeader(SshResult.EXIT_VALUE, result.getExitValue());
- exchange.getIn().setHeader(SshResult.STDERR, result.getStderr());
+ exchange.getIn().setBody(result.getStdout());
+ exchange.getIn().setHeader(SshResult.EXIT_VALUE, result.getExitValue());
+ exchange.getIn().setHeader(SshResult.STDERR, result.getStderr());
- try {
// send message to next processor in the route
getProcessor().process(exchange);
return 1; // number of messages polled
@@ -80,6 +79,7 @@ public class SshConsumer extends ScheduledPollConsumer {
if (exchange.getException() != null) {
getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException());
}
+ releaseExchange(exchange, false);
}
}
}
diff --git a/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java b/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java
index e1143d6..d6f122d 100644
--- a/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java
+++ b/components/camel-stomp/src/main/java/org/apache/camel/component/stomp/StompEndpoint.java
@@ -133,13 +133,15 @@ public class StompEndpoint extends DefaultEndpoint implements AsyncEndpoint, Hea
@Override
public void onSuccess(StompFrame value) {
if (!consumers.isEmpty()) {
- Exchange exchange = createExchange();
- exchange.getIn().setBody(value.content());
- exchange.getIn().setHeaders(value.headerMap().entrySet().stream()
- .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)));
for (StompConsumer consumer : consumers) {
+ Exchange exchange = consumer.createExchange(false);
+ exchange.getIn().setBody(value.content());
+ exchange.getIn().setHeaders(value.headerMap().entrySet().stream()
+ .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)));
consumer.processExchange(exchange);
+ consumer.releaseExchange(exchange, false);
}
+
}
}
});
diff --git a/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java b/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
index 8416abf..a88886a 100644
--- a/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
+++ b/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
@@ -237,7 +237,7 @@ public class StreamConsumer extends DefaultConsumer implements Runnable {
List<String> copy = new ArrayList<>(lines);
Object body = endpoint.getGroupStrategy().groupLines(copy);
// remember to inc index when we create an exchange
- Exchange exchange = endpoint.createExchange(body, index++, last);
+ Exchange exchange = createExchange(body, index++, last);
// clear lines
lines.clear();
@@ -247,7 +247,7 @@ public class StreamConsumer extends DefaultConsumer implements Runnable {
} else if (line != null) {
// single line
// remember to inc index when we create an exchange
- Exchange exchange = endpoint.createExchange(line, index++, last);
+ Exchange exchange = createExchange(line, index++, last);
getProcessor().process(exchange);
}
@@ -333,4 +333,12 @@ public class StreamConsumer extends DefaultConsumer implements Runnable {
}
}
+ protected Exchange createExchange(Object body, long index, boolean last) {
+ Exchange exchange = createExchange(true);
+ exchange.getIn().setBody(body);
+ exchange.getIn().setHeader(StreamConstants.STREAM_INDEX, index);
+ exchange.getIn().setHeader(StreamConstants.STREAM_COMPLETE, last);
+ return exchange;
+ }
+
}
diff --git a/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java b/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
index cf1f7c3..8a3a675 100644
--- a/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
+++ b/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
@@ -21,7 +21,6 @@ import java.nio.charset.Charset;
import org.apache.camel.Category;
import org.apache.camel.Component;
import org.apache.camel.Consumer;
-import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.spi.Metadata;
@@ -96,14 +95,6 @@ public class StreamEndpoint extends DefaultEndpoint {
return new StreamProducer(this, getEndpointUri());
}
- protected Exchange createExchange(Object body, long index, boolean last) {
- Exchange exchange = createExchange();
- exchange.getIn().setBody(body);
- exchange.getIn().setHeader(StreamConstants.STREAM_INDEX, index);
- exchange.getIn().setHeader(StreamConstants.STREAM_COMPLETE, last);
- return exchange;
- }
-
// Properties
//-------------------------------------------------------------------------
[camel] 03/03: CAMEL-16222: PooledExchangeFactory experiment
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch exchange-factory
in repository https://gitbox.apache.org/repos/asf/camel.git
commit c46d26549253ad71f2dcb66402d20cf845f02a6b
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Feb 22 10:47:35 2021 +0100
CAMEL-16222: PooledExchangeFactory experiment
---
.../src/main/java/org/apache/camel/component/nats/NatsConsumer.java | 2 +-
.../main/java/org/apache/camel/oaipmh/handler/AbstractHandler.java | 1 -
.../src/main/java/org/apache/camel/oaipmh/handler/Harvester.java | 6 ------
.../camel/component/sip/listener/SipSubscriptionListener.java | 1 -
.../main/java/org/apache/camel/component/slack/SlackConsumer.java | 1 -
.../src/main/java/org/apache/camel/component/sql/SqlConsumer.java | 1 -
6 files changed, 1 insertion(+), 11 deletions(-)
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
index 0c78677..84cc5c1 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
@@ -174,7 +174,7 @@ public class NatsConsumer extends DefaultConsumer {
con.publish(msg.getReplyTo(), data);
}
}
- } catch (Exception e){
+ } catch (Exception e) {
getExceptionHandler().handleException("Error during processing", exchange, e);
} finally {
releaseExchange(exchange, false);
diff --git a/components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/handler/AbstractHandler.java b/components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/handler/AbstractHandler.java
index c204028..a68a551 100644
--- a/components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/handler/AbstractHandler.java
+++ b/components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/handler/AbstractHandler.java
@@ -20,7 +20,6 @@ import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
-import org.apache.camel.RuntimeCamelException;
import org.apache.camel.oaipmh.component.OAIPMHConsumer;
import org.apache.camel.oaipmh.model.OAIPMHResponse;
import org.apache.camel.spi.ExceptionHandler;
diff --git a/components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/handler/Harvester.java b/components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/handler/Harvester.java
index 84a795b..b05ebc1 100644
--- a/components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/handler/Harvester.java
+++ b/components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/handler/Harvester.java
@@ -16,16 +16,10 @@
*/
package org.apache.camel.oaipmh.handler;
-import java.io.IOException;
import java.net.URI;
-import java.net.URISyntaxException;
import java.util.List;
import java.util.Optional;
-import javax.xml.parsers.ParserConfigurationException;
-
-import org.xml.sax.SAXException;
-
import org.apache.camel.oaipmh.component.model.OAIPMHVerb;
import org.apache.camel.oaipmh.model.OAIPMHResponse;
import org.apache.camel.oaipmh.utils.OAIPMHHttpClient;
diff --git a/components/camel-sip/src/main/java/org/apache/camel/component/sip/listener/SipSubscriptionListener.java b/components/camel-sip/src/main/java/org/apache/camel/component/sip/listener/SipSubscriptionListener.java
index 6738ed8..2392eb6 100644
--- a/components/camel-sip/src/main/java/org/apache/camel/component/sip/listener/SipSubscriptionListener.java
+++ b/components/camel-sip/src/main/java/org/apache/camel/component/sip/listener/SipSubscriptionListener.java
@@ -31,7 +31,6 @@ import javax.sip.header.ViaHeader;
import javax.sip.message.Request;
import javax.sip.message.Response;
-import org.apache.camel.CamelException;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.component.sip.SipSubscriber;
diff --git a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
index 8ea304c..fde7f10 100644
--- a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
+++ b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
@@ -25,7 +25,6 @@ import java.util.List;
import java.util.Queue;
import org.apache.camel.Exchange;
-import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.RuntimeCamelException;
diff --git a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
index 145b7f1..78635f4 100644
--- a/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
+++ b/components/camel-sql/src/main/java/org/apache/camel/component/sql/SqlConsumer.java
@@ -25,7 +25,6 @@ import java.util.List;
import java.util.Queue;
import org.apache.camel.Exchange;
-import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.RollbackExchangeException;
[camel] 01/03: CAMEL-16222: PooledExchangeFactory experiment
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch exchange-factory
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 2dbbf050d5ef9221f6d76736c0cbeacb51344fc0
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Feb 22 10:01:27 2021 +0100
CAMEL-16222: PooledExchangeFactory experiment
---
.../camel/component/reactive/streams/ReactiveStreamsConsumer.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
index 989b901..815c3e2 100644
--- a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
@@ -75,7 +75,7 @@ public class ReactiveStreamsConsumer extends DefaultConsumer {
public void onComplete() {
if (endpoint.isForwardOnComplete()) {
- Exchange exchange = endpoint.createExchange();
+ Exchange exchange = createExchange(true);
exchange.getIn().setHeader(ReactiveStreamsConstants.REACTIVE_STREAMS_EVENT_TYPE, "onComplete");
doSend(exchange, done -> {
@@ -85,7 +85,7 @@ public class ReactiveStreamsConsumer extends DefaultConsumer {
public void onError(Throwable error) {
if (endpoint.isForwardOnError()) {
- Exchange exchange = endpoint.createExchange();
+ Exchange exchange = createExchange(true);
exchange.getIn().setHeader(ReactiveStreamsConstants.REACTIVE_STREAMS_EVENT_TYPE, "onError");
exchange.getIn().setBody(error);