You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2015/10/26 19:08:20 UTC
[1/2] activemq-artemis git commit: This closes #218
Repository: activemq-artemis
Updated Branches:
refs/heads/master 992c34bc6 -> 7f60ff20a
This closes #218
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/7f60ff20
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/7f60ff20
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/7f60ff20
Branch: refs/heads/master
Commit: 7f60ff20a511094935e6ca181f652867aaa34657
Parents: 992c34b e864b79
Author: Clebert Suconic <cl...@apache.org>
Authored: Mon Oct 26 14:08:11 2015 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Mon Oct 26 14:08:11 2015 -0400
----------------------------------------------------------------------
.../ra/inflow/ActiveMQMessageHandler.java | 81 +++++++++++++-------
1 file changed, 53 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
[2/2] activemq-artemis git commit: ARTEMIS-280 - Resource Adapter
hangs on shutdown
Posted by cl...@apache.org.
ARTEMIS-280 - Resource Adapter hangs on shutdown
https://issues.apache.org/jira/browse/ARTEMIS-280
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/e864b79a
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/e864b79a
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/e864b79a
Branch: refs/heads/master
Commit: e864b79a6d0347589a3dcb077804eb956aa05e17
Parents: 992c34b
Author: Andy Taylor <an...@gmail.com>
Authored: Mon Oct 26 11:18:52 2015 +0000
Committer: Clebert Suconic <cl...@apache.org>
Committed: Mon Oct 26 14:08:11 2015 -0400
----------------------------------------------------------------------
.../ra/inflow/ActiveMQMessageHandler.java | 81 +++++++++++++-------
1 file changed, 53 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e864b79a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java
----------------------------------------------------------------------
diff --git a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java b/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java
index 149eea2..f6a8535 100644
--- a/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java
+++ b/artemis-ra/src/main/java/org/apache/activemq/artemis/ra/inflow/ActiveMQMessageHandler.java
@@ -32,6 +32,8 @@ import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientSession.QueueQuery;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
+import org.apache.activemq.artemis.api.core.client.FailoverEventListener;
+import org.apache.activemq.artemis.api.core.client.FailoverEventType;
import org.apache.activemq.artemis.api.core.client.MessageHandler;
import org.apache.activemq.artemis.core.client.impl.ClientConsumerInternal;
import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryInternal;
@@ -48,7 +50,7 @@ import org.apache.activemq.artemis.utils.VersionLoader;
/**
* The message handler
*/
-public class ActiveMQMessageHandler implements MessageHandler {
+public class ActiveMQMessageHandler implements MessageHandler, FailoverEventListener {
/**
* Trace enabled
@@ -80,6 +82,8 @@ public class ActiveMQMessageHandler implements MessageHandler {
private ClientSessionFactory cf;
+ private volatile boolean connected;
+
public ActiveMQMessageHandler(final ActiveMQActivation activation,
final TransactionManager tm,
final ClientSessionInternal session,
@@ -187,6 +191,8 @@ public class ActiveMQMessageHandler implements MessageHandler {
endpoint = endpointFactory.createEndpoint(null);
useXA = false;
}
+ connected = true;
+ session.addFailoverListener(this);
consumer.setMessageHandler(this);
}
@@ -224,39 +230,53 @@ public class ActiveMQMessageHandler implements MessageHandler {
ActiveMQRALogger.LOGGER.debug("Error releasing endpoint " + endpoint, t);
}
- try {
- consumer.close();
- if (activation.getTopicTemporaryQueue() != null) {
- // We need to delete temporary topics when the activation is stopped or messages will build up on the server
- SimpleString tmpQueue = activation.getTopicTemporaryQueue();
- QueueQuery subResponse = session.queueQuery(tmpQueue);
- if (subResponse.getConsumerCount() == 0) {
- // This is optional really, since we now use temporaryQueues, we could simply ignore this
- // and the server temporary queue would remove this as soon as the queue was removed
- session.deleteQueue(tmpQueue);
+ //only do this if we haven't been disconnected at some point whilst failing over
+ if (connected) {
+ try {
+ consumer.close();
+ if (activation.getTopicTemporaryQueue() != null) {
+ // We need to delete temporary topics when the activation is stopped or messages will build up on the server
+ SimpleString tmpQueue = activation.getTopicTemporaryQueue();
+ QueueQuery subResponse = session.queueQuery(tmpQueue);
+ if (subResponse.getConsumerCount() == 0) {
+ // This is optional really, since we now use temporaryQueues, we could simply ignore this
+ // and the server temporary queue would remove this as soon as the queue was removed
+ session.deleteQueue(tmpQueue);
+ }
}
}
- }
- catch (Throwable t) {
- ActiveMQRALogger.LOGGER.debug("Error closing core-queue consumer", t);
- }
-
- try {
- if (session != null) {
- session.close();
+ catch (Throwable t) {
+ ActiveMQRALogger.LOGGER.debug("Error closing core-queue consumer", t);
}
- }
- catch (Throwable t) {
- ActiveMQRALogger.LOGGER.debug("Error releasing session " + session, t);
- }
- try {
- if (cf != null) {
- cf.close();
+ try {
+ if (session != null) {
+ session.close();
+ }
+ }
+ catch (Throwable t) {
+ ActiveMQRALogger.LOGGER.debug("Error releasing session " + session, t);
+ }
+ try {
+ if (cf != null) {
+ cf.close();
+ }
+ }
+ catch (Throwable t) {
+ ActiveMQRALogger.LOGGER.debug("Error releasing session factory " + session, t);
}
}
- catch (Throwable t) {
- ActiveMQRALogger.LOGGER.debug("Error releasing session factory " + session, t);
+ else {
+ //otherwise we just clean up
+ try {
+ if (cf != null) {
+ cf.cleanup();
+ }
+ }
+ catch (Throwable t) {
+ ActiveMQRALogger.LOGGER.debug("Error releasing session factory " + session, t);
+ }
+
}
}
@@ -364,4 +384,9 @@ public class ActiveMQMessageHandler implements MessageHandler {
public void start() throws ActiveMQException {
session.start();
}
+
+ @Override
+ public void failoverEvent(FailoverEventType eventType) {
+ connected = eventType == FailoverEventType.FAILOVER_COMPLETED;
+ }
}