Author: davsclaus
Date: Wed Nov 9 13:22:55 2011
New Revision: 1199739
URL: http://svn.apache.org/viewvc?rev=1199739&view=rev
Log:
CAMEL-2740: Fixed memory leak when doing request/reply over JMS with a fixed
replyTo queue. Due correlationIDs in JMSMessage Selector not in sync with
current active correlation ids.
Added:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationListener.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/MessageSelectorCreator.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyHandler.java
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java?rev=1199739&r1=1199738&r2=1199739&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
Wed Nov 9 13:22:55 2011
@@ -106,14 +106,14 @@ public class DefaultTimeoutMap<K, V> ext
}
}
- public V remove(K id) {
+ public V remove(K key) {
TimeoutMapEntry<K, V> entry;
if (useLock) {
lock.lock();
}
try {
- entry = map.remove(id);
+ entry = map.remove(key);
} finally {
if (useLock) {
lock.unlock();
Added:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationListener.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationListener.java?rev=1199739&view=auto
==============================================================================
---
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationListener.java
(added)
+++
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationListener.java
Wed Nov 9 13:22:55 2011
@@ -0,0 +1,44 @@
+/**
+ * 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.jms.reply;
+
+/**
+ * Listener for events when correlation id's changes.
+ */
+public interface CorrelationListener {
+
+ /**
+ * Callback when a new correlation id is added
+ *
+ * @param key the correlation id
+ */
+ void onPut(String key);
+
+ /**
+ * Callback when a correlation id is removed
+ *
+ * @param key the correlation id
+ */
+ void onRemove(String key);
+
+ /**
+ * Callback when a correlation id is evicted due timeout
+ *
+ * @param key the correlation id
+ */
+ void onEviction(String key);
+}
Modified:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java?rev=1199739&r1=1199738&r2=1199739&view=diff
==============================================================================
---
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
(original)
+++
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/CorrelationTimeoutMap.java
Wed Nov 9 13:22:55 2011
@@ -29,11 +29,26 @@ import org.apache.camel.support.DefaultT
*/
public class CorrelationTimeoutMap extends DefaultTimeoutMap<String,
ReplyHandler> {
+ private CorrelationListener listener;
+
public CorrelationTimeoutMap(ScheduledExecutorService executor, long
requestMapPollTimeMillis) {
super(executor, requestMapPollTimeMillis);
}
+ public void setListener(CorrelationListener listener) {
+ // there is only one listener needed
+ this.listener = listener;
+ }
+
public boolean onEviction(String key, ReplyHandler value) {
+ try {
+ if (listener != null) {
+ listener.onEviction(key);
+ }
+ } catch (Throwable e) {
+ // ignore
+ }
+
// trigger timeout
value.onTimeout(key);
// return true to remove the element
@@ -43,6 +58,14 @@ public class CorrelationTimeoutMap exten
@Override
public void put(String key, ReplyHandler value, long timeoutMillis) {
+ try {
+ if (listener != null) {
+ listener.onPut(key);
+ }
+ } catch (Throwable e) {
+ // ignore
+ }
+
if (timeoutMillis <= 0) {
// no timeout (must use Integer.MAX_VALUE)
super.put(key, value, Integer.MAX_VALUE);
@@ -53,9 +76,17 @@ public class CorrelationTimeoutMap exten
}
@Override
- public ReplyHandler remove(String id) {
- ReplyHandler answer = super.remove(id);
- log.trace("Removed correlationID: {} -> {}", id, answer != null);
+ public ReplyHandler remove(String key) {
+ try {
+ if (listener != null) {
+ listener.onRemove(key);
+ }
+ } catch (Throwable e) {
+ // ignore
+ }
+
+ ReplyHandler answer = super.remove(key);
+ log.trace("Removed correlationID: {} -> {}", key, answer != null);
return answer;
}
Modified:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/MessageSelectorCreator.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/MessageSelectorCreator.java?rev=1199739&r1=1199738&r2=1199739&view=diff
==============================================================================
---
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/MessageSelectorCreator.java
(original)
+++
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/MessageSelectorCreator.java
Wed Nov 9 13:22:55 2011
@@ -16,8 +16,8 @@
*/
package org.apache.camel.component.jms.reply;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.LinkedHashSet;
+import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,26 +26,19 @@ import org.slf4j.LoggerFactory;
* A creator which can build the JMS message selector query string to use
* with a shared persistent reply-to queue, so we can select the correct
messages we expect as replies.
*/
-public class MessageSelectorCreator {
+public class MessageSelectorCreator implements CorrelationListener {
protected static final Logger LOG =
LoggerFactory.getLogger(MessageSelectorCreator.class);
- protected Map<String, String> correlationIds;
+ protected final CorrelationTimeoutMap timeoutMap;
+ protected final Set<String> correlationIds;
protected boolean dirty = true;
protected StringBuilder expression;
- public MessageSelectorCreator() {
- correlationIds = new HashMap<String, String>();
- }
-
- public synchronized void addCorrelationID(String id) {
- correlationIds.put(id, id);
- LOG.trace("Added correlationID: {}", id);
- dirty = true;
- }
-
- public synchronized void removeCorrelationID(String id) {
- boolean answer = correlationIds.remove(id) != null;
- LOG.trace("Removed correlationID: {} -> {}", id, answer);
- dirty = true;
+ public MessageSelectorCreator(CorrelationTimeoutMap timeoutMap) {
+ this.timeoutMap = timeoutMap;
+ this.timeoutMap.setListener(this);
+ // create local set of correlation ids, as its easier to keep track
+ // using the listener so we can flag the dirty flag upon changes
+ this.correlationIds = new LinkedHashSet<String>();
}
public synchronized String get() {
@@ -55,16 +48,16 @@ public class MessageSelectorCreator {
expression = new StringBuilder("JMSCorrelationID='");
- if (correlationIds.isEmpty()) {
+ if (correlationIds.size() == 0) {
// no id's so use a dummy to select nothing
expression.append("CamelDummyJmsMessageSelector'");
} else {
boolean first = true;
- for (Map.Entry<String, String> entry : correlationIds.entrySet()) {
+ for (String value : correlationIds) {
if (!first) {
expression.append(" OR JMSCorrelationID='");
}
- expression.append(entry.getValue()).append("'");
+ expression.append(value).append("'");
if (first) {
first = false;
}
@@ -75,4 +68,18 @@ public class MessageSelectorCreator {
return expression.toString();
}
+ public void onPut(String key) {
+ dirty = true;
+ correlationIds.add(key);
+ }
+
+ public void onRemove(String key) {
+ dirty = true;
+ correlationIds.remove(key);
+ }
+
+ public void onEviction(String key) {
+ dirty = true;
+ correlationIds.remove(key);
+ }
}
\ No newline at end of file
Modified:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyHandler.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyHandler.java?rev=1199739&r1=1199738&r2=1199739&view=diff
==============================================================================
---
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyHandler.java
(original)
+++
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyHandler.java
Wed Nov 9 13:22:55 2011
@@ -16,8 +16,6 @@
*/
package org.apache.camel.component.jms.reply;
-import javax.jms.Message;
-
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -36,27 +34,4 @@ public class PersistentQueueReplyHandler
this.dynamicMessageSelector = dynamicMessageSelector;
}
- @Override
- public void onReply(String correlationId, Message reply) {
- try {
- if (dynamicMessageSelector != null) {
- // remove correlation id from message selector
- dynamicMessageSelector.removeCorrelationID(correlationId);
- }
- } finally {
- super.onReply(correlationId, reply);
- }
- }
-
- @Override
- public void onTimeout(String correlationId) {
- try {
- if (dynamicMessageSelector != null) {
- // remove correlation id from message selector
- dynamicMessageSelector.removeCorrelationID(correlationId);
- }
- } finally {
- super.onTimeout(correlationId);
- }
- }
}
Modified:
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java?rev=1199739&r1=1199738&r2=1199739&view=diff
==============================================================================
---
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java
(original)
+++
camel/trunk/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/PersistentQueueReplyManager.java
Wed Nov 9 13:22:55 2011
@@ -46,10 +46,6 @@ public class PersistentQueueReplyManager
PersistentQueueReplyHandler handler = new
PersistentQueueReplyHandler(replyManager, exchange, callback,
originalCorrelationId, requestTimeout, dynamicMessageSelector);
correlation.put(correlationId, handler, requestTimeout);
- if (dynamicMessageSelector != null) {
- // also remember to keep the dynamic selector updated with the new
correlation id
- dynamicMessageSelector.addCorrelationID(correlationId);
- }
return correlationId;
}
@@ -63,14 +59,6 @@ public class PersistentQueueReplyManager
}
correlation.put(newCorrelationId, handler, requestTimeout);
-
- // no not arrived early
- if (dynamicMessageSelector != null) {
- // also remember to keep the dynamic selector updated with the new
correlation id
- // at first removing the old correlationID and then add the new
correlationID
- dynamicMessageSelector.removeCorrelationID(correlationId);
- dynamicMessageSelector.addCorrelationID(newCorrelationId);
- }
}
protected void handleReplyMessage(String correlationID, Message message) {
@@ -143,7 +131,7 @@ public class PersistentQueueReplyManager
log.debug("Using shared queue: " + endpoint.getReplyTo() + "
with fixed message selector [" + fixedMessageSelector + "] as reply listener: "
+ answer);
} else {
// use a dynamic message selector which will select the
message we want to receive as reply
- dynamicMessageSelector = new MessageSelectorCreator();
+ dynamicMessageSelector = new
MessageSelectorCreator(correlation);
answer = new
SharedPersistentQueueMessageListenerContainer(dynamicMessageSelector);
log.debug("Using shared queue: " + endpoint.getReplyTo() + "
with dynamic message selector as reply listener: " + answer);
}
