From 1c26c9aba0c2184e922e2fc80bd604ed0a7244ba Mon Sep 17 00:00:00 2001
From: Robert Davies <rajdavies@apache.org>
Date: Sun, 21 Feb 2010 07:36:55 +0000
Subject: [PATCH] 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
---
 .../activemq/network/jms/DestinationBridge.java  | 10 ++++++----
 .../activemq/network/jms/JmsQueueConnector.java  | 16 +++++++++++++++-
 2 files changed, 21 insertions(+), 5 deletions(-)

diff --git a/activemq-core/src/main/java/org/apache/activemq/network/jms/DestinationBridge.java b/activemq-core/src/main/java/org/apache/activemq/network/jms/DestinationBridge.java
index 12379be6bc..d4977264e4 100755
--- a/activemq-core/src/main/java/org/apache/activemq/network/jms/DestinationBridge.java
+++ b/activemq-core/src/main/java/org/apache/activemq/network/jms/DestinationBridge.java
@@ -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);
diff --git a/activemq-core/src/main/java/org/apache/activemq/network/jms/JmsQueueConnector.java b/activemq-core/src/main/java/org/apache/activemq/network/jms/JmsQueueConnector.java
index 51d0bfaba9..44a22dc47a 100755
--- a/activemq-core/src/main/java/org/apache/activemq/network/jms/JmsQueueConnector.java
+++ b/activemq-core/src/main/java/org/apache/activemq/network/jms/JmsQueueConnector.java
@@ -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) {