Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Applied patch for https://issues.apache.org/activemq/browse/AMQ-2455 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@912312 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2010-02-21 07:36:55 +00:00
parent 746253c8ef
commit 1c26c9aba0
2 changed files with 21 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
activemq-core/src/main/java/org/apache/activemq/network/jms/DestinationBridge.java
View File

@ -17,7 +17,6 @@
package org.apache.activemq.network.jms;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
@ -26,7 +25,6 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.naming.NamingException;
import org.apache.activemq.Service;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -109,8 +107,9 @@ public abstract class DestinationBridge implements Service, MessageListener {
}
public void onMessage(Message message) {
if (started.get() && message != null) {
int attempt = 0;
int attempt = 0;
while (started.get() && message != null) {
try {
if (attempt > 0) {
restartProducer();
@ -129,6 +128,7 @@ public abstract class DestinationBridge implements Service, MessageListener {
}
sendMessage(converted);
message.acknowledge();
return;
} catch (Exception e) {
LOG.error("failed to forward message on attempt: " + (++attempt) + " reason: " + e + " message: " + message, e);
if (maximumRetries > 0 && attempt >= maximumRetries) {
@ -168,6 +168,8 @@ public abstract class DestinationBridge implements Service, MessageListener {
protected void restartProducer() throws JMSException, NamingException {
try {
//don't reconnect immediately
Thread.sleep(1000);
getConnectionForProducer().close();
} catch (Exception e) {
LOG.debug("Ignoring failure to close producer connection: " + e, e);

16
activemq-core/src/main/java/org/apache/activemq/network/jms/JmsQueueConnector.java
View File

@ -182,6 +182,20 @@ public class JmsQueueConnector extends JmsConnector {
public void restartProducerConnection() throws NamingException, JMSException {
outboundQueueConnection = null;
initializeForeignQueueConnection();
// the outboundQueueConnection was reestablished - publish the new connection to the bridges
if (inboundQueueBridges != null) {
for (int i = 0; i < inboundQueueBridges.length; i++) {
InboundQueueBridge bridge = inboundQueueBridges[i];
bridge.setConsumerConnection(outboundQueueConnection);
}
}
if (outboundQueueBridges != null) {
for (int i = 0; i < outboundQueueBridges.length; i++) {
OutboundQueueBridge bridge = outboundQueueBridges[i];
bridge.setProducerConnection(outboundQueueConnection);
}
}
}
protected void initializeForeignQueueConnection() throws NamingException, JMSException {
@ -199,7 +213,7 @@ public class JmsQueueConnector extends JmsConnector {
outboundQueueConnection = outboundQueueConnectionFactory.createQueueConnection();
}
} else {
throw new JMSException("Cannot create localConnection - no information");
throw new JMSException("Cannot create foreignConnection - no information");
}
} else {
if (outboundUsername != null) {