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 0039b865..66fe8475 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 @@ -68,6 +68,7 @@ import com.ericsson.otp.erlang.OtpSelf; *

* @author Mark Pollack * @author Mark Fisher + * @author Dave Syer */ public class SimpleConnectionFactory implements ConnectionFactory, InitializingBean { @@ -75,24 +76,24 @@ public class SimpleConnectionFactory implements ConnectionFactory, InitializingB private boolean uniqueSelfNodeName = true; - private String selfNodeName; + private final String selfNodeName; - private String cookie; - - private String peerNodeName; + private final String peerNodeName; + private final String cookie; + private OtpSelf otpSelf; private OtpPeer otpPeer; public SimpleConnectionFactory(String selfNodeName, String peerNodeName, String cookie) { - this(selfNodeName, peerNodeName); + this.selfNodeName = selfNodeName; + this.peerNodeName = peerNodeName; this.cookie = cookie; } public SimpleConnectionFactory(String selfNodeName, String peerNodeName) { - this.selfNodeName = selfNodeName; - this.peerNodeName = peerNodeName; + this(selfNodeName, peerNodeName, null); } public Connection createConnection() throws UnknownHostException, OtpAuthException, IOException { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java index 26fb5883..ff4e7d0c 100755 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java @@ -48,15 +48,32 @@ import org.springframework.util.exec.Execute; import org.springframework.util.exec.Os; /** - * Rabbit broker administration implementation exposed via JMX annotations. + * Rabbit broker administration. Features: + * + * + * + * Depending on your platform, to {@link #startNode() start the broker} you might need to set some environment + * properties. The most common are available via constructors or setters in this class (e.g. + * {@link #setRabbitLogBaseDirectory(String) RABBITMQ_LOG_BASE}). All others you can set via the OS (any setting that + * RabbtMQ allows in its startup script), and some work via System properties as special convenience cases ( + * ERLANG_HOME and RABBITMQ_HOME ). * * @author Mark Pollack + * @author Dave Syer */ public class RabbitBrokerAdmin implements RabbitBrokerOperations { private static final String DEFAULT_VHOST = "/"; - private static String DEFAULT_HOST; + private static String DEFAULT_NODE_NAME; + + private static int DEFAULT_PORT = 5672; private static final String DEFAULT_ENCODING = "UTF-8"; @@ -72,36 +89,110 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { // TODO: extract into field for DI private SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); - private final String hostName; + private final String nodeName; private final String cookie; - // TODO: RABBITMQ_NODE_PORT=5672 + private final int port; + + private String rabbitLogBaseDirectory; + + private String rabbitMnesiaBaseDirectory; static { try { - DEFAULT_HOST = InetAddress.getLocalHost().getHostName(); + String hostName = InetAddress.getLocalHost().getHostName(); + DEFAULT_NODE_NAME = "rabbit@" + hostName; } catch (UnknownHostException e) { - DEFAULT_HOST = "localhost"; + DEFAULT_NODE_NAME = "rabbit@localhost"; } } public RabbitBrokerAdmin() { - this(DEFAULT_HOST); + this(DEFAULT_NODE_NAME); } - public RabbitBrokerAdmin(String hostName) { - this(hostName, null); + /** + * Create an instance by supplying the erlang node name (e.g. "rabbit@myserver"), or simply the hostname (if the + * alive name is "rabbit"). + * + * @param nodeName the node name or hostname to use + */ + public RabbitBrokerAdmin(String nodeName) { + this(nodeName, null); } - - public RabbitBrokerAdmin(String hostName, String cookie) { - if (Os.isFamily("windows") && !DEFAULT_HOST.equals(hostName)) { - hostName = hostName.toUpperCase(); + + /** + * Create an instance by supplying the erlang node name and cookie (unique string). + * + * @param nodeName the node name or hostname to use + * + * @param cookie the cookie value to use + */ + public RabbitBrokerAdmin(String nodeName, String cookie) { + this(nodeName, DEFAULT_PORT, cookie); + } + + /** + * Create an instance by supplying the erlang node name and port number. Use this on a UN*X system if you want to + * run the broker as a user without root privileges, supplying values that do not clash with the default broker + * (usually "rabbit@<servername>" and 5672). If, as well as managing an existing broker, you need to start the + * broker process, you will also need to set {@link #setRabbitLogBaseDirectory(String) RABBITMQ_LOG_BASE} and + * {@link #setRabbitMnesiaBaseDirectory(String)RABBITMQ_MNESIA_BASE} to point to writable directories). + * + * @param nodeName the node name or hostname to use + * @param port the port number (overriding the default which is 5672) + */ + public RabbitBrokerAdmin(String nodeName, int port) { + this(nodeName, port, null); + } + + /** + * Create an instance by supplying the erlang node name, port number and cookie (unique string). + * + * @param nodeName the node name or hostname to use + * @param port the port number (overriding the default which is 5672) + * @param cookie the cookie value to use + */ + public RabbitBrokerAdmin(String nodeName, int port, String cookie) { + + if (!nodeName.contains("@")) { + nodeName = "rabbit@" + nodeName; // it was just the host } + + String[] parts = nodeName.split("@"); + Assert.state(parts.length == 2, "The node name should be in the form alivename@host, e.g. rabbit@myserver"); + if (Os.isFamily("windows") && !DEFAULT_NODE_NAME.equals(nodeName)) { + nodeName = parts[0] + "@" + parts[1].toUpperCase(); + } + + this.port = port; this.cookie = cookie; - this.hostName = hostName; + this.nodeName = nodeName; this.executor.setDaemon(true); - initializeDefaultErlangTemplate(hostName); + + initializeDefaultErlangTemplate(nodeName); + + } + + /** + * The location of RABBITMQ_LOG_BASE to override the system default (which may be owned by another + * user). Only needed for launching the broker process. Can also be set as a system property. + * + * @param rabbitLogBaseDirectory the rabbit log base directory to set + */ + public void setRabbitLogBaseDirectory(String rabbitLogBaseDirectory) { + this.rabbitLogBaseDirectory = rabbitLogBaseDirectory; + } + + /** + * The location of RABBITMQ_MNESIA_BASE to override the system default (which may be owned by another + * user). Only needed for launching the broker process. Can also be set as a system property. + * + * @param rabbitMnesiaBaseDirectory the rabbit Mnesia base directory to set + */ + public void setRabbitMnesiaBaseDirectory(String rabbitMnesiaBaseDirectory) { + this.rabbitMnesiaBaseDirectory = rabbitMnesiaBaseDirectory; } /** @@ -293,12 +384,24 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { String[] commandline = new String[] { rabbitStartCommand }; List env = new ArrayList(); - addEnvironment(env, "RABBITMQ_LOG_BASE"); - addEnvironment(env, "RABBITMQ_MNESIA_BASE"); + + if (rabbitLogBaseDirectory != null) { + env.add("RABBITMQ_LOG_BASE=" + rabbitLogBaseDirectory); + } else { + addEnvironment(env, "RABBITMQ_LOG_BASE"); + } + if (rabbitMnesiaBaseDirectory != null) { + env.add("RABBITMQ_MNESIA_BASE=" + rabbitMnesiaBaseDirectory); + } else { + addEnvironment(env, "RABBITMQ_MNESIA_BASE"); + } addEnvironment(env, "ERLANG_HOME"); - // Make the hostname explicitly the same so the erl process knows who we are - env.add("HOSTNAME=" + hostName); + // Make the nodename explicitly the same so the erl process knows who we are + env.add("RABBITMQ_NODENAME=" + nodeName); + + // Set the port number for the new process + env.add("RABBITMQ_NODE_PORT=" + port); // Ask for a detached erl process so stdout doesn't get diverted to a black hole when the JVM dies (without this // you can start the Rabbit broker form Java but if you forget to stop it, the erl process is hosed). @@ -518,7 +621,7 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { } protected void initializeDefaultErlangTemplate(String host) { - String peerNodeName = "rabbit@" + host; + String peerNodeName = nodeName; logger.debug("Creating jinterface connection with peerNodeName = [" + peerNodeName + "]"); SimpleConnectionFactory otpConnectionFactory = new SimpleConnectionFactory("rabbit-spring-monitor", peerNodeName, this.cookie); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java index 63343fa5..a20b1fbb 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java @@ -61,9 +61,9 @@ public class RabbitBrokerAdminIntegrationTests { @BeforeClass public static void start() throws Exception { - System.setProperty("RABBITMQ_LOG_BASE", "target/rabbitmq/log"); - System.setProperty("RABBITMQ_MNESIA_BASE", "target/rabbitmq/mnesia"); - brokerAdmin = new RabbitBrokerAdmin(); + brokerAdmin = new RabbitBrokerAdmin("spring@localhost", 15672); + brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log"); + brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia"); brokerAdmin.setStartupTimeout(10000L); brokerAdmin.startNode(); } @@ -73,8 +73,6 @@ public class RabbitBrokerAdminIntegrationTests { if (Os.isFamily("windows") || Os.isFamily("dos")) { brokerAdmin.stopNode(); } - System.clearProperty("RABBITMQ_LOG_BASE"); - System.clearProperty("RABBITMQ_MNESIA_BASE"); } @Test @@ -110,8 +108,9 @@ public class RabbitBrokerAdminIntegrationTests { @Test public void repeatLifecycle() throws Exception { - for (int i = 1; i < 20; i++) { + for (int i = 1; i <= 20; i++) { testStatusAndBrokerLifecycle(); + Thread.sleep(200); if (i % 5 == 0) { logger.debug("i = " + i); } @@ -129,7 +128,7 @@ public class RabbitBrokerAdminIntegrationTests { private void assertBrokerAppRunning(RabbitStatus status) { assertEquals(1, status.getRunningNodes().size()); - assertTrue(status.getRunningNodes().get(0).getName().contains("rabbit")); + assertTrue(status.getRunningNodes().get(0).getName().contains("spring@localhost")); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java index 32374c72..688c0cfd 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java @@ -20,7 +20,6 @@ import java.io.File; import org.apache.commons.io.FileUtils; import org.apache.log4j.Level; -import org.junit.After; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -39,21 +38,16 @@ public class RabbitBrokerAdminStopIntegrationTests { @Before public void init() throws Exception { FileUtils.deleteDirectory(new File("target/rabbitmq")); - System.setProperty("RABBITMQ_LOG_BASE", "target/rabbitmq/log"); - System.setProperty("RABBITMQ_MNESIA_BASE", "target/rabbitmq/mnesia"); - } - - @After - public void close() throws Exception { - System.clearProperty("RABBITMQ_LOG_BASE"); - System.clearProperty("RABBITMQ_MNESIA_BASE"); } @Test - // @Ignore("NEEDS RABBITMQ_HOME to be set.") public void testStartNode() throws Exception { - final RabbitBrokerAdmin brokerAdmin = new RabbitBrokerAdmin(); + // Set up broker admin for non-root user + final RabbitBrokerAdmin brokerAdmin = new RabbitBrokerAdmin("spring@localhost", 15672); + brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log"); + brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia"); + brokerAdmin.setStartupTimeout(10000L); RabbitStatus status = brokerAdmin.getStatus(); @@ -71,7 +65,7 @@ public class RabbitBrokerAdminStopIntegrationTests { brokerAdmin.startBrokerApplication(); } status = brokerAdmin.getStatus(); - + try { assertFalse("Broker node did not start. Check logs for hints.", status.getNodes().isEmpty()); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java index 212ca8f6..224d5982 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java @@ -84,9 +84,7 @@ public class MessageListenerContainerLifecycleIntegrationTests { @Parameters public static List getParameters() { - @SuppressWarnings("unused") - Object[] debug = new Object[] { MessageCount.LOW, Concurrency.LOW, TransactionMode.OFF }; - // return Collections.singletonList(debug); + // return Collections.singletonList(new Object[] { MessageCount.LOW, Concurrency.LOW, TransactionMode.OFF }); return Arrays.asList( // new Object[] { MessageCount.HIGH, Concurrency.LOW, TransactionMode.ON }, // new Object[] { MessageCount.HIGH, Concurrency.LOW, TransactionMode.OFF }); @@ -115,21 +113,22 @@ public class MessageListenerContainerLifecycleIntegrationTests { container.start(); try { boolean waited = latch.await(50, TimeUnit.MILLISECONDS); - assertFalse("Expected time out waiting for message", waited); + logger.info("All messages received before stop: " + waited); + if (messageCount > 1) { + assertFalse("Expected not to receive all messages before stop", waited); + } container.stop(); Thread.sleep(500L); container.start(); if (transactional) { waited = latch.await(5, TimeUnit.SECONDS); assertTrue("Timed out waiting for message", waited); - } - else { + } else { waited = latch.await(500, TimeUnit.MILLISECONDS); // If non-transactional we half expect to lose messages - assertFalse("Expected time out waiting for message", waited); + logger.info("All messages received after stop: " + waited); } - } - finally { + } finally { // Wait for broker communication to finish before trying to stop // container Thread.sleep(300L); @@ -154,14 +153,14 @@ public class MessageListenerContainerLifecycleIntegrationTests { this.fail = fail; } - public void handleMessage(String value) { + public void handleMessage(String value) throws Exception { try { logger.debug(value + count.getAndIncrement()); + Thread.sleep(100L); if (fail) { throw new RuntimeException("Planned failure"); } - } - finally { + } finally { latch.countDown(); } }