diff --git a/spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java b/spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java index 86c0b6ab..5f8ead47 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java @@ -18,7 +18,10 @@ package org.springframework.erlang.connection; import java.io.IOException; import java.net.UnknownHostException; +import java.util.UUID; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.erlang.OtpIOException; import org.springframework.util.Assert; @@ -75,6 +78,10 @@ import com.ericsson.otp.erlang.OtpSelf; */ public class SimpleConnectionFactory implements ConnectionFactory, InitializingBean { + protected final Log logger = LogFactory.getLog(getClass()); + + private boolean uniqueSelfNodeName = true; + private String selfNodeName; private String cookie; @@ -106,22 +113,36 @@ public class SimpleConnectionFactory implements ConnectionFactory, InitializingB + "' to peer node '" + this.peerNodeName + "'", ex); } } + + public boolean isUniqueSelfNodeName() { + return uniqueSelfNodeName; + } + + public void setUniqueSelfNodeName(boolean uniqueSelfNodeName) { + this.uniqueSelfNodeName = uniqueSelfNodeName; + } public void afterPropertiesSet() { Assert.isTrue(this.selfNodeName != null || this.peerNodeName != null, "'selfNodeName' or 'peerNodeName' is required"); + String selfNodeNameToUse = this.selfNodeName; + if (isUniqueSelfNodeName()) { + selfNodeNameToUse = this.selfNodeName + "-" + UUID.randomUUID().toString(); + logger.debug("Creating OtpSelf with node name = [" + selfNodeNameToUse + "]"); + } try { if (this.cookie == null) { - this.otpSelf = new OtpSelf(this.selfNodeName); + this.otpSelf = new OtpSelf(selfNodeNameToUse.trim()); } else { - this.otpSelf = new OtpSelf(this.selfNodeName, this.cookie); + this.otpSelf = new OtpSelf(selfNodeNameToUse.trim(), this.cookie); } } catch (IOException e) { throw new OtpIOException(e); } - this.otpPeer = new OtpPeer(this.peerNodeName); + this.otpPeer = new OtpPeer(this.peerNodeName.trim()); + } } diff --git a/spring-erlang/src/main/java/org/springframework/erlang/connection/SingleConnectionFactory.java b/spring-erlang/src/main/java/org/springframework/erlang/connection/SingleConnectionFactory.java index 35d5f7ac..20a7e6b3 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/connection/SingleConnectionFactory.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/connection/SingleConnectionFactory.java @@ -24,9 +24,11 @@ import java.lang.reflect.Proxy; import java.net.UnknownHostException; import java.util.ArrayList; import java.util.List; +import java.util.UUID; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.erlang.OtpIOException; import org.springframework.util.Assert; @@ -78,9 +80,11 @@ import com.ericsson.otp.erlang.OtpSelf; * @author Mark Pollack */ public class SingleConnectionFactory implements ConnectionFactory, - InitializingBean { + InitializingBean, DisposableBean { protected final Log logger = LogFactory.getLog(getClass()); + + private boolean uniqueSelfNodeName = true; private String selfNodeName; @@ -112,6 +116,14 @@ public class SingleConnectionFactory implements ConnectionFactory, this.selfNodeName = selfNodeName; this.peerNodeName = peerNodeName; } + + public boolean isUniqueSelfNodeName() { + return uniqueSelfNodeName; + } + + public void setUniqueSelfNodeName(boolean uniqueSelfNodeName) { + this.uniqueSelfNodeName = uniqueSelfNodeName; + } public Connection createConnection() throws UnknownHostException, OtpAuthException { @@ -144,6 +156,29 @@ public class SingleConnectionFactory implements ConnectionFactory, this.connection = getSharedConnectionProxy(this.targetConnection); } } + + /** + * Close the underlying shared connection. + * The provider of this ConnectionFactory needs to care for proper shutdown. + *

As this bean implements DisposableBean, a bean factory will + * automatically invoke this on destruction of its cached singletons. + */ + public void destroy() { + resetConnection(); + } + + /** + * Reset the underlying shared Connection, to be reinitialized on next access. + */ + public void resetConnection() { + synchronized (this.connectionMonitor) { + if (this.targetConnection != null) { + closeConnection(this.targetConnection); + } + this.targetConnection = null; + this.connection = null; + } + } /** * Close the given Connection. @@ -207,16 +242,21 @@ public class SingleConnectionFactory implements ConnectionFactory, public void afterPropertiesSet() { Assert.isTrue(this.selfNodeName != null || this.peerNodeName != null, "'selfNodeName' or 'peerNodeName' is required"); + String selfNodeNameToUse = this.selfNodeName; + if (isUniqueSelfNodeName()) { + selfNodeNameToUse = this.selfNodeName + "-" + UUID.randomUUID().toString(); + logger.debug("Creating OtpSelf with node name = [" + selfNodeNameToUse + "]"); + } try { if (this.cookie == null) { - this.otpSelf = new OtpSelf(this.selfNodeName); + this.otpSelf = new OtpSelf(selfNodeNameToUse.trim()); } else { - this.otpSelf = new OtpSelf(this.selfNodeName, this.cookie); + this.otpSelf = new OtpSelf(selfNodeNameToUse.trim(), this.cookie); } } catch (IOException e) { throw new OtpIOException(e); } - this.otpPeer = new OtpPeer(this.peerNodeName); + this.otpPeer = new OtpPeer(this.peerNodeName.trim()); } private class SharedConnectionInvocationHandler implements diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java index 6bfff860..bfbdcd1b 100644 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java +++ b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java @@ -38,6 +38,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.erlang.OtpAuthException; import org.springframework.erlang.OtpIOException; import org.springframework.erlang.connection.SimpleConnectionFactory; +import org.springframework.erlang.connection.SingleConnectionFactory; import org.springframework.erlang.core.Application; import org.springframework.erlang.core.ErlangTemplate; import org.springframework.erlang.core.Node; @@ -376,7 +377,7 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { } String peerNodeName = "rabbit@" + host; logger.debug("Creating jinterface connection with peerNodeName = [" + peerNodeName + "]"); - SimpleConnectionFactory otpCf = new SimpleConnectionFactory("rabbit-spring-monitor", peerNodeName); + SingleConnectionFactory otpCf = new SingleConnectionFactory("rabbit-spring-monitor", peerNodeName); otpCf.afterPropertiesSet(); createErlangTemplate(otpCf); } diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java index 381407e4..6a920ab9 100644 --- a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java +++ b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java @@ -9,9 +9,8 @@ import java.net.UnknownHostException; import junit.framework.Assert; -import org.junit.Ignore; import org.junit.Test; -import org.springframework.erlang.connection.SimpleConnectionFactory; +import org.springframework.erlang.connection.SingleConnectionFactory; import org.springframework.erlang.core.ErlangTemplate; import com.ericsson.otp.erlang.OtpAuthException; @@ -26,14 +25,14 @@ import com.ericsson.otp.erlang.OtpSelf; //@Ignore("manual integration test only.") public class JInterfaceIntegrationTests { + private static int counter; @Test public void rawApi() { OtpConnection connection = null; try { OtpSelf self = new OtpSelf("rabbit-monitor"); - String hostName = "rabbit@vmc-ssrc-rh82"; - //+ InetAddress.getLocalHost().getHostName(); + String hostName = "rabbit@" + InetAddress.getLocalHost().getHostName(); OtpPeer peer = new OtpPeer(hostName); connection = self.connect(peer); // connection.sendRPC("erlang","date", new OtpErlangList()); @@ -77,7 +76,7 @@ public class JInterfaceIntegrationTests { //System.out.println("home = " + home); //System.out.println("peerNodeName = " + peerNodeName); - SimpleConnectionFactory cf = new SimpleConnectionFactory(selfNodeName, + SingleConnectionFactory cf = new SingleConnectionFactory(selfNodeName, peerNodeName); cf.afterPropertiesSet(); @@ -90,6 +89,8 @@ public class JInterfaceIntegrationTests { long number = (Long) template.executeAndConvertRpc("erlang", "abs", -161803399); Assert.assertEquals(161803399, number); + + cf.destroy(); } @@ -117,8 +118,8 @@ public class JInterfaceIntegrationTests { } public OtpConnection createConnection() throws Exception { - OtpSelf self = new OtpSelf("rabbit-monitor"); - OtpPeer peer = new OtpPeer("rabbit");// + InetAddress.getLocalHost().getHostName()); + OtpSelf self = new OtpSelf("rabbit-monitor-" + counter++); + OtpPeer peer = new OtpPeer("rabbit@" + InetAddress.getLocalHost().getHostName()); return self.connect(peer); } diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java index 6fc81b8d..0efbf3c7 100644 --- a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java +++ b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java @@ -23,14 +23,16 @@ import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.junit.AfterClass; import org.junit.BeforeClass; +import org.junit.Ignore; import org.junit.Test; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; import org.springframework.erlang.OtpIOException; /** + * + * This test class assumes that you are already running the rabbitmq broker. * @author Mark Pollack */ public class RabbitBrokerAdminIntegrationTests { @@ -47,15 +49,6 @@ public class RabbitBrokerAdminIntegrationTests { connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); brokerAdmin = new RabbitBrokerAdmin(connectionFactory); - logger.info("Starting broker node"); - brokerAdmin.startNode(); - Thread.sleep(1000L); - } - - @AfterClass - public static void tearDown() { - logger.info("Shutting down broker node"); - brokerAdmin.stopNode(); } @Test @@ -102,7 +95,6 @@ public class RabbitBrokerAdminIntegrationTests { for (int i = 1; i< 100; i++) { testStatusAndBrokerLifecycle(); System.out.println("i = " + i); - //Thread.sleep(1000); } } @@ -123,7 +115,7 @@ public class RabbitBrokerAdminIntegrationTests { @Test - //@Ignore("NEEDS RABBITMQ_HOME to be set.") + @Ignore("Test Manually") public void testStartNode() { try { brokerAdmin.stopNode(); @@ -132,12 +124,10 @@ public class RabbitBrokerAdminIntegrationTests { } brokerAdmin.startNode(); assertEquals(1,1); - brokerAdmin.stopNode(); } @Test public void testGetQueues() throws Exception { - Thread.sleep(1000L); brokerAdmin.declareQueue(new Queue("test.queue")); assertEquals("/", connectionFactory.getVirtualHost()); List queues = brokerAdmin.getQueues();