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 2024/03/23 17:00:36 UTC
(camel) branch jms-inout updated: camel-jms - Using in/out should use single atomic operation with the timeout map for the correlation id. Could help with org.apache.camel.component.jms.JmsProducerConcurrentWithReplyTest.testConcurrentProducers
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch jms-inout
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/jms-inout by this push:
new 5edc42ce7ef camel-jms - Using in/out should use single atomic operation with the timeout map for the correlation id. Could help with org.apache.camel.component.jms.JmsProducerConcurrentWithReplyTest.testConcurrentProducers
5edc42ce7ef is described below
commit 5edc42ce7ef3e4e1b5a291be43922ad38ffd26fa
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Sat Mar 23 18:00:21 2024 +0100
camel-jms - Using in/out should use single atomic operation with the timeout map for the correlation id. Could help with org.apache.camel.component.jms.JmsProducerConcurrentWithReplyTest.testConcurrentProducers
---
.../apache/camel/component/jms/reply/ReplyManagerSupport.java | 10 +++++-----
.../apache/camel/component/sjms/reply/QueueReplyManager.java | 5 +----
.../camel/component/sjms/reply/TemporaryQueueReplyManager.java | 3 +--
.../main/java/org/apache/camel/support/DefaultTimeoutMap.java | 5 +++++
4 files changed, 12 insertions(+), 11 deletions(-)
diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
index b8e12e72078..31234500eee 100644
--- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
+++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/ReplyManagerSupport.java
@@ -231,13 +231,13 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
/**
* <b>IMPORTANT:</b> This logic is only being used due to high performance in-memory only testing using InOut over
- * JMS. Its unlikely to happen in a real life situation with communication to a remote broker, which always will be
- * slower to send back reply, before Camel had a chance to update it's internal correlation map.
+ * JMS. It is unlikely to happen in a real life situation with communication to a remote broker, which always will
+ * be slower to send back reply, before Camel had a chance to update the internal correlation map.
*/
protected ReplyHandler waitForProvisionCorrelationToBeUpdated(String correlationID, Message message) {
// race condition, when using messageID as correlationID then we store a provisional correlation id
// at first, which gets updated with the JMSMessageID after the message has been sent. And in the unlikely
- // event that the reply comes back really really fast, and the correlation map hasn't yet been updated
+ // event that the reply comes back really fast, and the correlation map hasn't yet been updated
// from the provisional id to the JMSMessageID. If so we have to wait a bit and lookup again.
if (log.isWarnEnabled()) {
log.warn("Early reply received with correlationID [{}] -> {}", correlationID, message);
@@ -255,8 +255,8 @@ public abstract class ReplyManagerSupport extends ServiceSupport implements Repl
}
private ReplyHandler getReplyHandler(String correlationID) {
- log.trace("Early reply not found handler. Waiting a bit longer.");
- return correlation.get(correlationID);
+ log.trace("Early reply not found. Waiting a bit longer.");
+ return correlation.remove(correlationID); // get and remove
}
@Override
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/QueueReplyManager.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/QueueReplyManager.java
index fb67ba02739..662f40fa6cc 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/QueueReplyManager.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/QueueReplyManager.java
@@ -45,16 +45,13 @@ public class QueueReplyManager extends ReplyManagerSupport {
// should not happen that we can't find the handler
return;
}
-
correlation.put(newCorrelationId, handler, requestTimeout);
}
@Override
protected void handleReplyMessage(String correlationID, Message message, Session session) {
- ReplyHandler handler = correlation.get(correlationID);
-
+ ReplyHandler handler = correlation.remove(correlationID);
if (handler != null) {
- correlation.remove(correlationID);
handler.onReply(correlationID, message, session);
} else {
// we could not correlate the received reply message to a matching request and therefore
diff --git a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/TemporaryQueueReplyManager.java b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/TemporaryQueueReplyManager.java
index 6c3ce420c1e..98553e9a23a 100644
--- a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/TemporaryQueueReplyManager.java
+++ b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/reply/TemporaryQueueReplyManager.java
@@ -63,9 +63,8 @@ public class TemporaryQueueReplyManager extends ReplyManagerSupport {
@Override
protected void handleReplyMessage(String correlationID, Message message, Session session) {
- ReplyHandler handler = correlation.get(correlationID);
+ ReplyHandler handler = correlation.remove(correlationID);
if (handler != null) {
- correlation.remove(correlationID);
handler.onReply(correlationID, message, session);
} else {
// we could not correlate the received reply message to a matching request and therefore
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java b/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
index 5be87e40ef0..b6cda4f3ecc 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
@@ -134,6 +134,11 @@ public class DefaultTimeoutMap<K, V> extends ServiceSupport implements TimeoutMa
@Override
public V remove(K key) {
+ // if no contains, the lock is not necessary
+ if (!map.containsKey(key)) {
+ return null;
+ }
+
V value = null;
lock.lock();
try {