Author: ningjiang
Date: Thu Jun 2 11:06:17 2011
New Revision: 1130495
URL: http://svn.apache.org/viewvc?rev=1130495&view=rev
Log:
Merged revisions 1130459,1130462-1130463 via svnmerge from
https://svn.apache.org/repos/asf/cxf/trunk
........
r1130459 | ningjiang | 2011-06-02 18:04:11 +0800 (Thu, 02 Jun 2011) | 2 lines
CXF-3567
Do not use timer for JMSContinuation timeout task
........
r1130462 | ningjiang | 2011-06-02 18:12:54 +0800 (Thu, 02 Jun 2011) | 2 lines
CXF-3567
Fixed the unit test error
........
r1130463 | ningjiang | 2011-06-02 18:13:50 +0800 (Thu, 02 Jun 2011) | 1 line
CXF-3567 Added the Messages.properties file
........
Added:
cxf/branches/2.3.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/Messages.properties
- copied unchanged from r1130463,
cxf/trunk/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/Messages.properties
Modified:
cxf/branches/2.3.x-fixes/ (props changed)
cxf/branches/2.3.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
cxf/branches/2.3.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java
Propchange: cxf/branches/2.3.x-fixes/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Thu Jun 2 11:06:17 2011
@@ -1 +1 @@
-/cxf/trunk:1128092,1128202-1128206,1130068
+/cxf/trunk:1128092,1128202-1128206,1130068,1130459-1130463
Propchange: cxf/branches/2.3.x-fixes/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Modified:
cxf/branches/2.3.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.3.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java?rev=1130495&r1=1130494&r2=1130495&view=diff
==============================================================================
---
cxf/branches/2.3.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
(original)
+++
cxf/branches/2.3.x-fixes/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/continuations/JMSContinuation.java
Thu Jun 2 11:06:17 2011
@@ -20,20 +20,22 @@
package org.apache.cxf.transport.jms.continuations;
import java.util.Collection;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.logging.Logger;
import org.apache.cxf.Bus;
import org.apache.cxf.BusFactory;
+import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.continuations.Continuation;
import org.apache.cxf.message.Message;
import org.apache.cxf.transport.MessageObserver;
import org.apache.cxf.transport.jms.JMSConfiguration;
+import org.apache.cxf.workqueue.WorkQueue;
+import org.apache.cxf.workqueue.WorkQueueManager;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
public class JMSContinuation implements Continuation {
-
+ private static final Logger LOG =
LogUtils.getL7dLogger(JMSContinuation.class);
private Bus bus;
private Message inMessage;
private MessageObserver incomingObserver;
@@ -46,7 +48,8 @@ public class JMSContinuation implements
private volatile boolean isNew = true;
private volatile boolean isPending;
private volatile boolean isResumed;
- private Timer timer;
+ private volatile boolean isCanceled;
+ private WorkQueue workQueue;
public JMSContinuation(Bus b, Message m, MessageObserver observer,
Collection<JMSContinuation> cList,
@@ -58,7 +61,16 @@ public class JMSContinuation implements
continuations = cList;
this.jmsListener = jmsListener;
this.jmsConfig = jmsConfig;
- }
+ WorkQueueManager manager = bus.getExtension(WorkQueueManager.class);
+ if (manager != null) {
+ workQueue = manager.getNamedWorkQueue("jms-continuation");
+ if (workQueue == null) {
+ workQueue = manager.getAutomaticWorkQueue();
+ }
+ } else {
+ LOG.warning("ERROR_GETTING_WORK_QUEUE");
+ }
+ }
public Object getObject() {
return userObject;
@@ -128,12 +140,10 @@ public class JMSContinuation implements
}
protected void createTimerTask(long timeout) {
- timer = new Timer();
- timer.schedule(new TimerTask() {
+ workQueue.schedule(new Runnable() {
public void run() {
synchronized (JMSContinuation.this) {
- if (isPending) {
- cancelTimerTask();
+ if (isPending && !isCanceled) {
doResume();
}
}
@@ -141,11 +151,8 @@ public class JMSContinuation implements
}, timeout);
}
- protected void cancelTimerTask() {
- if (timer != null) {
- timer.cancel();
- timer = null;
- }
+ protected synchronized void cancelTimerTask() {
+ isCanceled = true;
}
protected void updateContinuations(boolean remove) {
Modified:
cxf/branches/2.3.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java
URL:
http://svn.apache.org/viewvc/cxf/branches/2.3.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java?rev=1130495&r1=1130494&r2=1130495&view=diff
==============================================================================
---
cxf/branches/2.3.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java
(original)
+++
cxf/branches/2.3.x-fixes/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/continuations/JMSContinuationProviderTest.java
Thu Jun 2 11:06:17 2011
@@ -19,16 +19,26 @@
package org.apache.cxf.transport.jms.continuations;
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
import org.apache.cxf.continuations.Continuation;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.ExchangeImpl;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageImpl;
+
+import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Test;
public class JMSContinuationProviderTest extends Assert {
+ private static Bus bus = BusFactory.getDefaultBus();
+
+ @AfterClass
+ public static void cleanUpBus() {
+ BusFactory.setDefaultBus(null);
+ }
@Test
public void testNoContinuationForOneWay() {
@@ -46,7 +56,7 @@ public class JMSContinuationProviderTest
Message m = new MessageImpl();
m.setExchange(new ExchangeImpl());
JMSContinuationProvider provider =
- new JMSContinuationProvider(null, m, null, null, null, null);
+ new JMSContinuationProvider(bus, m, null, null, null, null);
Continuation cw = provider.getContinuation();
assertTrue(cw.isNew());
assertSame(cw, m.get(JMSContinuation.class));
@@ -56,7 +66,7 @@ public class JMSContinuationProviderTest
public void testGetExistingContinuation() {
Message m = new MessageImpl();
m.setExchange(new ExchangeImpl());
- JMSContinuation cw = new JMSContinuation(null, m, null, null, null,
null);
+ JMSContinuation cw = new JMSContinuation(bus, m, null, null, null,
null);
m.put(JMSContinuation.class, cw);
JMSContinuationProvider provider = new JMSContinuationProvider(null,
m, null, null, null, null);
assertSame(cw, provider.getContinuation());
