Author: dkulp
Date: Thu Sep 3 17:20:40 2009
New Revision: 811045
URL: http://svn.apache.org/viewvc?rev=811045&view=rev
Log:
Some thread safety issues in JMSConduit
Added:
cxf/branches/2.2.x-fixes/.gitignore
Modified:
cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConduit.java
cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSTransportFactory.java
cxf/branches/2.2.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java
Added: cxf/branches/2.2.x-fixes/.gitignore
URL:
http://svn.apache.org/viewvc/cxf/branches/2.2.x-fixes/.gitignore?rev=811045&view=auto
==============================================================================
--- cxf/branches/2.2.x-fixes/.gitignore (added)
+++ cxf/branches/2.2.x-fixes/.gitignore Thu Sep 3 17:20:40 2009
@@ -0,0 +1,8 @@
+.checkstyle
+.classpath
+.pmd
+.project
+.ruleset
+.settings/
+target/
+velocity.log
Modified:
cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConduit.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConduit.java?rev=811045&r1=811044&r2=811045&view=diff
==============================================================================
---
cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConduit.java
(original)
+++
cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConduit.java
Thu Sep 3 17:20:40 2009
@@ -53,6 +53,7 @@
import org.springframework.jms.connection.SingleConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.JmsUtils;
@@ -76,9 +77,14 @@
private String conduitId;
private AtomicLong messageCount;
private JMSBusLifeCycleListener listener;
+ private Bus bus;
- public JMSConduit(EndpointInfo endpointInfo, EndpointReferenceType target,
JMSConfiguration jmsConfig) {
+ public JMSConduit(EndpointInfo endpointInfo,
+ EndpointReferenceType target,
+ JMSConfiguration jmsConfig,
+ Bus b) {
super(target);
+ bus = b;
this.jmsConfig = jmsConfig;
this.endpointInfo = endpointInfo;
correlationMap = new ConcurrentHashMap<String, Exchange>();
@@ -100,6 +106,29 @@
JMSOutputStream out = new JMSOutputStream(this, message.getExchange(),
isTextPayload);
message.setContent(OutputStream.class, out);
}
+
+ private synchronized AbstractMessageListenerContainer getJMSListener() {
+ if (jmsListener == null) {
+ jmsListener = JMSFactory.createJmsListener(jmsConfig, this,
+
jmsConfig.getReplyDestination(),
+ conduitId,
+ false);
+ addBusListener();
+ }
+ return jmsListener;
+ }
+ private synchronized AbstractMessageListenerContainer getAllListener() {
+ if (allListener == null) {
+ allListener = JMSFactory.createJmsListener(jmsConfig,
+ this,
+
jmsConfig.getReplyDestination(),
+ null,
+ true);
+ addBusListener();
+ }
+
+ return allListener;
+ }
/**
* Send the JMS Request out and if not oneWay receive the response
@@ -130,27 +159,12 @@
final JmsTemplate jmsTemplate =
JMSFactory.createJmsTemplate(jmsConfig, headers);
String userCID = headers != null ? headers.getJMSCorrelationID() :
null;
- DefaultMessageListenerContainer jmsList = jmsListener;
+ AbstractMessageListenerContainer jmsList = jmsListener;
if (!exchange.isOneWay()) {
if (userCID == null || !jmsConfig.isUseConduitIdSelector()) {
- if (jmsListener == null) {
- jmsListener = JMSFactory.createJmsListener(jmsConfig,
this,
-
jmsConfig.getReplyDestination(),
- conduitId,
- false);
- addBusListener(exchange.get(Bus.class));
- }
- jmsList = jmsListener;
+ jmsList = getJMSListener();
} else {
- if (allListener == null) {
- allListener = JMSFactory.createJmsListener(jmsConfig,
- this,
-
jmsConfig.getReplyDestination(),
- null,
- true);
- addBusListener(exchange.get(Bus.class));
- }
- jmsList = allListener;
+ jmsList = getAllListener();
}
}
@@ -245,7 +259,7 @@
}
}
}
- private synchronized void addBusListener(Bus bus) {
+ private synchronized void addBusListener() {
if (listener == null && bus != null) {
BusLifeCycleManager blcm =
bus.getExtension(BusLifeCycleManager.class);
if (blcm != null) {
Modified:
cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSTransportFactory.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSTransportFactory.java?rev=811045&r1=811044&r2=811045&view=diff
==============================================================================
---
cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSTransportFactory.java
(original)
+++
cxf/branches/2.2.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSTransportFactory.java
Thu Sep 3 17:20:40 2009
@@ -63,7 +63,7 @@
public Conduit getConduit(EndpointInfo endpointInfo, EndpointReferenceType
target) throws IOException {
JMSOldConfigHolder old = new JMSOldConfigHolder();
JMSConfiguration jmsConf =
old.createJMSConfigurationFromEndpointInfo(bus, endpointInfo, true);
- return new JMSConduit(endpointInfo, target, jmsConf);
+ return new JMSConduit(endpointInfo, target, jmsConf, bus);
}
/**
Modified:
cxf/branches/2.2.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.2.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java?rev=811045&r1=811044&r2=811045&view=diff
==============================================================================
---
cxf/branches/2.2.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java
(original)
+++
cxf/branches/2.2.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/AbstractJMSTester.java
Thu Sep 3 17:20:40 2009
@@ -117,7 +117,7 @@
JMSConfiguration jmsConfig = new JMSOldConfigHolder()
.createJMSConfigurationFromEndpointInfo(bus, endpointInfo, true);
- JMSConduit jmsConduit = new JMSConduit(endpointInfo, target,
jmsConfig);
+ JMSConduit jmsConduit = new JMSConduit(endpointInfo, target,
jmsConfig, bus);
if (send) {
// setMessageObserver
observer = new MessageObserver() {
