https://issues.apache.org/jira/browse/AMQ-6334

Fix error handling for connection establishment on the bridge.
fenglove · Jun 23, 2016 · 03785a4 · 03785a4
1 parent d563e90
commit 03785a4
Show file tree

Hide file tree

Showing 2 changed files with 202 additions and 162 deletions.
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/jms/SimpleJmsQueueConnector.java b/activemq-broker/src/main/java/org/apache/activemq/network/jms/SimpleJmsQueueConnector.java
@@ -158,122 +158,142 @@ public void setOutboundQueueConnectionFactory(QueueConnectionFactory foreignQueu
     @Override
     protected void initializeForeignConnection() throws NamingException, JMSException {
 
-        final QueueConnection newConnection;
-
-        if (foreignConnection.get() == null) {
-            // get the connection factories
-            if (outboundQueueConnectionFactory == null) {
-                // look it up from JNDI
-                if (outboundQueueConnectionFactoryName != null) {
-                    outboundQueueConnectionFactory = jndiOutboundTemplate
-                        .lookup(outboundQueueConnectionFactoryName, QueueConnectionFactory.class);
+        QueueConnection newConnection = null;
+
+        try {
+            if (foreignConnection.get() == null) {
+                // get the connection factories
+                if (outboundQueueConnectionFactory == null) {
+                    // look it up from JNDI
+                    if (outboundQueueConnectionFactoryName != null) {
+                        outboundQueueConnectionFactory = jndiOutboundTemplate
+                            .lookup(outboundQueueConnectionFactoryName, QueueConnectionFactory.class);
+                        if (outboundUsername != null) {
+                            newConnection = outboundQueueConnectionFactory
+                                .createQueueConnection(outboundUsername, outboundPassword);
+                        } else {
+                            newConnection = outboundQueueConnectionFactory.createQueueConnection();
+                        }
+                    } else {
+                        throw new JMSException("Cannot create foreignConnection - no information");
+                    }
+                } else {
                     if (outboundUsername != null) {
                         newConnection = outboundQueueConnectionFactory
                             .createQueueConnection(outboundUsername, outboundPassword);
                     } else {
                         newConnection = outboundQueueConnectionFactory.createQueueConnection();
                     }
-                } else {
-                    throw new JMSException("Cannot create foreignConnection - no information");
                 }
             } else {
-                if (outboundUsername != null) {
-                    newConnection = outboundQueueConnectionFactory
-                        .createQueueConnection(outboundUsername, outboundPassword);
-                } else {
-                    newConnection = outboundQueueConnectionFactory.createQueueConnection();
-                }
+                // Clear if for now in case something goes wrong during the init.
+                newConnection = (QueueConnection) foreignConnection.getAndSet(null);
             }
-        } else {
-            // Clear if for now in case something goes wrong during the init.
-            newConnection = (QueueConnection) foreignConnection.getAndSet(null);
-        }
 
-        if (outboundClientId != null && outboundClientId.length() > 0) {
-            newConnection.setClientID(getOutboundClientId());
-        }
-        newConnection.start();
+            // Register for any async error notifications now so we can reset in the
+            // case where there's not a lot of activity and a connection drops.
+            newConnection.setExceptionListener(new ExceptionListener() {
+                @Override
+                public void onException(JMSException exception) {
+                    handleConnectionFailure(foreignConnection.get());
+                }
+            });
+
+            if (outboundClientId != null && outboundClientId.length() > 0) {
+                newConnection.setClientID(getOutboundClientId());
+            }
+            newConnection.start();
 
-        outboundMessageConvertor.setConnection(newConnection);
+            outboundMessageConvertor.setConnection(newConnection);
 
-        // Configure the bridges with the new Outbound connection.
-        initializeInboundDestinationBridgesOutboundSide(newConnection);
-        initializeOutboundDestinationBridgesOutboundSide(newConnection);
+            // Configure the bridges with the new Outbound connection.
+            initializeInboundDestinationBridgesOutboundSide(newConnection);
+            initializeOutboundDestinationBridgesOutboundSide(newConnection);
 
-        // Register for any async error notifications now so we can reset in the
-        // case where there's not a lot of activity and a connection drops.
-        newConnection.setExceptionListener(new ExceptionListener() {
-            @Override
-            public void onException(JMSException exception) {
-                handleConnectionFailure(newConnection);
+            // At this point all looks good, so this our current connection now.
+            foreignConnection.set(newConnection);
+        } catch (Exception ex) {
+            if (newConnection != null) {
+                try {
+                    newConnection.close();
+                } catch (Exception ignore) {}
             }
-        });
 
-        // At this point all looks good, so this our current connection now.
-        foreignConnection.set(newConnection);
+            throw ex;
+        }
     }
 
     @Override
     protected void initializeLocalConnection() throws NamingException, JMSException {
 
-        final QueueConnection newConnection;
-
-        if (localConnection.get() == null) {
-            // get the connection factories
-            if (localQueueConnectionFactory == null) {
-                if (embeddedConnectionFactory == null) {
-                    // look it up from JNDI
-                    if (localConnectionFactoryName != null) {
-                        localQueueConnectionFactory = jndiLocalTemplate
-                            .lookup(localConnectionFactoryName, QueueConnectionFactory.class);
-                        if (localUsername != null) {
-                            newConnection = localQueueConnectionFactory
-                                .createQueueConnection(localUsername, localPassword);
+        QueueConnection newConnection = null;
+
+        try {
+            if (localConnection.get() == null) {
+                // get the connection factories
+                if (localQueueConnectionFactory == null) {
+                    if (embeddedConnectionFactory == null) {
+                        // look it up from JNDI
+                        if (localConnectionFactoryName != null) {
+                            localQueueConnectionFactory = jndiLocalTemplate
+                                .lookup(localConnectionFactoryName, QueueConnectionFactory.class);
+                            if (localUsername != null) {
+                                newConnection = localQueueConnectionFactory
+                                    .createQueueConnection(localUsername, localPassword);
+                            } else {
+                                newConnection = localQueueConnectionFactory.createQueueConnection();
+                            }
                         } else {
-                            newConnection = localQueueConnectionFactory.createQueueConnection();
+                            throw new JMSException("Cannot create localConnection - no information");
                         }
                     } else {
-                        throw new JMSException("Cannot create localConnection - no information");
+                        newConnection = embeddedConnectionFactory.createQueueConnection();
                     }
                 } else {
-                    newConnection = embeddedConnectionFactory.createQueueConnection();
+                    if (localUsername != null) {
+                        newConnection = localQueueConnectionFactory.
+                                createQueueConnection(localUsername, localPassword);
+                    } else {
+                        newConnection = localQueueConnectionFactory.createQueueConnection();
+                    }
                 }
+
             } else {
-                if (localUsername != null) {
-                    newConnection = localQueueConnectionFactory.
-                            createQueueConnection(localUsername, localPassword);
-                } else {
-                    newConnection = localQueueConnectionFactory.createQueueConnection();
-                }
+                // Clear if for now in case something goes wrong during the init.
+                newConnection = (QueueConnection) localConnection.getAndSet(null);
             }
 
-        } else {
-            // Clear if for now in case something goes wrong during the init.
-            newConnection = (QueueConnection) localConnection.getAndSet(null);
-        }
+            // Register for any async error notifications now so we can reset in the
+            // case where there's not a lot of activity and a connection drops.
+            newConnection.setExceptionListener(new ExceptionListener() {
+                @Override
+                public void onException(JMSException exception) {
+                    handleConnectionFailure(localConnection.get());
+                }
+            });
 
-        if (localClientId != null && localClientId.length() > 0) {
-            newConnection.setClientID(getLocalClientId());
-        }
-        newConnection.start();
+            if (localClientId != null && localClientId.length() > 0) {
+                newConnection.setClientID(getLocalClientId());
+            }
+            newConnection.start();
 
-        inboundMessageConvertor.setConnection(newConnection);
+            inboundMessageConvertor.setConnection(newConnection);
 
-        // Configure the bridges with the new Local connection.
-        initializeInboundDestinationBridgesLocalSide(newConnection);
-        initializeOutboundDestinationBridgesLocalSide(newConnection);
+            // Configure the bridges with the new Local connection.
+            initializeInboundDestinationBridgesLocalSide(newConnection);
+            initializeOutboundDestinationBridgesLocalSide(newConnection);
 
-        // Register for any async error notifications now so we can reset in the
-        // case where there's not a lot of activity and a connection drops.
-        newConnection.setExceptionListener(new ExceptionListener() {
-            @Override
-            public void onException(JMSException exception) {
-                handleConnectionFailure(newConnection);
+            // At this point all looks good, so this our current connection now.
+            localConnection.set(newConnection);
+        } catch (Exception ex) {
+            if (newConnection != null) {
+                try {
+                    newConnection.close();
+                } catch (Exception ignore) {}
             }
-        });
 
-        // At this point all looks good, so this our current connection now.
-        localConnection.set(newConnection);
+            throw ex;
+        }
     }
 
     protected void initializeInboundDestinationBridgesOutboundSide(QueueConnection connection) throws JMSException {