You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by cs...@apache.org on 2015/01/26 11:23:17 UTC
[1/5] cxf git commit: CXF-6199 Adding concurrentConsumers
Repository: cxf
Updated Branches:
refs/heads/3.0.x-fixes 3c874549a -> fbfc4b3cb
CXF-6199 Adding concurrentConsumers
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/5ae70f42
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/5ae70f42
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/5ae70f42
Branch: refs/heads/3.0.x-fixes
Commit: 5ae70f42be238e5c4947d2cefc93dad1773ba0c8
Parents: 3c87454
Author: Christian Schneider <ch...@die-schneider.net>
Authored: Fri Jan 16 16:47:36 2015 +0100
Committer: Christian Schneider <ch...@die-schneider.net>
Committed: Mon Jan 26 10:41:18 2015 +0100
----------------------------------------------------------------------
.../cxf/transport/jms/JMSConfiguration.java | 9 +++
.../cxf/transport/jms/JMSDestination.java | 17 +++--
.../cxf/transport/jms/uri/JMSEndpoint.java | 14 ++++
.../util/AbstractMessageListenerContainer.java | 22 -------
.../jms/util/MessageListenerContainer.java | 68 +-------------------
.../util/PollingMessageListenerContainer.java | 42 ++++++++----
.../cxf/transport/jms/uri/JMSEndpointTest.java | 3 +-
7 files changed, 67 insertions(+), 108 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/5ae70f42/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java
index b4ec3f0..5dc0e31 100644
--- a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java
+++ b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfiguration.java
@@ -54,6 +54,7 @@ public class JMSConfiguration {
private boolean sessionTransacted;
private boolean createSecurityContext = true;
+ private int concurrentConsumers = 1;
private int maxSuspendedContinuations = DEFAULT_VALUE;
private int reconnectPercentOfMax = 70;
@@ -320,6 +321,14 @@ public class JMSConfiguration {
public void setTransactionManager(Object transactionManager) {
}
+ public int getConcurrentConsumers() {
+ return concurrentConsumers;
+ }
+
+ public void setConcurrentConsumers(int concurrentConsumers) {
+ this.concurrentConsumers = concurrentConsumers;
+ }
+
public int getMaxSuspendedContinuations() {
return maxSuspendedContinuations;
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/5ae70f42/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java
index e4127d3..9f0fcbc 100644
--- a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java
+++ b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSDestination.java
@@ -48,10 +48,8 @@ import org.apache.cxf.service.model.EndpointInfo;
import org.apache.cxf.transport.AbstractMultiplexDestination;
import org.apache.cxf.transport.Conduit;
import org.apache.cxf.transport.jms.continuations.JMSContinuationProvider;
-import org.apache.cxf.transport.jms.util.AbstractMessageListenerContainer;
import org.apache.cxf.transport.jms.util.JMSListenerContainer;
import org.apache.cxf.transport.jms.util.JMSUtil;
-import org.apache.cxf.transport.jms.util.MessageListenerContainer;
import org.apache.cxf.transport.jms.util.PollingMessageListenerContainer;
import org.apache.cxf.transport.jms.util.ResourceCloser;
@@ -125,15 +123,18 @@ public class JMSDestination extends AbstractMultiplexDestination implements Mess
connection = JMSFactory.createConnection(jmsConfig);
connection.setExceptionListener(new ExceptionListener() {
public void onException(JMSException exception) {
- LOG.log(Level.WARNING, "Exception on JMS connection. Trying to reconnect", exception);
- restartConnection();
+ if (!shutdown) {
+ LOG.log(Level.WARNING, "Exception on JMS connection. Trying to reconnect", exception);
+ restartConnection();
+ }
}
});
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = jmsConfig.getTargetDestination(session);
- AbstractMessageListenerContainer container = jmsConfig.getTransactionManager() != null
- ? new PollingMessageListenerContainer(connection, destination, this)
- : new MessageListenerContainer(connection, destination, this);
+
+ PollingMessageListenerContainer container = new PollingMessageListenerContainer(connection,
+ destination, this);
+ container.setConcurrentConsumers(jmsConfig.getConcurrentConsumers());
container.setTransactionManager(jmsConfig.getTransactionManager());
container.setMessageSelector(jmsConfig.getMessageSelector());
container.setTransacted(jmsConfig.isSessionTransacted());
@@ -191,6 +192,8 @@ public class JMSDestination extends AbstractMultiplexDestination implements Mess
getLogger().log(Level.FINE, "JMSDestination shutdown()");
this.deactivate();
}
+
+
/**
* Convert JMS message received by ListenerThread to CXF message and inform incomingObserver that a
http://git-wip-us.apache.org/repos/asf/cxf/blob/5ae70f42/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/uri/JMSEndpoint.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/uri/JMSEndpoint.java b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/uri/JMSEndpoint.java
index 885c1f8..9d104ac 100644
--- a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/uri/JMSEndpoint.java
+++ b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/uri/JMSEndpoint.java
@@ -82,6 +82,7 @@ public class JMSEndpoint {
private String topicReplyToName;
private boolean useConduitIdSelector = true;
private String username;
+ private int concurrentConsumers = 1;
/**
* @param uri
@@ -342,6 +343,19 @@ public class JMSEndpoint {
public void setUsername(String username) {
this.username = username;
}
+
+ public int getConcurrentConsumers() {
+ return concurrentConsumers;
+ }
+
+ public void setConcurrentConsumers(int concurrentConsumers) {
+ this.concurrentConsumers = concurrentConsumers;
+ }
+
+ public void setConcurrentConsumers(String concurrentConsumers) {
+ this.concurrentConsumers = new Integer(concurrentConsumers);
+ }
+
public String getPassword() {
return password;
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/5ae70f42/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/AbstractMessageListenerContainer.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/AbstractMessageListenerContainer.java b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/AbstractMessageListenerContainer.java
index f33b3c8..f5affe4 100644
--- a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/AbstractMessageListenerContainer.java
+++ b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/AbstractMessageListenerContainer.java
@@ -24,7 +24,6 @@ import java.util.logging.Logger;
import javax.jms.Connection;
import javax.jms.Destination;
-import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.transaction.TransactionManager;
@@ -41,8 +40,6 @@ public abstract class AbstractMessageListenerContainer implements JMSListenerCon
protected int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
protected String messageSelector;
protected boolean running;
- protected MessageConsumer consumer;
- protected Session session;
protected Executor executor;
protected String durableSubscriptionName;
protected boolean pubSubNoLocal;
@@ -99,23 +96,4 @@ public abstract class AbstractMessageListenerContainer implements JMSListenerCon
this.transactionManager = transactionManager;
}
-
-
-
-
- /*
- protected TransactionManager getTransactionManager() {
- if (this.transactionManager == null) {
- try {
- InitialContext ctx = new InitialContext();
- this.transactionManager = (TransactionManager)ctx
- .lookup("javax.transaction.TransactionManager");
- } catch (NamingException e) {
- // Ignore
- }
- }
- return this.transactionManager;
- }
- */
-
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/5ae70f42/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/MessageListenerContainer.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/MessageListenerContainer.java b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/MessageListenerContainer.java
index 3a7b2e3..605c43f 100644
--- a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/MessageListenerContainer.java
+++ b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/MessageListenerContainer.java
@@ -18,18 +18,16 @@
*/
package org.apache.cxf.transport.jms.util;
-import java.util.concurrent.Executor;
import java.util.logging.Level;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
+import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.Topic;
-import javax.jms.XASession;
-import javax.transaction.TransactionManager;
/**
* Listen for messages on a queue or topic asynchronously by registering a
@@ -39,6 +37,8 @@ import javax.transaction.TransactionManager;
* This has to be handled outside.
*/
public class MessageListenerContainer extends AbstractMessageListenerContainer {
+ private MessageConsumer consumer;
+ private Session session;
public MessageListenerContainer(Connection connection, Destination destination,
MessageListener listenerHandler) {
this.connection = connection;
@@ -82,29 +82,6 @@ public class MessageListenerContainer extends AbstractMessageListenerContainer {
ResourceCloser.close(connection);
}
- static class DispachingListener implements MessageListener {
- private Executor executor;
- private MessageListener listenerHandler;
-
- public DispachingListener(Executor executor, MessageListener listenerHandler) {
- this.executor = executor;
- this.listenerHandler = listenerHandler;
- }
-
- @Override
- public void onMessage(final Message message) {
- executor.execute(new Runnable() {
-
- @Override
- public void run() {
- listenerHandler.onMessage(message);
- }
-
- });
- }
-
- }
-
static class LocalTransactionalMessageListener implements MessageListener {
private MessageListener listenerHandler;
private Session session;
@@ -137,43 +114,4 @@ public class MessageListenerContainer extends AbstractMessageListenerContainer {
}
- static class XATransactionalMessageListener implements MessageListener {
- private TransactionManager tm;
- private MessageListener listenerHandler;
- private XASession session;
-
- public XATransactionalMessageListener(TransactionManager tm, Session session, MessageListener listenerHandler) {
- if (tm == null) {
- throw new IllegalArgumentException("Must supply a transaction manager");
- }
- if (!(session instanceof XASession)) {
- throw new IllegalArgumentException("Must supply an XASession");
- }
- this.tm = tm;
- this.session = (XASession)session;
- this.listenerHandler = listenerHandler;
- }
-
- @Override
- public void onMessage(Message message) {
- try {
- tm.begin();
- tm.getTransaction().enlistResource(session.getXAResource());
- listenerHandler.onMessage(message);
- tm.commit();
- } catch (Throwable e) {
- safeRollback(e);
- }
- }
-
- private void safeRollback(Throwable t) {
- LOG.log(Level.WARNING, "Exception while processing jms message in cxf. Rolling back" , t);
- try {
- tm.rollback();
- } catch (Exception e) {
- LOG.log(Level.WARNING, "Rollback of JTA transaction failed", e);
- }
- }
-
- }
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/5ae70f42/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java
index 2a6ca01..9c2e29e 100644
--- a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java
+++ b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java
@@ -30,6 +30,7 @@ import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
+import javax.jms.Topic;
import org.apache.cxf.common.logging.LogUtils;
@@ -38,7 +39,7 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
private ExecutorService pollers;
- private int numListenerThreads = 1;
+ private int concurrentConsumers = 1;
public PollingMessageListenerContainer(Connection connection, Destination destination,
MessageListener listenerHandler) {
@@ -47,20 +48,27 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
this.listenerHandler = listenerHandler;
}
- class Poller implements Runnable {
+ private class Poller implements Runnable {
@Override
public void run() {
- ResourceCloser closer = new ResourceCloser();
while (running) {
+ MessageConsumer consumer = null;
+ Session session = null;
try {
if (transactionManager != null) {
transactionManager.begin();
}
- Session session = closer.register(connection.createSession(transacted, acknowledgeMode));
-
- MessageConsumer consumer = closer.register(session.createConsumer(destination,
- messageSelector));
+
+ session = connection.createSession(transacted, acknowledgeMode);
+ if (durableSubscriptionName != null && destination instanceof Topic) {
+ consumer = session.createDurableSubscriber((Topic)destination,
+ durableSubscriptionName,
+ messageSelector,
+ pubSubNoLocal);
+ } else {
+ consumer = session.createConsumer(destination, messageSelector);
+ }
Message message = consumer.receive(1000);
try {
if (message != null) {
@@ -68,7 +76,7 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
}
if (transactionManager != null) {
transactionManager.commit();
- } else {
+ } else if (session.getTransacted()) {
session.commit();
}
} catch (Exception e) {
@@ -77,7 +85,8 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
} catch (Exception e) {
LOG.log(Level.WARNING, "Unexpected exception", e);
} finally {
- closer.close();
+ ResourceCloser.close(consumer);
+ ResourceCloser.close(session);
}
}
@@ -91,7 +100,9 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
if (transactionManager != null) {
transactionManager.rollback();
} else {
- session.rollback();
+ if (session.getTransacted()) {
+ session.rollback();
+ }
}
} catch (Exception e1) {
LOG.log(Level.WARNING, "Rollback of Local transaction failed", e1);
@@ -104,14 +115,15 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
return;
}
running = true;
- pollers = Executors.newFixedThreadPool(numListenerThreads);
- for (int c = 0; c < numListenerThreads; c++) {
+ pollers = Executors.newFixedThreadPool(concurrentConsumers);
+ for (int c = 0; c < concurrentConsumers; c++) {
pollers.execute(new Poller());
}
}
@Override
public void stop() {
+ LOG.fine("Shuttting down " + this.getClass().getSimpleName());
if (!running) {
return;
}
@@ -122,12 +134,16 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
} catch (InterruptedException e) {
// Ignore
}
+ pollers.shutdownNow();
pollers = null;
}
@Override
public void shutdown() {
stop();
- ResourceCloser.close(connection);
+ }
+
+ public void setConcurrentConsumers(int concurrentConsumers) {
+ this.concurrentConsumers = concurrentConsumers;
}
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/5ae70f42/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/uri/JMSEndpointTest.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/uri/JMSEndpointTest.java b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/uri/JMSEndpointTest.java
index b7f5041..f52f28e 100644
--- a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/uri/JMSEndpointTest.java
+++ b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/uri/JMSEndpointTest.java
@@ -29,10 +29,11 @@ public class JMSEndpointTest extends Assert {
@Test
public void testBasicQueue() throws Exception {
- JMSEndpoint endpoint = new JMSEndpoint("jms:queue:Foo.Bar");
+ JMSEndpoint endpoint = new JMSEndpoint("jms:queue:Foo.Bar?concurrentConsumers=21");
assertEquals(JMSEndpoint.QUEUE, endpoint.getJmsVariant());
assertEquals("Foo.Bar", endpoint.getDestinationName());
assertEquals(JMSEndpoint.QUEUE, endpoint.getJmsVariant());
+ assertEquals(21, endpoint.getConcurrentConsumers());
}
@Test
[3/5] cxf git commit: CXF-6199 Separate XA and local transaction
handling
Posted by cs...@apache.org.
CXF-6199 Separate XA and local transaction handling
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/592cd43b
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/592cd43b
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/592cd43b
Branch: refs/heads/3.0.x-fixes
Commit: 592cd43b2bf0a1ae4483a1a72798e4ab238ffa9e
Parents: 15fa235
Author: Christian Schneider <ch...@die-schneider.net>
Authored: Tue Jan 20 00:07:57 2015 +0100
Committer: Christian Schneider <ch...@die-schneider.net>
Committed: Mon Jan 26 10:41:47 2015 +0100
----------------------------------------------------------------------
.../util/PollingMessageListenerContainer.java | 113 +++++++++++++------
.../transport/jms/util/MessageListenerTest.java | 2 +-
2 files changed, 79 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/592cd43b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java
index 9c2e29e..8c68338 100644
--- a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java
+++ b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/util/PollingMessageListenerContainer.java
@@ -26,6 +26,7 @@ import java.util.logging.Logger;
import javax.jms.Connection;
import javax.jms.Destination;
+import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
@@ -56,34 +57,25 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
MessageConsumer consumer = null;
Session session = null;
try {
- if (transactionManager != null) {
- transactionManager.begin();
- }
-
+ // Create session early to optimize performance
session = connection.createSession(transacted, acknowledgeMode);
- if (durableSubscriptionName != null && destination instanceof Topic) {
- consumer = session.createDurableSubscriber((Topic)destination,
- durableSubscriptionName,
- messageSelector,
- pubSubNoLocal);
- } else {
- consumer = session.createConsumer(destination, messageSelector);
- }
- Message message = consumer.receive(1000);
- try {
- if (message != null) {
- listenerHandler.onMessage(message);
- }
- if (transactionManager != null) {
- transactionManager.commit();
- } else if (session.getTransacted()) {
- session.commit();
+ consumer = createConsumer(session);
+ while (running) {
+ Message message = consumer.receive(1000);
+ try {
+ if (message != null) {
+ listenerHandler.onMessage(message);
+ }
+ if (session.getTransacted()) {
+ session.commit();
+ }
+ } catch (Exception e) {
+ LOG.log(Level.WARNING, "Exception while processing jms message in cxf. Rolling back", e);
+ safeRollBack(session, e);
}
- } catch (Exception e) {
- safeRollBack(session, e);
}
} catch (Exception e) {
- LOG.log(Level.WARNING, "Unexpected exception", e);
+ LOG.log(Level.WARNING, "Unexpected exception. Restarting session and consumer", e);
} finally {
ResourceCloser.close(consumer);
ResourceCloser.close(session);
@@ -91,21 +83,71 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
}
}
+
+ private void safeRollBack(Session session, Exception e) {
+ try {
+ if (session.getTransacted()) {
+ session.rollback();
+ }
+ } catch (Exception e1) {
+ LOG.log(Level.WARNING, "Rollback of Local transaction failed", e1);
+ }
+ }
}
+
+ private class XAPoller implements Runnable {
- private void safeRollBack(Session session, Exception e) {
- LOG.log(Level.WARNING, "Exception while processing jms message in cxf. Rolling back", e);
- try {
- if (transactionManager != null) {
- transactionManager.rollback();
- } else {
- if (session.getTransacted()) {
- session.rollback();
+ @Override
+ public void run() {
+ while (running) {
+ MessageConsumer consumer = null;
+ Session session = null;
+ try {
+ transactionManager.begin();
+ /*
+ * Create session inside transaction to give it the
+ * chance to enlist itself as a resource
+ */
+ session = connection.createSession(transacted, acknowledgeMode);
+ consumer = createConsumer(session);
+ Message message = consumer.receive(1000);
+ try {
+ if (message != null) {
+ listenerHandler.onMessage(message);
+ }
+ transactionManager.commit();
+ } catch (Exception e) {
+ LOG.log(Level.WARNING, "Exception while processing jms message in cxf. Rolling back", e);
+ safeRollBack(session);
+ } finally {
+ ResourceCloser.close(consumer);
+ ResourceCloser.close(session);
+ }
+ } catch (Exception e) {
+ LOG.log(Level.WARNING, "Unexpected exception. Restarting session and consumer", e);
}
+
+ }
+
+ }
+
+ private void safeRollBack(Session session) {
+ try {
+ transactionManager.rollback();
+ } catch (Exception e) {
+ LOG.log(Level.WARNING, "Rollback of XA transaction failed", e);
}
- } catch (Exception e1) {
- LOG.log(Level.WARNING, "Rollback of Local transaction failed", e1);
+ }
+
+ }
+
+ private MessageConsumer createConsumer(Session session) throws JMSException {
+ if (durableSubscriptionName != null && destination instanceof Topic) {
+ return session.createDurableSubscriber((Topic)destination, durableSubscriptionName,
+ messageSelector, pubSubNoLocal);
+ } else {
+ return session.createConsumer(destination, messageSelector);
}
}
@@ -117,7 +159,8 @@ public class PollingMessageListenerContainer extends AbstractMessageListenerCont
running = true;
pollers = Executors.newFixedThreadPool(concurrentConsumers);
for (int c = 0; c < concurrentConsumers; c++) {
- pollers.execute(new Poller());
+ Runnable poller = (transactionManager != null) ? new XAPoller() : new Poller();
+ pollers.execute(poller);
}
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/592cd43b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/util/MessageListenerTest.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/util/MessageListenerTest.java b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/util/MessageListenerTest.java
index 2484277..8b635e8 100644
--- a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/util/MessageListenerTest.java
+++ b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/util/MessageListenerTest.java
@@ -173,7 +173,7 @@ public class MessageListenerTest {
// + ", expecting: " + expectedNum);
Thread.sleep(100);
} while ((System.currentTimeMillis() - startTime < timeout) && expectedNum != actualNum);
- Assert.assertEquals(message + " -> number of messages", expectedNum, actualNum);
+ Assert.assertEquals(message + " -> number of messages on queue", expectedNum, actualNum);
}
private void sendMessage(Connection connection, Destination dest, String content) throws JMSException,
[4/5] cxf git commit: CXF-6187 Adding jndiURL
Posted by cs...@apache.org.
CXF-6187 Adding jndiURL
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/1d9eda92
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/1d9eda92
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/1d9eda92
Branch: refs/heads/3.0.x-fixes
Commit: 1d9eda92abf428ba55fffdb6c13b19df2c57d82e
Parents: 592cd43
Author: Christian Schneider <ch...@die-schneider.net>
Authored: Fri Jan 23 18:12:47 2015 +0100
Committer: Christian Schneider <ch...@die-schneider.net>
Committed: Mon Jan 26 10:42:23 2015 +0100
----------------------------------------------------------------------
.../src/main/java/demo/client/ClientJMS.java | 3 +-
.../src/main/java/demo/server/ServerJMS.java | 43 ++++++++++++--------
2 files changed, 27 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/1d9eda92/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/client/ClientJMS.java
----------------------------------------------------------------------
diff --git a/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/client/ClientJMS.java b/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/client/ClientJMS.java
index 43ecb89..86b16b9 100644
--- a/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/client/ClientJMS.java
+++ b/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/client/ClientJMS.java
@@ -30,7 +30,8 @@ import demo.service.HelloWorld;
public final class ClientJMS {
private static final String JMS_ENDPOINT_URI = "jms:queue:test.cxf.jmstransport.queue?timeToLive=1000"
+ "&jndiConnectionFactoryName=ConnectionFactory" + "&jndiInitialContextFactory"
- + "=org.apache.activemq.jndi.ActiveMQInitialContextFactory";
+ + "=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
+ + "&jndiURL=tcp://localhost:61616";
private static final QName SERVICE_QNAME =
new QName("http://impl.service.demo/", "HelloWorldImplService");
http://git-wip-us.apache.org/repos/asf/cxf/blob/1d9eda92/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/server/ServerJMS.java
----------------------------------------------------------------------
diff --git a/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/server/ServerJMS.java b/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/server/ServerJMS.java
index 6d03861..8da3b9f 100644
--- a/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/server/ServerJMS.java
+++ b/distribution/src/main/release/samples/java_first_jms/src/main/java/demo/server/ServerJMS.java
@@ -19,6 +19,7 @@
package demo.server;
+import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import javax.xml.ws.Endpoint;
@@ -34,7 +35,8 @@ public final class ServerJMS {
private static final String JMS_ENDPOINT_URI = "jms:queue:test.cxf.jmstransport.queue?timeToLive=1000"
+ "&jndiConnectionFactoryName=ConnectionFactory"
+ "&jndiInitialContextFactory"
- + "=org.apache.activemq.jndi.ActiveMQInitialContextFactory";
+ + "=org.apache.activemq.jndi.ActiveMQInitialContextFactory"
+ + "&jndiURL=tcp://localhost:61616";
private ServerJMS() {
//
@@ -55,23 +57,7 @@ public final class ServerJMS {
}
if (launchAmqBroker) {
- /*
- * The following make it easier to run this against something other than ActiveMQ. You will have
- * to get a JMS broker onto the right port of localhost.
- */
- Class<?> brokerClass = ServerJMS.class.getClassLoader()
- .loadClass("org.apache.activemq.broker.BrokerService");
- if (brokerClass == null) {
- System.err.println("ActiveMQ is not in the classpath, cannot launch broker.");
- return;
- }
- Object broker = brokerClass.newInstance();
- Method addConnectorMethod = brokerClass.getMethod("addConnector", String.class);
- addConnectorMethod.invoke(broker, "tcp://localhost:61616");
- Method setDataDirectory = brokerClass.getMethod("setDataDirectory", String.class);
- setDataDirectory.invoke(broker, "target/activemq-data");
- Method startMethod = brokerClass.getMethod("start");
- startMethod.invoke(broker);
+ launchAMQBroker();
}
if (jaxws) {
@@ -86,6 +72,27 @@ public final class ServerJMS {
System.exit(0);
}
+ private static void launchAMQBroker() throws ClassNotFoundException, InstantiationException,
+ IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+ /*
+ * The following make it easier to run this against something other than ActiveMQ. You will have
+ * to get a JMS broker onto the right port of localhost.
+ */
+ Class<?> brokerClass = ServerJMS.class.getClassLoader()
+ .loadClass("org.apache.activemq.broker.BrokerService");
+ if (brokerClass == null) {
+ System.err.println("ActiveMQ is not in the classpath, cannot launch broker.");
+ return;
+ }
+ Object broker = brokerClass.newInstance();
+ Method addConnectorMethod = brokerClass.getMethod("addConnector", String.class);
+ addConnectorMethod.invoke(broker, "tcp://localhost:61616");
+ Method setDataDirectory = brokerClass.getMethod("setDataDirectory", String.class);
+ setDataDirectory.invoke(broker, "target/activemq-data");
+ Method startMethod = brokerClass.getMethod("start");
+ startMethod.invoke(broker);
+ }
+
private static void launchCxfApi() {
Object implementor = new HelloWorldImpl();
JaxWsServerFactoryBean svrFactory = new JaxWsServerFactoryBean();
[2/5] cxf git commit: CXF-6199 Make sure concurrentConsumers can be
configured on JMSEndpoint
Posted by cs...@apache.org.
CXF-6199 Make sure concurrentConsumers can be configured on JMSEndpoint
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/15fa235a
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/15fa235a
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/15fa235a
Branch: refs/heads/3.0.x-fixes
Commit: 15fa235ac9239571ed10c9ab83c4146e73b72942
Parents: 5ae70f4
Author: Christian Schneider <ch...@die-schneider.net>
Authored: Tue Jan 20 00:02:19 2015 +0100
Committer: Christian Schneider <ch...@die-schneider.net>
Committed: Mon Jan 26 10:41:32 2015 +0100
----------------------------------------------------------------------
.../java/org/apache/cxf/transport/jms/JMSConfigFactory.java | 1 +
.../org/apache/cxf/transport/jms/JMSConfigFactoryTest.java | 8 ++++++++
2 files changed, 9 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/15fa235a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfigFactory.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfigFactory.java b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfigFactory.java
index e5ec995..464fc7a 100644
--- a/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfigFactory.java
+++ b/rt/transports/jms/src/main/java/org/apache/cxf/transport/jms/JMSConfigFactory.java
@@ -85,6 +85,7 @@ public final class JMSConfigFactory {
jmsConfig.setConduitSelectorPrefix(endpoint.getConduitIdSelectorPrefix());
jmsConfig.setUserName(endpoint.getUsername());
jmsConfig.setPassword(endpoint.getPassword());
+ jmsConfig.setConcurrentConsumers(endpoint.getConcurrentConsumers());
TransactionManager tm = getTransactionManager(bus, endpoint);
jmsConfig.setTransactionManager(tm);
http://git-wip-us.apache.org/repos/asf/cxf/blob/15fa235a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSConfigFactoryTest.java
----------------------------------------------------------------------
diff --git a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSConfigFactoryTest.java b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSConfigFactoryTest.java
index 1198b9e..9008732 100644
--- a/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSConfigFactoryTest.java
+++ b/rt/transports/jms/src/test/java/org/apache/cxf/transport/jms/JMSConfigFactoryTest.java
@@ -79,4 +79,12 @@ public class JMSConfigFactoryTest extends AbstractJMSTester {
Assert.assertEquals("java:/comp/TransactionManager", endpoint.getJndiTransactionManagerName());
// TODO Check JNDI lookup
}
+
+ @Test
+ public void testConcurrentConsumers() {
+ JMSEndpoint endpoint = new JMSEndpoint("jms:queue:Foo.Bar?concurrentConsumers=4");
+ JMSConfiguration jmsConfig = JMSConfigFactory.createFromEndpoint(bus, endpoint);
+ Assert.assertEquals(4, jmsConfig.getConcurrentConsumers());
+
+ }
}
[5/5] cxf git commit: CXF-6206 Adding an interceptor to create 401
response
Posted by cs...@apache.org.
CXF-6206 Adding an interceptor to create 401 response
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/fbfc4b3c
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/fbfc4b3c
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/fbfc4b3c
Branch: refs/heads/3.0.x-fixes
Commit: fbfc4b3cb4d6b3e1e2d5b0cb03d7e1b88947115d
Parents: 1d9eda9
Author: Christian Schneider <ch...@die-schneider.net>
Authored: Wed Jan 21 09:59:38 2015 +0100
Committer: Christian Schneider <ch...@die-schneider.net>
Committed: Mon Jan 26 10:43:02 2015 +0100
----------------------------------------------------------------------
.../http/HttpAuthenticationFaultHandler.java | 75 ++++++++++++++++++++
1 file changed, 75 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/fbfc4b3c/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpAuthenticationFaultHandler.java
----------------------------------------------------------------------
diff --git a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpAuthenticationFaultHandler.java b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpAuthenticationFaultHandler.java
new file mode 100644
index 0000000..3167669
--- /dev/null
+++ b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpAuthenticationFaultHandler.java
@@ -0,0 +1,75 @@
+/**
+ * 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.cxf.transport.http;
+
+import java.io.IOException;
+
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.cxf.interceptor.Fault;
+import org.apache.cxf.interceptor.security.AuthenticationException;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.AbstractPhaseInterceptor;
+import org.apache.cxf.phase.Phase;
+
+/**
+ * Translates an AuthenticationException into a 401 response
+ */
+public class HttpAuthenticationFaultHandler extends AbstractPhaseInterceptor<Message> {
+ String authenticationType;
+ String realm;
+
+ public HttpAuthenticationFaultHandler() {
+ super(Phase.UNMARSHAL);
+ this.authenticationType = "Basic";
+ this.realm = "CXF service";
+ }
+
+ @Override
+ public void handleMessage(Message message) throws Fault {
+ // Nothing
+ }
+
+ @Override
+ public void handleFault(Message message) {
+ Exception ex = message.getContent(Exception.class);
+ if (ex instanceof AuthenticationException) {
+ HttpServletResponse resp = (HttpServletResponse)message.getExchange()
+ .getInMessage().get(AbstractHTTPDestination.HTTP_RESPONSE);
+ resp.setStatus(HttpServletResponse.SC_UNAUTHORIZED);
+ resp.setHeader("WWW-Authenticate", authenticationType + " realm=\"" + realm + "\"");
+ resp.setContentType("text/plain");
+ try {
+ resp.getWriter().write(ex.getMessage());
+ resp.getWriter().flush();
+ message.getInterceptorChain().abort();
+ } catch (IOException e) {
+ // TODO
+ }
+ }
+ }
+
+ public void setAuthenticationType(String authenticationType) {
+ this.authenticationType = authenticationType;
+ }
+
+ public void setRealm(String realm) {
+ this.realm = realm;
+ }
+}