From fa81e49809c6559977cf8a1b2aacf7e0d6f36e9e Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Fri, 7 Jan 2011 12:50:47 +0000 Subject: [PATCH] AMQP-59: Merge admin project - Try to get access to error message when connect fails - Set default hostname to OS value - Add -detached --- .../springframework/amqp/core/AmqpAdmin.java | 47 +- .../src/test/resources/log4j.properties | 1 + .../connection/ConnectionParameters.java | 5 +- .../connection/SimpleConnectionFactory.java | 112 +++-- .../erlang/core/Application.java | 5 +- .../erlang/core/ErlangTemplate.java | 4 +- .../org/springframework/erlang/core/Node.java | 5 +- .../springframework/util/exec/Execute.java | 128 +++--- .../util/exec/ExecuteWatchdog.java | 9 +- spring-rabbit-admin/.classpath | 9 - spring-rabbit-admin/.project | 35 -- .../.settings/org.maven.ide.eclipse.prefs | 9 - spring-rabbit-admin/.springBeans | 13 - spring-rabbit-admin/pom.xml | 127 ------ .../amqp/rabbit/test/RabbitConfiguration.java | 42 -- .../test/RabbitTestExecutionListener.java | 136 ------ .../test/SpringRabbitJUnit4ClassRunner.java | 78 ---- ...TestExecutionListenerIntegrationTests.java | 27 -- .../src/test/resources/log4j.properties | 9 - ...onfigureBrokerAnntotationTests-context.xml | 18 - spring-rabbit-admin/template.mf | 13 - spring-rabbit/pom.xml | 5 + .../amqp/rabbit/admin/QueueInfo.java | 0 .../admin/RabbitAdminAuthException.java | 0 .../amqp/rabbit/admin/RabbitBrokerAdmin.java | 401 ++++++++++++------ .../rabbit/admin/RabbitBrokerOperations.java | 38 +- .../admin/RabbitControlErlangConverter.java | 0 .../amqp/rabbit/admin/RabbitStatus.java | 32 +- .../amqp/rabbit/connection/Connection.java | 2 +- .../amqp/rabbit/core/RabbitAdmin.java | 10 + .../admin/JInterfaceIntegrationTests.java | 79 ++-- .../amqp/rabbit/admin/PojoHandler.java | 0 .../RabbitBrokerAdminIntegrationTests.java | 116 ++--- ...RabbitBrokerAdminStopIntegrationTests.java | 71 ++-- .../admin/RabbitConsumerConfiguration.java | 0 .../admin/RabbitProducerConfiguration.java | 0 .../admin/RabbitTemplateConsumerExample.java | 0 .../admin/RabbitTemplateProducerExample.java | 0 .../amqp/rabbit/admin/TestConstants.java | 0 .../rabbit/admin/TestRabbitConfiguration.java | 0 .../amqp/rabbit/test/BrokerPanic.java | 51 +++ 41 files changed, 634 insertions(+), 1003 deletions(-) delete mode 100644 spring-rabbit-admin/.classpath delete mode 100644 spring-rabbit-admin/.project delete mode 100644 spring-rabbit-admin/.settings/org.maven.ide.eclipse.prefs delete mode 100644 spring-rabbit-admin/.springBeans delete mode 100644 spring-rabbit-admin/pom.xml delete mode 100644 spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitConfiguration.java delete mode 100644 spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java delete mode 100644 spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/SpringRabbitJUnit4ClassRunner.java delete mode 100644 spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListenerIntegrationTests.java delete mode 100644 spring-rabbit-admin/src/test/resources/log4j.properties delete mode 100644 spring-rabbit-admin/src/test/resources/org/springframework/amqp/rabbit/test/ConfigureBrokerAnntotationTests-context.xml delete mode 100644 spring-rabbit-admin/template.mf rename {spring-rabbit-admin => spring-rabbit}/src/main/java/org/springframework/amqp/rabbit/admin/QueueInfo.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminAuthException.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java (50%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java (80%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/main/java/org/springframework/amqp/rabbit/admin/RabbitControlErlangConverter.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java (74%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java (65%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/PojoHandler.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java (51%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java (51%) rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/RabbitConsumerConfiguration.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/RabbitProducerConfiguration.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateConsumerExample.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateProducerExample.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/TestConstants.java (100%) mode change 100644 => 100755 rename {spring-rabbit-admin => spring-rabbit}/src/test/java/org/springframework/amqp/rabbit/admin/TestRabbitConfiguration.java (100%) mode change 100644 => 100755 create mode 100755 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerPanic.java diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpAdmin.java b/spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpAdmin.java index 60c34820..fd0748fd 100644 --- a/spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpAdmin.java +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpAdmin.java @@ -1,17 +1,14 @@ /* * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. */ package org.springframework.amqp.core; @@ -30,18 +27,17 @@ public interface AmqpAdmin { void declareExchange(Exchange exchange); /** - * Delete an exchange. Look at implementation specific subclass for implementation specific behavior, for example + * Delete an exchange. Look at implementation specific subclass for implementation specific behavior, for example * for RabbitMQ this will delete the exchange without regard for whether it is in use or not. * @param exchangeName the name of the exchange */ void deleteExchange(String exchangeName); - // Queue Operations - + /** - * Declare a queue whose name is automatically named. It is created with - * exclusive = true, autoDelete=true, and durable = false. + * Declare a queue whose name is automatically named. It is created with exclusive = true, autoDelete=true, and + * durable = false. */ Queue declareQueue(); @@ -52,29 +48,28 @@ public interface AmqpAdmin { void declareQueue(Queue queue); /** - * Delete a queue, without regard for whether it is in use or has messages on it + * Delete a queue, without regard for whether it is in use or has messages on it * @param queueName the name of the queue */ void deleteQueue(String queueName); - // Note that nowait option is not readily exposed in Rabbit Java API but is for Rabbit .NET API. + // Note that nowait option is not readily exposed in Rabbit Java API but is for Rabbit .NET API. /** * Delete a queue * @param queueName the name of the queue * @param unused true if the queue should be deleted only if not in use - * @param empty true if the queue should be deleted only if empty + * @param empty true if the queue should be deleted only if empty */ void deleteQueue(String queueName, boolean unused, boolean empty); /** - * Purges the contents of the given queue. + * Purges the contents of the given queue. * @param queueName the name of the queue * @param noWait true to not await completion of the purge */ void purgeQueue(String queueName, boolean noWait); - // Binding opertaions /** @@ -83,7 +78,11 @@ public interface AmqpAdmin { */ void declareBinding(Binding binding); - - //Note unbindQueue/removeBinding was not introduced until 0.9 of the specification. + /** + * Remove a binding of a queue to an exchange. Note unbindQueue/removeBinding was not introduced until 0.9 of the + * specification. + * @param binding a description of the binding to remove. + */ + void removeBinding(Binding binding); } diff --git a/spring-amqp-core/src/test/resources/log4j.properties b/spring-amqp-core/src/test/resources/log4j.properties index 6ddfb097..4bcc6e81 100644 --- a/spring-amqp-core/src/test/resources/log4j.properties +++ b/spring-amqp-core/src/test/resources/log4j.properties @@ -6,4 +6,5 @@ log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n log4j.category.org.springframework.amqp=DEBUG log4j.category.org.springframework.beans.factory=INFO +log4j.category.org.springframework.util.exec.Execute=TRACE diff --git a/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionParameters.java b/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionParameters.java index 6a98ede4..8032ad64 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionParameters.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionParameters.java @@ -17,6 +17,8 @@ package org.springframework.erlang.connection; +import org.springframework.util.Assert; + import com.ericsson.otp.erlang.OtpPeer; import com.ericsson.otp.erlang.OtpSelf; @@ -32,7 +34,8 @@ public class ConnectionParameters { private OtpPeer otpPeer; public ConnectionParameters(OtpSelf otpSelf, OtpPeer otpPeer) { - //TODO assert not null... + Assert.notNull(otpSelf, "OtpSelf must be non-null"); + Assert.notNull(otpPeer, "OtpPeer must be non-null"); this.otpSelf = otpSelf; this.otpPeer = otpPeer; } 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 5f8ead47..52c1fa51 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 @@ -1,17 +1,14 @@ /* * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. */ package org.springframework.erlang.connection; @@ -31,47 +28,42 @@ import com.ericsson.otp.erlang.OtpPeer; import com.ericsson.otp.erlang.OtpSelf; /** - * A simple implementation of {@link ConnectionFactory} that return a new connection - * for each invocation of the createConnection method. - * - * Note that use of this ConnectionFactory with ErlangTemplate has unstable behavior when - * invoked frequently and will be deprecated. See {@link SingleConnectionFactory} for an - * alternative implementation. - * - * - * Provides a more traditional API to creating a connection to a remote erlang node than - * the JInterface API. - * - *

The following is taken from the JInterface javadocs that describe the valid - * node names that can be used. These naming constraints apply to the string values - * you pass into the node names in SimpleConnectionFactory's constructor. *

- * About nodenames: Erlang nodenames consist of two components, an alivename and - * a hostname separated by '@'. Additionally, there are two nodename formats: - * short and long. Short names are of the form "alive@hostname", while long - * names are of the form "alive@host.fully.qualified.domainname". Erlang has - * special requirements regarding the use of the short and long formats, in - * particular they cannot be mixed freely in a network of communicating nodes, - * however Jinterface makes no distinction. See the Erlang documentation for - * more information about nodenames. + * A simple implementation of {@link ConnectionFactory} that return a new connection for each invocation of the + * createConnection method. + *

+ *

+ * Note that use of this ConnectionFactory with ErlangTemplate has unstable behavior when invoked frequently and will be + * deprecated. See {@link SingleConnectionFactory} for an alternative implementation. + *

+ *

+ * Provides a more traditional API to creating a connection to a remote erlang node than the JInterface API. + *

+ *

+ * The following is taken from the JInterface javadocs that describe the valid node names that can be used. These naming + * constraints apply to the string values you pass into the node names in SimpleConnectionFactory's constructor. + *

+ *

+ * About nodenames: Erlang nodenames consist of two components, an alivename and a hostname separated by '@'. + * Additionally, there are two nodename formats: short and long. Short names are of the form "alive@hostname", while + * long names are of the form "alive@host.fully.qualified.domainname". Erlang has special requirements regarding the use + * of the short and long formats, in particular they cannot be mixed freely in a network of communicating nodes, however + * Jinterface makes no distinction. See the Erlang documentation for more information about nodenames. *

* *

- * The constructors for the AbstractNode classes will create names exactly as - * you provide them as long as the name contains '@'. If the string you provide - * contains no '@', it will be treated as an alivename and the name of the local - * host will be appended, resulting in a shortname. Nodenames longer than 255 - * characters will be truncated without warning. + * The constructors for the AbstractNode classes will create names exactly as you provide them as long as the name + * contains '@'. If the string you provide contains no '@', it will be treated as an alivename and the name of the local + * host will be appended, resulting in a shortname. Nodenames longer than 255 characters will be truncated without + * warning. *

* *

- * Upon initialization, this class attempts to read the file .erlang.cookie in - * the user's home directory, and uses the trimmed first line of the file as the - * default cookie by those constructors lacking a cookie argument. If for any - * reason the file cannot be found or read, the default cookie will be set to - * the empty string (""). The location of a user's home directory is determined - * using the system property "user.home", which may not be automatically set on - * all platforms. + * Upon initialization, this class attempts to read the file .erlang.cookie in the user's home directory, and uses the + * trimmed first line of the file as the default cookie by those constructors lacking a cookie argument. If for any + * reason the file cannot be found or read, the default cookie will be set to the empty string (""). The location of a + * user's home directory is determined using the system property "user.home", which may not be automatically set on all + * platforms. *

* @author Mark Pollack * @author Mark Fisher @@ -79,9 +71,9 @@ 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; @@ -92,7 +84,6 @@ public class SimpleConnectionFactory implements ConnectionFactory, InitializingB private OtpPeer otpPeer; - public SimpleConnectionFactory(String selfNodeName, String cookie, String peerNodeName) { this(selfNodeName, peerNodeName); this.cookie = cookie; @@ -103,17 +94,15 @@ public class SimpleConnectionFactory implements ConnectionFactory, InitializingB this.peerNodeName = peerNodeName; } - public Connection createConnection() throws UnknownHostException, OtpAuthException, IOException { try { return new DefaultConnection(otpSelf.connect(otpPeer)); - } - catch (IOException ex) { - throw new OtpIOException("failed to connect from '" + this.selfNodeName - + "' to peer node '" + this.peerNodeName + "'", ex); + } catch (IOException ex) { + throw new OtpIOException("failed to connect from '" + this.selfNodeName + "' to peer node '" + + this.peerNodeName + "'", ex); } } - + public boolean isUniqueSelfNodeName() { return uniqueSelfNodeName; } @@ -123,8 +112,9 @@ public class SimpleConnectionFactory implements ConnectionFactory, InitializingB } public void afterPropertiesSet() { - Assert.isTrue(this.selfNodeName != null || this.peerNodeName != null, - "'selfNodeName' or 'peerNodeName' is required"); + + Assert.isTrue(this.selfNodeName != null && this.peerNodeName != null, + "'selfNodeName' and 'peerNodeName' are required"); String selfNodeNameToUse = this.selfNodeName; if (isUniqueSelfNodeName()) { selfNodeNameToUse = this.selfNodeName + "-" + UUID.randomUUID().toString(); @@ -133,16 +123,14 @@ public class SimpleConnectionFactory implements ConnectionFactory, InitializingB try { if (this.cookie == null) { this.otpSelf = new OtpSelf(selfNodeNameToUse.trim()); - } - else { + } else { this.otpSelf = new OtpSelf(selfNodeNameToUse.trim(), this.cookie); } - } - catch (IOException e) { + } catch (IOException e) { throw new OtpIOException(e); } this.otpPeer = new OtpPeer(this.peerNodeName.trim()); - + } } diff --git a/spring-erlang/src/main/java/org/springframework/erlang/core/Application.java b/spring-erlang/src/main/java/org/springframework/erlang/core/Application.java index 21480961..8ceca86b 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/core/Application.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/core/Application.java @@ -16,6 +16,8 @@ package org.springframework.erlang.core; +import java.io.Serializable; + /** * Describes an Erlang application. Only three fields are supported as that is the level * of information that rabbitmq returns when performing a status request. @@ -25,7 +27,8 @@ package org.springframework.erlang.core; * @author Mark Pollack * */ -public class Application { +@SuppressWarnings("serial") +public class Application implements Serializable { private String description; diff --git a/spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java b/spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java index 44525667..dd5242e1 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java @@ -124,9 +124,7 @@ public class ErlangTemplate extends ErlangAccessor implements ErlangOperations { finally { org.springframework.erlang.connection.ConnectionFactoryUtils.releaseConnection(con, getConnectionFactory()); } - - - + // TODO: physically close and reopen the connection if there is an exception } diff --git a/spring-erlang/src/main/java/org/springframework/erlang/core/Node.java b/spring-erlang/src/main/java/org/springframework/erlang/core/Node.java index 21ef3203..9a6efe5f 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/core/Node.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/core/Node.java @@ -19,13 +19,16 @@ */ package org.springframework.erlang.core; +import java.io.Serializable; + /** * Simple description class for an Erlang node. * * @author Mark Pollack * */ -public class Node { +@SuppressWarnings("serial") +public class Node implements Serializable { private String name; diff --git a/spring-erlang/src/main/java/org/springframework/util/exec/Execute.java b/spring-erlang/src/main/java/org/springframework/util/exec/Execute.java index 0fa825d0..a29a67b0 100644 --- a/spring-erlang/src/main/java/org/springframework/util/exec/Execute.java +++ b/spring-erlang/src/main/java/org/springframework/util/exec/Execute.java @@ -1,17 +1,14 @@ /* * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. */ package org.springframework.util.exec; @@ -24,21 +21,22 @@ import java.io.StringReader; import java.util.StringTokenizer; import java.util.Vector; -/* Derived from ant exec task. All the 'backward compat with jdk1.1, 1.2' - removed. Since jdk1.3 supports working dir, no need for scripts. - - All ant-specific code has been removed as well, this is a completely - independent component. - - Costin - */ +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; /** - * Runs an external program. + * Runs an external program. Derived from ant exec task. All the 'backward compat with jdk1.1, 1.2' removed. Since + * jdk1.3 supports working dir, no need for scripts. All ant-specific code has been removed as well, this is a + * completely independent component. + * * * @author thomas.haas@softwired-inc.com + * @author Costin Leau */ public class Execute { + + private static Log log = LogFactory.getLog(Execute.class); + /** Invalid exit code. **/ public final static int INVALID = Integer.MAX_VALUE; @@ -72,8 +70,7 @@ public class Execute { // Just try to use what we got } - BufferedReader in = new BufferedReader(new StringReader( - out.toString())); + BufferedReader in = new BufferedReader(new StringReader(out.toString())); String var = null; String line, lineSep = System.getProperty("line.separator"); while ((line = in.readLine()) != null) { @@ -128,8 +125,7 @@ public class Execute { } /** - * Creates a new execute object using PumpStreamHandler for - * stream handling. + * Creates a new execute object using PumpStreamHandler for stream handling. */ public Execute() { this(new PumpStreamHandler(), null); @@ -138,9 +134,7 @@ public class Execute { /** * Creates a new execute object. * - * @param streamHandler - * the stream handler used to handle the input and output streams - * of the subprocess. + * @param streamHandler the stream handler used to handle the input and output streams of the subprocess. */ public Execute(ExecuteStreamHandler streamHandler) { this(streamHandler, null); @@ -149,12 +143,8 @@ public class Execute { /** * Creates a new execute object. * - * @param streamHandler - * the stream handler used to handle the input and output streams - * of the subprocess. - * @param watchdog - * a watchdog for the subprocess or null to to - * disable a timeout for the subprocess. + * @param streamHandler the stream handler used to handle the input and output streams of the subprocess. + * @param watchdog a watchdog for the subprocess or null to to disable a timeout for the subprocess. */ public Execute(ExecuteStreamHandler streamHandler, ExecuteWatchdog watchdog) { this.streamHandler = streamHandler; @@ -177,8 +167,7 @@ public class Execute { /** * Sets the commandline of the subprocess to launch. * - * @param commandline - * the commandline of the subprocess to launch + * @param commandline the commandline of the subprocess to launch */ public void setCommandline(String[] commandline) { cmdl = commandline; @@ -187,8 +176,7 @@ public class Execute { /** * Set whether to propagate the default environment or not. * - * @param newenv - * whether to propagate the process environment. + * @param newenv whether to propagate the process environment. */ public void setNewenvironment(boolean newenv) { newEnvironment = newenv; @@ -208,9 +196,8 @@ public class Execute { /** * Sets the environment variables for the subprocess to launch. * - * @param env - * array of Strings, each element of which has an environment - * variable settings in format key=value + * @param env array of Strings, each element of which has an environment variable settings in format + * key=value */ public void setEnvironment(String[] env) { this.env = env; @@ -219,13 +206,10 @@ public class Execute { /** * Sets the working directory of the process to execute. * - *

- * This is emulated using the antRun scripts unless the OS is Windows NT in - * which case a cmd.exe is spawned, or MRJ and setting user.dir works, or - * JDK 1.3 and there is official support in java.lang.Runtime. + *

This is emulated using the antRun scripts unless the OS is Windows NT in which case a cmd.exe is spawned, or + * MRJ and setting user.dir works, or JDK 1.3 and there is official support in java.lang.Runtime. * - * @param wd - * the working directory of the process. + * @param wd the working directory of the process. */ public void setWorkingDirectory(File wd) { workingDirectory = wd; @@ -235,12 +219,10 @@ public class Execute { * Runs a process defined by the command line and returns its exit status. * * @return the exit status of the subprocess or INVALID - * @throws Exception - * if launching of the subprocess failed + * @throws Exception if launching of the subprocess failed */ public int execute() throws Exception { - process = Runtime.getRuntime().exec(getCommandline(), getEnvironment(), - workingDirectory); + process = Runtime.getRuntime().exec(getCommandline(), getEnvironment(), workingDirectory); try { streamHandler.setProcessInputStream(process.getOutputStream()); streamHandler.setProcessOutputStream(process.getInputStream()); @@ -251,23 +233,29 @@ public class Execute { } streamHandler.start(); - if (watchdog != null) + if (watchdog != null) { watchdog.start(process, Thread.currentThread()); + } - if (log.isTraceEnabled()) + if (log.isTraceEnabled()) { log.trace("Waiting process "); + } waitFor(process); process = null; - if (log.isTraceEnabled()) + if (log.isTraceEnabled()) { log.trace("End waiting, stop threads "); - if (watchdog != null) + } + if (watchdog != null) { watchdog.stop(); - if (log.isTraceEnabled()) + } + if (log.isTraceEnabled()) { log.trace("Watchdog stopped "); + } streamHandler.stop(); - if (log.isTraceEnabled()) + if (log.isTraceEnabled()) { log.trace("Stream handler stopped "); + } if (watchdog != null) { Exception ex = watchdog.getException(); if (ex != null) @@ -302,6 +290,7 @@ public class Execute { setExitValue(process.exitValue()); } catch (InterruptedException e) { log.info("waitFor() interrupted "); + Thread.currentThread().interrupt(); } } @@ -312,8 +301,7 @@ public class Execute { /** * query the exit value of the process. * - * @return the exit value, 1 if the process was killed, or Project.INVALID - * if no exit value has been received + * @return the exit value, 1 if the process was killed, or Project.INVALID if no exit value has been received */ public int getExitValue() { return exitValue; @@ -355,25 +343,19 @@ public class Execute { return execute(envVars, v, baseDir); } - public static int execute(Vector envVars, Vector cmd, - File baseDir) { + public static int execute(Vector envVars, Vector cmd, File baseDir) { return execute(envVars, cmd, baseDir, 10000 /* default time to wait */); } /** * Wrapper for common execution patterns * - * @param envVars - * Environment variables to execute with (optional) - * @param cmd - * a vector of the commands to execute - * @param baseDir - * the base directory to run from (optional) - * @param timeToWait - * milliseconds to wait for completion + * @param envVars Environment variables to execute with (optional) + * @param cmd a vector of the commands to execute + * @param baseDir the base directory to run from (optional) + * @param timeToWait milliseconds to wait for completion */ - public static int execute(Vector envVars, Vector cmd, - File baseDir, int timeToWait) { + public static int execute(Vector envVars, Vector cmd, File baseDir, int timeToWait) { try { // We can collect the out or provide in if needed ExecuteWatchdog watchdog = new ExecuteWatchdog(timeToWait); @@ -407,13 +389,9 @@ public class Execute { log.debug("Exit value " + status); return status; } catch (Exception ex) { - // ex.printStackTrace(); System.err.println("An error has occurred in Execute."); return -1; } } - private static org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory - .getLog(Execute.class); - } diff --git a/spring-erlang/src/main/java/org/springframework/util/exec/ExecuteWatchdog.java b/spring-erlang/src/main/java/org/springframework/util/exec/ExecuteWatchdog.java index 16f25723..014b52a9 100644 --- a/spring-erlang/src/main/java/org/springframework/util/exec/ExecuteWatchdog.java +++ b/spring-erlang/src/main/java/org/springframework/util/exec/ExecuteWatchdog.java @@ -103,7 +103,7 @@ public class ExecuteWatchdog implements Runnable { /** - * Watches the process and terminates it, if it runs for to long. + * Watches the process and terminates it, if it runs for too long. */ public synchronized void run() { try { @@ -117,7 +117,7 @@ public class ExecuteWatchdog implements Runnable { } catch (InterruptedException e) {} } - // if we are here, either someone stopped the watchdog, + // If we are here, either someone stopped the watchdog, // we are on timeout and the process must be killed, or // we are on timeout and the process has already stopped. try { @@ -125,13 +125,11 @@ public class ExecuteWatchdog implements Runnable { // before being here process.exitValue(); } catch (IllegalThreadStateException e){ - // the process is not terminated, if this is really + // The process is not terminated, if this is really // a timeout and not a manual stop then kill it. - //System.out.println("ExecuteWatchdog: timeout"); if (watch){ killedProcess = true; if( ! dontkill ) { - //System.out.println("ExecuteWatchdog: destroying process"); process.destroy(); } if( execThread != null ) { @@ -144,7 +142,6 @@ public class ExecuteWatchdog implements Runnable { } finally { cleanUp(); } - //System.out.println("ExecuteWatchdog: done"); } diff --git a/spring-rabbit-admin/.classpath b/spring-rabbit-admin/.classpath deleted file mode 100644 index 3bde1346..00000000 --- a/spring-rabbit-admin/.classpath +++ /dev/null @@ -1,9 +0,0 @@ - - - - - - - - - diff --git a/spring-rabbit-admin/.project b/spring-rabbit-admin/.project deleted file mode 100644 index df9ae64f..00000000 --- a/spring-rabbit-admin/.project +++ /dev/null @@ -1,35 +0,0 @@ - - - spring-rabbit-admin - - - - - - org.eclipse.wst.common.project.facet.core.builder - - - - - org.eclipse.jdt.core.javabuilder - - - - - org.maven.ide.eclipse.maven2Builder - - - - - org.springframework.ide.eclipse.core.springbuilder - - - - - - org.springframework.ide.eclipse.core.springnature - org.maven.ide.eclipse.maven2Nature - org.eclipse.jdt.core.javanature - org.eclipse.wst.common.project.facet.core.nature - - diff --git a/spring-rabbit-admin/.settings/org.maven.ide.eclipse.prefs b/spring-rabbit-admin/.settings/org.maven.ide.eclipse.prefs deleted file mode 100644 index 3b616447..00000000 --- a/spring-rabbit-admin/.settings/org.maven.ide.eclipse.prefs +++ /dev/null @@ -1,9 +0,0 @@ -#Tue Jun 29 21:59:01 EDT 2010 -activeProfiles= -eclipse.preferences.version=1 -fullBuildGoals=process-test-resources -includeModules=false -resolveWorkspaceProjects=true -resourceFilterGoals=process-resources resources\:testResources -skipCompilerPlugin=true -version=1 diff --git a/spring-rabbit-admin/.springBeans b/spring-rabbit-admin/.springBeans deleted file mode 100644 index dbe284d5..00000000 --- a/spring-rabbit-admin/.springBeans +++ /dev/null @@ -1,13 +0,0 @@ - - - 1 - - - - - - - - - - diff --git a/spring-rabbit-admin/pom.xml b/spring-rabbit-admin/pom.xml deleted file mode 100644 index 6eb4e22e..00000000 --- a/spring-rabbit-admin/pom.xml +++ /dev/null @@ -1,127 +0,0 @@ - - 4.0.0 - - org.springframework.amqp - spring-amqp-parent - 1.0.0.BUILD-SNAPSHOT - ../spring-amqp-parent/pom.xml - - spring-rabbit-admin - jar - Spring RabbitMQ Administrative Support - - - com.rabbitmq - amqp-client - - - org.springframework.amqp - spring-amqp - - - org.springframework.amqp - spring-rabbit - - - org.springframework.amqp - spring-erlang - - - - - org.springframework - spring-aop - - - org.springframework - spring-oxm - - - org.springframework - spring-tx - - - org.springframework - spring-test - compile - - - - - commons-io - commons-io - - - - - org.slf4j - slf4j-api - - - org.slf4j - jcl-over-slf4j - compile - - - org.slf4j - slf4j-log4j12 - runtime - - - log4j - log4j - - - javax.mail - mail - - - javax.jms - jms - - - com.sun.jdmk - jmxtools - - - com.sun.jmx - jmxri - - - runtime - - - - javax.annotation - jsr250-api - true - - - - cglib - cglib-nodep - true - - - - org.mockito - mockito-all - test - - - - junit - junit - compile - - - - - - - com.springsource.bundlor - com.springsource.bundlor.maven - - - - diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitConfiguration.java b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitConfiguration.java deleted file mode 100644 index fa5261bf..00000000 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitConfiguration.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.amqp.rabbit.test; - -import java.lang.annotation.Documented; -import java.lang.annotation.ElementType; -import java.lang.annotation.Inherited; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -/** - * RabbitConfiguration defines class-level metadata for configuring Rabbit integration tests. - * - * @author Mark Pollack - * - */ -@Documented -@Inherited -@Retention(RetentionPolicy.RUNTIME) -@Target(ElementType.TYPE) -public @interface RabbitConfiguration { - - String hostname() default "localhost"; - - String username() default "guest"; - - String password() default "guest"; -} diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java deleted file mode 100644 index a32a3a03..00000000 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java +++ /dev/null @@ -1,136 +0,0 @@ -/* - * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.amqp.rabbit.test; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.amqp.rabbit.admin.RabbitBrokerAdmin; -import org.springframework.amqp.rabbit.admin.RabbitStatus; -import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; -import org.springframework.core.annotation.AnnotationUtils; -import org.springframework.erlang.OtpIOException; -import org.springframework.test.context.TestContext; -import org.springframework.test.context.support.AbstractTestExecutionListener; -import org.springframework.test.context.transaction.TransactionConfiguration; -import org.springframework.test.context.transaction.TransactionConfigurationAttributes; - -/** - * A TestExeuctionListener that will start/reset the RabbitMQ node before the methods in the - * TestClass are executed. If the node was started before the test methods were executed, - * the node will be stopped after the test methods are executed . - * - * @author Mark Pollack - */ -public class RabbitTestExecutionListener extends AbstractTestExecutionListener{ - - private static final Log logger = LogFactory.getLog(RabbitTestExecutionListener.class); - - private RabbitBrokerAdmin rabbitAdminTemplate; - - private boolean startedNode; - - - @Override - public void beforeTestClass(TestContext testContext) throws Exception { - initializeRabbitAdminTemplate(testContext); - recycleBrokerApp(); - } - - @Override - public void afterTestClass(TestContext testContext) throws Exception { - if (startedNode) { - rabbitAdminTemplate.stopNode(); - startedNode = false; - } - } - - private void recycleBrokerApp() { - try { - RabbitStatus status = rabbitAdminTemplate.getStatus(); - logger.debug(status); - rabbitAdminTemplate.stopBrokerApplication(); - rabbitAdminTemplate.resetNode(); - rabbitAdminTemplate.startBrokerApplication(); - } - catch (OtpIOException e) { - // Can't connect because broker node isn't running. - rabbitAdminTemplate.startNode(); - startedNode = true; - //TODO - need to wait on output for 'broker running' - try { - Thread.sleep(2000); - } - catch (InterruptedException e1) { - logger.error("Error waiting for broker to start"); - } - } - } - - /** - * Retrieves the {@link TransactionConfigurationAttributes} for the - * specified {@link Class class} which may optionally declare or inherit a - * {@link TransactionConfiguration @TransactionConfiguration}. If a - * {@link TransactionConfiguration} annotation is not present for the - * supplied class, the default values for attributes defined in - * {@link TransactionConfiguration} will be used instead. - * @param clazz the Class object corresponding to the test class for which - * the configuration attributes should be retrieved - */ - private void initializeRabbitAdminTemplate(TestContext testContext) { - - //Could potentially get the information from the BeanFactory by implementing a BeanFactoryPostProcessor - if (this.rabbitAdminTemplate == null) { - Class clazz = testContext.getTestClass(); - Class annotationType = RabbitConfiguration.class; - RabbitConfiguration config = clazz.getAnnotation(annotationType); - if (logger.isDebugEnabled()) { - logger.debug("Retrieved @RabbitConfiguration [" + config + "] for test class [" + clazz + "]"); - } - String hostname; - String username; - String password; - if (config != null) { - hostname = config.hostname(); - username = config.username(); - password = config.password(); - } - else { - hostname = (String) AnnotationUtils.getDefaultValue(annotationType, "hostname"); - username = (String) AnnotationUtils.getDefaultValue(annotationType, "username"); - password = (String) AnnotationUtils.getDefaultValue(annotationType, "password"); - } - if (logger.isDebugEnabled()) { - logger.debug( - String.format("Retrieved hostname=[%s] username=[%s], password=[%s] for class [%s]", - hostname, username, password, clazz)); - } - SingleConnectionFactory connectionFactory; - if (hostname.equals("localhost")) { - //This will try to get the local host name - connectionFactory = new SingleConnectionFactory(); - } - else { - connectionFactory = new SingleConnectionFactory(hostname); - } - connectionFactory.setUsername(username); - connectionFactory.setPassword(password); - rabbitAdminTemplate = new RabbitBrokerAdmin(connectionFactory); - } - } - -} diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/SpringRabbitJUnit4ClassRunner.java b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/SpringRabbitJUnit4ClassRunner.java deleted file mode 100644 index b2770ab1..00000000 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/SpringRabbitJUnit4ClassRunner.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.amqp.rabbit.test; - -import java.lang.reflect.Field; -import java.util.ArrayList; - -import org.junit.runners.model.InitializationError; -import org.springframework.test.context.TestContextManager; -import org.springframework.test.context.TestExecutionListener; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.util.ReflectionUtils; - -/** - * A specialized JUnit4 class runner so that we can add a RabbitTestExecutionListener. - * - * The RabbitTestExecutionListener needs to run before the DependencyInjectionTestExecutionListener - * so that the RabbitMQ broker application and be 'reset' or the RabbitMQ node - * started before the application context is created so that RabbitMQ broker configuration can - * execute when the applicatin context is created. - * - * - * @author Mark Pollack - * - */ -public class SpringRabbitJUnit4ClassRunner extends SpringJUnit4ClassRunner { - - public SpringRabbitJUnit4ClassRunner(Class clazz) - throws InitializationError { - super(clazz); - } - - - /** - * Creates a new {@link TestContextManager} for the supplied test class and - * the configured default ContextLoader class name. - * Can be overridden by subclasses. - * - * @param clazz the test class to be managed - * @see #getDefaultContextLoaderClassName(Class) - */ - protected TestContextManager createTestContextManager(Class clazz) { - TestContextManager mgr = new TestContextManager(clazz, getDefaultContextLoaderClassName(clazz)); - - Field executionListenersField = ReflectionUtils.findField(TestContextManager.class, "testExecutionListeners"); - executionListenersField.setAccessible(true); - @SuppressWarnings("unchecked") - ArrayList executionListeners = - (ArrayList) ReflectionUtils.getField(executionListenersField, mgr); - ArrayList newExecutionListeners = new ArrayList(); - newExecutionListeners.add(new RabbitTestExecutionListener()); - for (TestExecutionListener testExecutionListener : executionListeners) { - newExecutionListeners.add(testExecutionListener); - } - ReflectionUtils.setField(executionListenersField, mgr, newExecutionListeners); - - //puts to the back of the list - need it before context is created as broker configuration instructions are - //executed as part of the context creation. - //mgr.registerTestExecutionListeners(new RabbitTestExecutionListener()); - - return mgr; - } - - -} diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListenerIntegrationTests.java b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListenerIntegrationTests.java deleted file mode 100644 index d231ded9..00000000 --- a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListenerIntegrationTests.java +++ /dev/null @@ -1,27 +0,0 @@ -package org.springframework.amqp.rabbit.test; - -import org.junit.Ignore; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.springframework.test.context.ContextConfiguration; - -@RunWith(SpringRabbitJUnit4ClassRunner.class) -@ContextConfiguration -@RabbitConfiguration -@Ignore // only works on Windows at the moment -public class RabbitTestExecutionListenerIntegrationTests { - - @Test - public void doNothing() throws InterruptedException { - Thread.sleep(1000); - System.out.println("inside DO NOTHING"); - System.out.println("inside DO NOTHING"); - } - - @Test - public void doNothingAgain() throws InterruptedException { - Thread.sleep(1000); - System.out.println("inside DO AGAIN"); - System.out.println("inside DO AGAIN"); - } -} diff --git a/spring-rabbit-admin/src/test/resources/log4j.properties b/spring-rabbit-admin/src/test/resources/log4j.properties deleted file mode 100644 index 6ddfb097..00000000 --- a/spring-rabbit-admin/src/test/resources/log4j.properties +++ /dev/null @@ -1,9 +0,0 @@ -log4j.rootCategory=INFO, stdout - -log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n - -log4j.category.org.springframework.amqp=DEBUG -log4j.category.org.springframework.beans.factory=INFO - diff --git a/spring-rabbit-admin/src/test/resources/org/springframework/amqp/rabbit/test/ConfigureBrokerAnntotationTests-context.xml b/spring-rabbit-admin/src/test/resources/org/springframework/amqp/rabbit/test/ConfigureBrokerAnntotationTests-context.xml deleted file mode 100644 index 50181ac6..00000000 --- a/spring-rabbit-admin/src/test/resources/org/springframework/amqp/rabbit/test/ConfigureBrokerAnntotationTests-context.xml +++ /dev/null @@ -1,18 +0,0 @@ - - - - - - diff --git a/spring-rabbit-admin/template.mf b/spring-rabbit-admin/template.mf deleted file mode 100644 index 1c320dcc..00000000 --- a/spring-rabbit-admin/template.mf +++ /dev/null @@ -1,13 +0,0 @@ -Bundle-SymbolicName: org.springframework.amqp.rabbit.admin -Bundle-Name: Spring RabbitMQ Administrative Support -Bundle-Vendor: SpringSource -Bundle-ManifestVersion: 2 -Import-Template: - org.springframework.*;version="[3.0.5, 4.0.0)", - org.springframework.amqp.*;version="[1.0.0, 1.1.0)", - org.apache.commons.logging;version="[1.1.1, 2.0.0)", - com.rabbitmq.*;version="[1.8.1, 2.0.0)", - com.ericsson.otp.erlang;version="[1.5.3, 2.0.0)", - org.junit.*;version="0", - org.w3c.dom.*;version="0", - javax.xml.*;version="0" diff --git a/spring-rabbit/pom.xml b/spring-rabbit/pom.xml index ab8232eb..e961025e 100644 --- a/spring-rabbit/pom.xml +++ b/spring-rabbit/pom.xml @@ -19,6 +19,11 @@ org.springframework.amqp spring-amqp + + org.springframework.amqp + spring-erlang + true + diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/QueueInfo.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/QueueInfo.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/QueueInfo.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/QueueInfo.java diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminAuthException.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminAuthException.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminAuthException.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminAuthException.java diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java old mode 100644 new mode 100755 similarity index 50% rename from spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java index d94ff3f9..edf4d237 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java @@ -15,43 +15,38 @@ package org.springframework.amqp.rabbit.admin; import java.io.File; import java.io.FilenameFilter; +import java.io.UnsupportedEncodingException; +import java.net.InetAddress; +import java.net.UnknownHostException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.HashMap; import java.util.List; +import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.regex.Pattern; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.amqp.core.Binding; -import org.springframework.amqp.core.Exchange; -import org.springframework.amqp.core.Queue; -import org.springframework.amqp.rabbit.connection.ConnectionFactory; -import org.springframework.amqp.rabbit.core.ChannelCallback; -import org.springframework.amqp.rabbit.core.RabbitAdmin; -import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.erlang.OtpAuthException; -import org.springframework.erlang.OtpIOException; -import org.springframework.erlang.connection.SingleConnectionFactory; +import org.springframework.erlang.OtpException; +import org.springframework.erlang.connection.ConnectionFactory; +import org.springframework.erlang.connection.SimpleConnectionFactory; import org.springframework.erlang.core.Application; import org.springframework.erlang.core.ErlangTemplate; import org.springframework.erlang.core.Node; import org.springframework.jmx.export.annotation.ManagedOperation; -import org.springframework.jmx.export.annotation.ManagedOperationParameter; -import org.springframework.jmx.export.annotation.ManagedOperationParameters; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.util.exec.Execute; import org.springframework.util.exec.Os; -import com.rabbitmq.client.AMQP.Exchange.DeleteOk; -import com.rabbitmq.client.Channel; - /** * Rabbit broker administration implementation exposed via JMX annotations. * @@ -59,135 +54,97 @@ import com.rabbitmq.client.Channel; */ public class RabbitBrokerAdmin implements RabbitBrokerOperations { + private static final String DEFAULT_VHOST = "/"; + + private static String DEFAULT_HOST; + + private static final String DEFAULT_ENCODING = "UTF-8"; + /** Logger available to subclasses */ protected final Log logger = LogFactory.getLog(getClass()); - private RabbitTemplate rabbitTemplate; - - private RabbitAdmin rabbitAdmin; - private ErlangTemplate erlangTemplate; - private String virtualHost; + private String encoding = DEFAULT_ENCODING; - public RabbitBrokerAdmin(ConnectionFactory connectionFactory) { - this.virtualHost = connectionFactory.getVirtualHost(); - this.rabbitTemplate = new RabbitTemplate(connectionFactory); - this.rabbitAdmin = new RabbitAdmin(connectionFactory); - initializeDefaultErlangTemplate(rabbitTemplate); + private long timeout = 0; + + // TODO: extract into field for DI + private SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); + + private final String hostName; + + // TODO: RABBITMQ_NODE_PORT=5672 + + static { + try { + DEFAULT_HOST = InetAddress.getLocalHost().getHostName(); + } catch (UnknownHostException e) { + DEFAULT_HOST = "localhost"; + } } - // Exchange Operations + public RabbitBrokerAdmin() { + this(DEFAULT_HOST); + } - public void declareExchange(Exchange exchange) { - rabbitAdmin.declareExchange(exchange); + public RabbitBrokerAdmin(String hostName) { + if (Os.isFamily("windows") && !DEFAULT_HOST.equals(hostName)) { + hostName = hostName.toUpperCase(); + } + this.hostName = hostName; + this.executor.setDaemon(true); + initializeDefaultErlangTemplate(hostName); } /** - * Declare an exchange specifying its durability and auto-delete behavior. Explicit arguments are given so as to - * make this method easily accessible from JMX management consoles. Durable exchanges last until they are deleted, - * they will survive a server restart. Auto-deleted exchanges last until they are no longer used - * - * @param exchangeName the name of the exchange - * @param exchangeType the exchange type - * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart) - * @param autoDelete true if the server should delete the exchange when it is no longer in use + * The encoding to use for converting host names to byte arrays (which is needed on the remote side). + * @param encoding the encoding to use (default UTF-8) */ - @ManagedOperation - public void declareExchange(final String exchangeName, final String exchangeType, final boolean durable, - final boolean autoDelete) { - rabbitTemplate.execute(new ChannelCallback() { - public Object doInRabbit(Channel channel) throws Exception { - channel.exchangeDeclare(exchangeName, exchangeType, durable, autoDelete, new HashMap()); - return null; - } - }); + public void setEncoding(String encoding) { + this.encoding = encoding; } - @ManagedOperation(description = "Delete a exchange, without regard for whether it is in use or has messages on it") - @ManagedOperationParameters(@ManagedOperationParameter(name = "exchange", description = "the name of the exchange")) - public void deleteExchange(String exchangeName) { - rabbitAdmin.deleteExchange(exchangeName); - - } - - @ManagedOperation - public DeleteOk deleteExchange(final String exchangeName, final boolean ifUnused) { - return rabbitTemplate.execute(new ChannelCallback() { - public DeleteOk doInRabbit(Channel channel) throws Exception { - channel.exchangeDelete(exchangeName, ifUnused); - return null; - } - }); - } - - // Queue Operations - - @ManagedOperation - public Queue declareQueue() { - return rabbitAdmin.declareQueue(); - } - - @ManagedOperation - public void declareQueue(Queue queue) { - rabbitAdmin.declareQueue(queue); - } - - @ManagedOperation - public void deleteQueue(String queueName) { - rabbitAdmin.deleteQueue(queueName); - } - - @ManagedOperation - public void deleteQueue(String queueName, boolean unused, boolean empty) { - rabbitAdmin.deleteQueue(queueName, unused, empty); - - } - - @ManagedOperation - public void purgeQueue(String queueName, boolean noWait) { - rabbitAdmin.purgeQueue(queueName, noWait); - + /** + * Timeout (milliseconds) to wait for the broker to come up. If the provided timeout is greater than zero then we + * wait for that period for the broker to be ready. If it is not ready after that time the process is stopped. + * Defaults to 0 (no wait). + * + * @param timeout the timeout value to set in milliseconds + */ + public void setStartupTimeout(long timeout) { + this.timeout = timeout; } @SuppressWarnings("unchecked") public List getQueues() { return (List) erlangTemplate.executeAndConvertRpc("rabbit_amqqueue", "info_all", - virtualHost.getBytes()); + getBytes(DEFAULT_VHOST)); } - // Binding operations - public void declareBinding(Binding binding) { - rabbitAdmin.declareBinding(binding); - } - - public void removeBinding(final Binding binding) { - rabbitTemplate.execute(new ChannelCallback() { - public Object doInRabbit(Channel channel) throws Exception { - channel.queueUnbind(binding.getQueue(), binding.getExchange(), binding.getRoutingKey(), - binding.getArguments()); - return null; - } - }); + @SuppressWarnings("unchecked") + public List getQueues(String virtualHost) { + return (List) erlangTemplate.executeAndConvertRpc("rabbit_amqqueue", "info_all", + getBytes(virtualHost)); } // User management @ManagedOperation() public void addUser(String username, String password) { - erlangTemplate.executeAndConvertRpc("rabbit_access_control", "add_user", username.getBytes(), - password.getBytes()); + erlangTemplate + .executeAndConvertRpc("rabbit_access_control", "add_user", getBytes(username), getBytes(password)); } @ManagedOperation public void deleteUser(String username) { - erlangTemplate.executeAndConvertRpc("rabbit_access_control", "delete_user", username.getBytes()); + erlangTemplate.executeAndConvertRpc("rabbit_access_control", "delete_user", getBytes(username)); } @ManagedOperation public void changeUserPassword(String username, String newPassword) { - erlangTemplate.executeAndConvertRpc("rabbit_access_control", "change_password", username.getBytes(), - newPassword.getBytes()); + erlangTemplate.executeAndConvertRpc("rabbit_access_control", "change_password", getBytes(username), + getBytes(newPassword)); } @SuppressWarnings("unchecked") @@ -239,38 +196,85 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { @ManagedOperation public void startBrokerApplication() { - logger.debug("Starting Rabbit Application."); - erlangTemplate.executeAndConvertRpc("rabbit", "start"); + RabbitStatus status = getStatus(); + if (status.isReady()) { + logger.info("Rabbit Application already running."); + return; + } + if (!status.isAlive()) { + logger.info("Rabbit Process not running."); + startNode(); + return; + } + logger.info("Starting Rabbit Application."); + + // This call in particular seems to be prone to hanging, so do it in the background... + final CountDownLatch latch = new CountDownLatch(1); + Future result = executor.submit(new Callable() { + public Object call() throws Exception { + try { + return erlangTemplate.executeAndConvertRpc("rabbit", "start"); + } finally { + latch.countDown(); + } + } + }); + boolean started = false; + try { + started = latch.await(timeout, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + result.cancel(true); + return; + } + if (timeout > 0 && started) { + if (!waitForReadyState() && !result.isDone()) { + result.cancel(true); + } + } } @ManagedOperation public void stopBrokerApplication() { - logger.debug("Stopping Rabbit Application."); + logger.info("Stopping Rabbit Application."); erlangTemplate.executeAndConvertRpc("rabbit", "stop"); + if (timeout > 0) { + waitForUnreadyState(); + } } @ManagedOperation public void startNode() { - logger.debug("Starting RabbitMQ node by shelling out command line."); + RabbitStatus status = getStatus(); + if (status.isAlive()) { + logger.info("Rabbit Process already running."); + startBrokerApplication(); + return; + } + + if (!status.isRunning() && status.isReady()) { + logger.info("Rabbit Process not running but status is ready. Restarting."); + stopNode(); + } + + logger.info("Starting RabbitMQ node by shelling out command line."); final Execute execute = new Execute(); String rabbitStartScript = null; String hint = ""; if (Os.isFamily("windows") || Os.isFamily("dos")) { - rabbitStartScript = "rabbitmq-server.bat"; + rabbitStartScript = "sbin/rabbitmq-server.bat"; } else if (Os.isFamily("unix") || Os.isFamily("mac")) { - rabbitStartScript = "rabbitmq-server"; + rabbitStartScript = "bin/rabbitmq-server"; hint = "Depending on your platform it might help to set RABBITMQ_LOG_BASE and RABBITMQ_MNESIA_BASE System properties to an empty directory."; } Assert.notNull(rabbitStartScript, "unsupported OS family"); String rabbitHome = System.getProperty("RABBITMQ_HOME", System.getenv("RABBITMQ_HOME")); - String rabbitBin = "bin"; if (rabbitHome == null) { if (Os.isFamily("windows") || Os.isFamily("dos")) { rabbitHome = findDirectoryName("c:/Program Files", "rabbitmq"); - rabbitBin = "sbin"; } else if (Os.isFamily("unix") || Os.isFamily("mac")) { rabbitHome = "/usr/lib/rabbitmq"; } @@ -278,19 +282,23 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { Assert.notNull(rabbitHome, "RABBITMQ_HOME system property (or environment variable) not set."); rabbitHome = StringUtils.cleanPath(rabbitHome); - String rabbitStartCommand = rabbitHome + System.getProperty("file.separator") + rabbitBin - + System.getProperty("file.separator") + rabbitStartScript; + String rabbitStartCommand = rabbitHome + "/" + rabbitStartScript; + String[] commandline = new String[] { rabbitStartCommand }; List env = new ArrayList(); addEnvironment(env, "RABBITMQ_LOG_BASE"); addEnvironment(env, "RABBITMQ_MNESIA_BASE"); addEnvironment(env, "ERLANG_HOME"); - execute.setCommandline(new String[] { rabbitStartCommand }); - execute.setEnvironment(env.toArray(new String[0])); + // Make the hostname explicitly the same so the erl process knows who we are + env.add("HOSTNAME=" + hostName); - // TODO: extract into field for DI - SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); + // 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). + env.add("RABBITMQ_SERVER_ERL_ARGS=-detached"); + + execute.setCommandline(commandline); + execute.setEnvironment(env.toArray(new String[0])); final CountDownLatch running = new CountDownLatch(1); final AtomicBoolean finished = new AtomicBoolean(false); @@ -302,27 +310,113 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { running.countDown(); int exit = execute.execute(); finished.set(true); + logger.info("Finished broker launcher process with exit code=" + exit); if (exit != 0) { throw new IllegalStateException("Could not start process." + errorHint); } } catch (Exception e) { - logger.error("failed to start node", e); + logger.error("Failed to start node", e); } } }); try { logger.info("Waiting for Rabbit process to be started"); - Assert.state(running.await(1000L, TimeUnit.MILLISECONDS), "Timed out waiting for Rabbit process to start."); - Thread.sleep(100L); + Assert.state(running.await(1000L, TimeUnit.MILLISECONDS), + "Timed out waiting for thread to start Rabbit process."); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } if (finished.get()) { - throw new IllegalStateException("Expected broker process to start in background, but it has exited early."); + // throw new + // IllegalStateException("Expected broker process to start in background, but it has exited early."); } + if (timeout > 0) { + waitForReadyState(); + } + + } + + private boolean waitForReadyState() { + return waitForState(new StatusCallback() { + public boolean get(RabbitStatus status) { + return status.isReady(); + } + }, "ready"); + } + + private boolean waitForUnreadyState() { + return waitForState(new StatusCallback() { + public boolean get(RabbitStatus status) { + return !status.isRunning(); + } + }, "unready"); + } + + private boolean waitForStoppedState() { + return waitForState(new StatusCallback() { + public boolean get(RabbitStatus status) { + return !status.isReady() && !status.isRunning(); + } + }, "stopped"); + } + + private boolean waitForState(final StatusCallback callable, String state) { + + if (timeout <= 0) { + return true; + } + + RabbitStatus status = getStatus(); + + if (!callable.get(status)) { + + logger.info("Waiting for broker to enter state: " + state); + + Future started = executor.submit(new Callable() { + public RabbitStatus call() throws Exception { + RabbitStatus status = getStatus(); + while (!callable.get(status)) { + // Any less than 1000L and we tend to clog up the socket? + Thread.sleep(500L); + status = getStatus(); + } + return status; + } + }); + + try { + status = started.get(timeout, TimeUnit.MILLISECONDS); + // This seems to help... really it just means we didn't get the right status data + Thread.sleep(500L); + } catch (TimeoutException e) { + started.cancel(true); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (ExecutionException e) { + logger.error("Exception checking broker status for " + state, e.getCause()); + } + + if (!callable.get(status)) { + logger.error("Rabbit broker not in " + state + " state after timeout. Stopping process."); + stopNode(); + return false; + } else { + logger.info("Finished waiting for broker to enter state: " + state); + if (logger.isDebugEnabled()) { + logger.info("Status: " + status); + } + return true; + } + + } else { + logger.info("Broker already in state: " + state); + } + + return true; + } /** @@ -367,12 +461,15 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { @ManagedOperation public void stopNode() { - logger.debug("Stopping RabbitMQ node."); + logger.info("Stopping RabbitMQ node."); try { erlangTemplate.executeAndConvertRpc("rabbit", "stop_and_halt"); } catch (Exception e) { logger.error("Failed to send stop signal", e); } + if (timeout >= 0) { + waitForStoppedState(); + } } @ManagedOperation @@ -390,15 +487,18 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { public RabbitStatus getStatus() { try { return (RabbitStatus) getErlangTemplate().executeAndConvertRpc("rabbit", "status"); - } catch (OtpIOException e) { - logger.info("Ignoring OtpIOException (assuming that the broker is simply not running)"); - return new RabbitStatus(Collections. emptyList(), Collections. emptyList(), - Collections. emptyList()); } catch (OtpAuthException e) { throw new RabbitAdminAuthException( "Could not authorise connection to Erlang process. This can happen if the broker is running, " + "but as root or rabbitmq and the current user is not authorised to connect. Try starting the " + "broker again as a different user.", e); + } catch (OtpException e) { + logger.debug("Ignoring OtpException (assuming that the broker is simply not running)"); + if (logger.isTraceEnabled()) { + logger.trace("Status not available owing to exception", e); + } + return new RabbitStatus(Collections. emptyList(), Collections. emptyList(), + Collections. emptyList()); } } @@ -410,22 +510,41 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations { return this.erlangTemplate; } - protected void initializeDefaultErlangTemplate(RabbitTemplate rabbitTemplate) { - String host = rabbitTemplate.getConnectionFactory().getHost(); - if (Os.isFamily("windows")) { - host = host.toUpperCase(); - } + protected void initializeDefaultErlangTemplate(String host) { String peerNodeName = "rabbit@" + host; logger.debug("Creating jinterface connection with peerNodeName = [" + peerNodeName + "]"); - SingleConnectionFactory otpCf = new SingleConnectionFactory("rabbit-spring-monitor", peerNodeName); - otpCf.afterPropertiesSet(); - createErlangTemplate(otpCf); + SimpleConnectionFactory otpConnectionFactory = new SimpleConnectionFactory("rabbit-spring-monitor", + peerNodeName); + otpConnectionFactory.afterPropertiesSet(); + createErlangTemplate(otpConnectionFactory); } - protected void createErlangTemplate(org.springframework.erlang.connection.ConnectionFactory otpCf) { - erlangTemplate = new ErlangTemplate(otpCf); + protected void createErlangTemplate(ConnectionFactory otpConnectionFactory) { + erlangTemplate = new ErlangTemplate(otpConnectionFactory); erlangTemplate.setErlangConverter(new RabbitControlErlangConverter()); erlangTemplate.afterPropertiesSet(); } + /** + * Safely convert a string to its bytes using the encoding provided. + * + * @see #setEncoding(String) + * + * @param string the value to convert + * + * @return the bytes from the string using the encoding provided + * + * @throws IllegalStateException if the encoding is ont supported + */ + private byte[] getBytes(String string) { + try { + return string.getBytes(encoding); + } catch (UnsupportedEncodingException e) { + throw new IllegalStateException("Unsupported encoding: " + encoding); + } + } + + private static interface StatusCallback { + boolean get(RabbitStatus status); + } } diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java old mode 100644 new mode 100755 similarity index 80% rename from spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java index 95d6a111..23d37536 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java @@ -1,17 +1,14 @@ /* * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. */ package org.springframework.amqp.rabbit.admin; @@ -19,11 +16,6 @@ package org.springframework.amqp.rabbit.admin; import java.util.List; import java.util.regex.Pattern; -import org.springframework.amqp.core.AmqpAdmin; -import org.springframework.amqp.core.Binding; - -import com.rabbitmq.client.AMQP; - /** * Performs administration tasks for RabbitMQ broker administration.

Goal is to support full CRUD of Exchanges, * Queues, Bindings, User, VHosts, etc.

Current implementations expose operations with basic type arguments via JMX. @@ -31,18 +23,14 @@ import com.rabbitmq.client.AMQP; * @author Mark Pollack * */ -public interface RabbitBrokerOperations extends AmqpAdmin { - - // Exchange Operations - - AMQP.Exchange.DeleteOk deleteExchange(String exchangeName, boolean ifUnused); - - void removeBinding(Binding binding); +public interface RabbitBrokerOperations { // Queue operations public List getQueues(); + public List getQueues(String virtualHost); + // Message Delivery void recoverAsync(boolean requeue); @@ -83,7 +71,7 @@ public interface RabbitBrokerOperations extends AmqpAdmin { /** * Starts the RabbitMQ application on an already running node. This command is typically run after performing other - * management actions that required the RabbitMQ application to be stopped, e.g. reset. + * management actions that required the RabbitMQ application to be stopped, e.g. reset. */ void startBrokerApplication(); diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitControlErlangConverter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitControlErlangConverter.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitControlErlangConverter.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitControlErlangConverter.java diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java old mode 100644 new mode 100755 similarity index 74% rename from spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java rename to spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java index 4177644c..ec89fb35 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java @@ -16,6 +16,7 @@ package org.springframework.amqp.rabbit.admin; +import java.io.Serializable; import java.util.List; import org.springframework.erlang.core.Application; @@ -27,16 +28,15 @@ import org.springframework.erlang.core.Node; * @author Mark Pollack * */ -public class RabbitStatus { +@SuppressWarnings("serial") +public class RabbitStatus implements Serializable { private List runningApplications; private List nodes; private List runningNodes; - - - + public RabbitStatus(List runningApplications, List nodes, List runningNodes) { super(); @@ -45,6 +45,27 @@ public class RabbitStatus { this.runningNodes = runningNodes; } + /** + * @return true if the broker process is running but not necessarily the application + */ + public boolean isAlive() { + return !nodes.isEmpty(); + } + + /** + * @return true if the broker process is running + */ + public boolean isRunning() { + return !runningNodes.isEmpty(); + } + + /** + * @return true if the broker application is running + */ + public boolean isReady() { + return isRunning() && !runningApplications.isEmpty(); + } + public List getRunningApplications() { return runningApplications; } @@ -63,7 +84,4 @@ public class RabbitStatus { + ", runningNodes=" + runningNodes + ", nodes=" + nodes + "]"; } - - - } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java index e8c01367..9f4f7540 100755 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java @@ -15,7 +15,7 @@ public interface Connection { /** * Create a new channel, using an internally allocated channel number. - * @param transactional TODO + * @param transactional true if the channel should support transactions * @return a new channel descriptor, or null if none is available * @throws IOException if an I/O problem is encountered */ diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index de3d58e7..208e298a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -177,6 +177,16 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, SmartLif }); } + @ManagedOperation + public void removeBinding(final Binding binding) { + rabbitTemplate.execute(new ChannelCallback() { + public Object doInRabbit(Channel channel) throws Exception { + channel.queueUnbind(binding.getQueue(), binding.getExchange(), binding.getRoutingKey(), + binding.getArguments()); + return null; + } + }); + } // Lifecycle implementation diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java old mode 100644 new mode 100755 similarity index 65% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java index ddf9a523..983887be --- a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java @@ -11,85 +11,67 @@ import junit.framework.Assert; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Ignore; import org.junit.Test; import org.springframework.erlang.connection.SingleConnectionFactory; import org.springframework.erlang.core.ErlangTemplate; import org.springframework.util.exec.Os; -import com.ericsson.otp.erlang.OtpAuthException; import com.ericsson.otp.erlang.OtpConnection; import com.ericsson.otp.erlang.OtpErlangBinary; -import com.ericsson.otp.erlang.OtpErlangExit; import com.ericsson.otp.erlang.OtpErlangList; import com.ericsson.otp.erlang.OtpErlangObject; import com.ericsson.otp.erlang.OtpPeer; import com.ericsson.otp.erlang.OtpSelf; -// @Ignore("manual integration test only.") +@Ignore("Manual integration test only.") public class JInterfaceIntegrationTests { private static Log logger = LogFactory.getLog(JInterfaceIntegrationTests.class); private static int counter; - @Test - public void rawApi() { - OtpConnection connection = null; - try { - OtpSelf self = new OtpSelf("rabbit-monitor"); + private OtpConnection connection = null; - String hostName = "rabbit@" + getHostName(); - OtpPeer peer = new OtpPeer(hostName); - connection = self.connect(peer); - // connection.sendRPC("erlang","date", new OtpErlangList()); - // connection.sendRPC("rabbit_access_control", "list_vhosts", new - // OtpErlangList()); - OtpErlangObject[] objectArray = { new OtpErlangBinary("/".getBytes()) }; - - connection.sendRPC("rabbit_amqqueue", "info_all", new OtpErlangList(objectArray)); - - // connection.sendRPC("rabbit_amqqueue", "stat_all", new - // OtpErlangList()); - - OtpErlangObject received = connection.receiveRPC(); - System.out.println(received); - System.out.println(received.getClass()); - } catch (IOException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } catch (OtpAuthException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } catch (OtpErlangExit e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } finally { - if (connection != null) { - connection.close(); - } + @After + public void close() { + if (connection != null) { + connection.close(); } + } + + @Test + public void testRawApi() throws Exception { + + OtpSelf self = new OtpSelf("rabbit-monitor"); + + String hostName = "rabbit@" + getHostName(); + OtpPeer peer = new OtpPeer(hostName); + connection = self.connect(peer); + + OtpErlangObject[] objectArray = { new OtpErlangBinary("/".getBytes()) }; + + connection.sendRPC("rabbit_amqqueue", "info_all", new OtpErlangList(objectArray)); + + OtpErlangObject received = connection.receiveRPC(); + System.out.println(received); + System.out.println(received.getClass()); } @Test public void otpTemplate() throws UnknownHostException { + String selfNodeName = "rabbit-monitor"; String peerNodeName = "rabbit@" + getHostName(); - // String home = System.getProperty("user.home"); - // System.out.println("home = " + home); - // System.out.println("peerNodeName = " + peerNodeName); - SingleConnectionFactory cf = new SingleConnectionFactory(selfNodeName, peerNodeName); cf.afterPropertiesSet(); ErlangTemplate template = new ErlangTemplate(cf); template.afterPropertiesSet(); - // OtpErlangObject result = template.executeRpc("rabbit_amqqueue", "info_all", "/".getBytes()); - // System.out.println(result); - // System.out.println(result.getClass()); - long number = (Long) template.executeAndConvertRpc("erlang", "abs", -161803399); Assert.assertEquals(161803399, number); @@ -106,13 +88,14 @@ public class JInterfaceIntegrationTests { } @Test - public void rawOtpConnect() throws Exception { + public void testRawOtpConnect() throws Exception { createConnection(); } @Test public void stressTest() throws Exception { - // String cookie = readCookie(); + String cookie = readCookie(); + logger.info("Cookie: "+cookie); OtpConnection con = createConnection(); boolean recycleConnection = false; for (int i = 0; i < 100; i++) { @@ -137,7 +120,7 @@ public class JInterfaceIntegrationTests { throws Exception, UnknownHostException { con.sendRPC(module, function, new OtpErlangList()); OtpErlangObject response = con.receiveRPC(); - // System.out.println(module + " response received = " + response.toString()); + logger.debug(module + " response received = " + response.toString()); if (recycleConnection) { con.close(); con = createConnection(); diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/PojoHandler.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/PojoHandler.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/PojoHandler.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/PojoHandler.java diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java old mode 100644 new mode 100755 similarity index 51% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java index 16007766..63343fa5 --- a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java @@ -1,17 +1,14 @@ /* * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. */ package org.springframework.amqp.rabbit.admin; @@ -23,66 +20,94 @@ import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.log4j.Level; import org.junit.AfterClass; +import org.junit.Before; import org.junit.BeforeClass; +import org.junit.Rule; import org.junit.Test; import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.test.BrokerPanic; +import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; +import org.springframework.util.exec.Os; /** - * - * This test class assumes that you are already running the rabbitmq broker. * * @author Mark Pollack + * @author Dave Syer */ public class RabbitBrokerAdminIntegrationTests { - private static Log logger = LogFactory - .getLog(RabbitBrokerAdminIntegrationTests.class); + private static Log logger = LogFactory.getLog(RabbitBrokerAdminIntegrationTests.class); + + @Rule + public Log4jLevelAdjuster logLevel = new Log4jLevelAdjuster(Level.INFO, RabbitBrokerAdmin.class); + + /* + * Ensure broker dies if a test fails (otherwise the erl process has to be killed manually) + */ + @Rule + public BrokerPanic panic = new BrokerPanic(); private static RabbitBrokerAdmin brokerAdmin; - private static SingleConnectionFactory connectionFactory; + @Before + public void init() throws Exception { + panic.setBrokerAdmin(brokerAdmin); + } @BeforeClass - public static void setUp() throws Exception { - connectionFactory = new SingleConnectionFactory(); - connectionFactory.setUsername("guest"); - connectionFactory.setPassword("guest"); - brokerAdmin = new RabbitBrokerAdmin(connectionFactory); - RabbitStatus status = brokerAdmin.getStatus(); - if (status.getNodes().isEmpty()) { - brokerAdmin.startNode(); - Thread.sleep(1000L); - } else { - brokerAdmin.startBrokerApplication(); - } + 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.setStartupTimeout(10000L); + brokerAdmin.startNode(); } - + @AfterClass - public static void close() { - brokerAdmin.stopNode(); + public static void stop() throws Exception { + if (Os.isFamily("windows") || Os.isFamily("dos")) { + brokerAdmin.stopNode(); + } + System.clearProperty("RABBITMQ_LOG_BASE"); + System.clearProperty("RABBITMQ_MNESIA_BASE"); } @Test - // @Ignore public void integrationTestsUserCrud() throws Exception { List users = brokerAdmin.listUsers(); if (users.contains("joe")) { brokerAdmin.deleteUser("joe"); } - Thread.sleep(1000L); + Thread.sleep(200L); brokerAdmin.addUser("joe", "trader"); - Thread.sleep(1000L); + Thread.sleep(200L); brokerAdmin.changeUserPassword("joe", "sales"); - Thread.sleep(1000L); + Thread.sleep(200L); users = brokerAdmin.listUsers(); if (users.contains("joe")) { - Thread.sleep(1000L); + Thread.sleep(200L); brokerAdmin.deleteUser("joe"); } } + @Test + public void testStatusAndBrokerLifecycle() throws Exception { + + brokerAdmin.stopBrokerApplication(); + RabbitStatus status = brokerAdmin.getStatus(); + assertEquals(0, status.getRunningNodes().size()); + + brokerAdmin.startBrokerApplication(); + status = brokerAdmin.getStatus(); + assertBrokerAppRunning(status); + + } + @Test public void repeatLifecycle() throws Exception { for (int i = 1; i < 20; i++) { @@ -93,23 +118,10 @@ public class RabbitBrokerAdminIntegrationTests { } } - // @Test - public void testStatusAndBrokerLifecycle() throws Exception { - - RabbitStatus status = brokerAdmin.getStatus(); - - brokerAdmin.stopBrokerApplication(); - status = brokerAdmin.getStatus(); - assertEquals(0, status.getRunningNodes().size()); - - brokerAdmin.startBrokerApplication(); - status = brokerAdmin.getStatus(); - assertBrokerAppRunning(status); - } - @Test public void testGetQueues() throws Exception { - brokerAdmin.declareQueue(new Queue("test.queue")); + ConnectionFactory connectionFactory = new SingleConnectionFactory(); + new RabbitAdmin(connectionFactory).declareQueue(new Queue("test.queue")); assertEquals("/", connectionFactory.getVirtualHost()); List queues = brokerAdmin.getQueues(); assertEquals("test.queue", queues.get(0).getName()); diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java similarity index 51% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java index abf0ab37..32374c72 100755 --- a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java @@ -1,38 +1,41 @@ /* * Copyright 2002-2010 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. */ package org.springframework.amqp.rabbit.admin; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; 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; -import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; -import org.springframework.erlang.OtpIOException; +import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; +import org.springframework.erlang.OtpException; /** * @author Mark Pollack * @author Dave Syer */ public class RabbitBrokerAdminStopIntegrationTests { - + + @Rule + public Log4jLevelAdjuster logLevel = new Log4jLevelAdjuster(Level.INFO, RabbitBrokerAdmin.class); + @Before public void init() throws Exception { FileUtils.deleteDirectory(new File("target/rabbitmq")); @@ -50,51 +53,39 @@ public class RabbitBrokerAdminStopIntegrationTests { // @Ignore("NEEDS RABBITMQ_HOME to be set.") public void testStartNode() throws Exception { - RabbitBrokerAdmin brokerAdmin; - - SingleConnectionFactory connectionFactory; - - connectionFactory = new SingleConnectionFactory(); - connectionFactory.setUsername("guest"); - connectionFactory.setPassword("guest"); - brokerAdmin = new RabbitBrokerAdmin(connectionFactory); + final RabbitBrokerAdmin brokerAdmin = new RabbitBrokerAdmin(); + brokerAdmin.setStartupTimeout(10000L); RabbitStatus status = brokerAdmin.getStatus(); try { // Stop it if it is already running - if (status.getRunningApplications().size() > 0) { + if (status.isReady()) { brokerAdmin.stopBrokerApplication(); Thread.sleep(1000L); } - } catch (OtpIOException e) { + } catch (OtpException e) { // Not useful for test. } status = brokerAdmin.getStatus(); - if (status.getNodes().isEmpty()) { - brokerAdmin.startNode(); - } else { + if (!status.isRunning()) { brokerAdmin.startBrokerApplication(); } - Thread.sleep(1000L); status = brokerAdmin.getStatus(); - - assertFalse("Broker node did not start. Check logs for hints.", status - .getNodes().isEmpty()); - + try { - assertFalse("Broker node not running. Check logs for hints.", - status.getRunningNodes().isEmpty()); - assertFalse( - "Broker application not running. Check logs for hints.", - status.getRunningApplications().isEmpty()); - // assertEquals(1, 1); + assertFalse("Broker node did not start. Check logs for hints.", status.getNodes().isEmpty()); + assertTrue("Broker node not running. Check logs for hints.", status.isRunning()); + assertTrue("Broker application not running. Check logs for hints.", status.isReady()); + + Thread.sleep(1000L); brokerAdmin.stopBrokerApplication(); Thread.sleep(1000L); + } finally { brokerAdmin.stopNode(); - Thread.sleep(2000L); } + } } diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitConsumerConfiguration.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitConsumerConfiguration.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitConsumerConfiguration.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitConsumerConfiguration.java diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitProducerConfiguration.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitProducerConfiguration.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitProducerConfiguration.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitProducerConfiguration.java diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateConsumerExample.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateConsumerExample.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateConsumerExample.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateConsumerExample.java diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateProducerExample.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateProducerExample.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateProducerExample.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateProducerExample.java diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/TestConstants.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/TestConstants.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/TestConstants.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/TestConstants.java diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/TestRabbitConfiguration.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/TestRabbitConfiguration.java old mode 100644 new mode 100755 similarity index 100% rename from spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/TestRabbitConfiguration.java rename to spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/TestRabbitConfiguration.java diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerPanic.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerPanic.java new file mode 100755 index 00000000..823ca501 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerPanic.java @@ -0,0 +1,51 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package org.springframework.amqp.rabbit.test; + +import org.junit.rules.MethodRule; +import org.junit.runners.model.FrameworkMethod; +import org.junit.runners.model.Statement; +import org.springframework.amqp.rabbit.admin.RabbitBrokerAdmin; + +public class BrokerPanic implements MethodRule { + + private RabbitBrokerAdmin brokerAdmin; + + /** + * @param brokerAdmin the brokerAdmin to set + */ + public void setBrokerAdmin(RabbitBrokerAdmin brokerAdmin) { + this.brokerAdmin = brokerAdmin; + } + + public Statement apply(final Statement base, final FrameworkMethod method, Object target) { + return new Statement() { + @Override + public void evaluate() throws Throwable { + try { + base.evaluate(); + } catch (Throwable t) { + if (brokerAdmin != null) { + try { + brokerAdmin.stopNode(); + } catch (Throwable e) { + // don't hide original error (so ignored) + } + } + throw t; + } + } + }; + } + +}