Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-59: Merge admin project

Browse Source 
- Try to get access to error message when connect fails
- Set default hostname to OS value
- Add -detached
This commit is contained in:
Dave Syer
2011-01-07 12:50:47 +00:00
parent 5a137fdb1d
commit fa81e49809
41 changed files with 634 additions and 1003 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpAdmin.java
View File

@@ -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);
}

1
spring-amqp-core/src/test/resources/log4j.properties
View File

@@ -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

5
spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionParameters.java
View File

@@ -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;
}

112
spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java
View File

@@ -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.
*
* <p>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.
* <p>
* 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.
* </p>
* <p>
* 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.
* </p>
* <p>
* Provides a more traditional API to creating a connection to a remote erlang node than the JInterface API.
* </p>
* <p>
* 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.
* </p>
* <p>
* 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.
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
* 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.
* </p>
* @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());
}
}

5
spring-erlang/src/main/java/org/springframework/erlang/core/Application.java
View File

@@ -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;

4
spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java
View File

@@ -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
}

5
spring-erlang/src/main/java/org/springframework/erlang/core/Node.java
View File

@@ -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;

128
spring-erlang/src/main/java/org/springframework/util/exec/Execute.java
View File

@@ -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 <code>PumpStreamHandler</code> for
* stream handling.
* Creates a new execute object using <code>PumpStreamHandler</code> 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 <code>null</code> 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 <code>null</code> 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 <em>key=value</em>
* @param env array of Strings, each element of which has an environment variable settings in format
* <em>key=value</em>
*/
public void setEnvironment(String[] env) {
this.env = env;
@@ -219,13 +206,10 @@ public class Execute {
/**
* Sets the working directory of the process to execute.
*
* <p>
* 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.
* <p> 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 <code>INVALID</code>
* @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<String> envVars, Vector<String> cmd,
File baseDir) {
public static int execute(Vector<String> envVars, Vector<String> 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<String> envVars, Vector<String> cmd,
File baseDir, int timeToWait) {
public static int execute(Vector<String> envVars, Vector<String> 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);
}

9
spring-erlang/src/main/java/org/springframework/util/exec/ExecuteWatchdog.java
View File

@@ -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");
}

9
spring-rabbit-admin/.classpath
View File

@@ -1,9 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
<classpathentry kind="src" output="target/classes" path="src/main/java"/>
<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
<classpathentry kind="src" output="target/test-classes" path="src/test/resources"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.6"/>
<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/>
<classpathentry kind="output" path="target/classes"/>
</classpath>

35
spring-rabbit-admin/.project
View File

@@ -1,35 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<projectDescription>
<name>spring-rabbit-admin</name>
<comment></comment>
<projects>
</projects>
<buildSpec>
<buildCommand>
<name>org.eclipse.wst.common.project.facet.core.builder</name>
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.eclipse.jdt.core.javabuilder</name>
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.maven.ide.eclipse.maven2Builder</name>
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.springframework.ide.eclipse.core.springbuilder</name>
<arguments>
</arguments>
</buildCommand>
</buildSpec>
<natures>
<nature>org.springframework.ide.eclipse.core.springnature</nature>
<nature>org.maven.ide.eclipse.maven2Nature</nature>
<nature>org.eclipse.jdt.core.javanature</nature>
<nature>org.eclipse.wst.common.project.facet.core.nature</nature>
</natures>
</projectDescription>

9
spring-rabbit-admin/.settings/org.maven.ide.eclipse.prefs
View File

@@ -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

13
spring-rabbit-admin/.springBeans
View File

@@ -1,13 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.3.3.201005160334-M1]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[false]]></enableImports>
<configs>
</configs>
<configSets>
</configSets>
</beansProjectDescription>

127
spring-rabbit-admin/pom.xml
View File

@@ -1,127 +0,0 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-amqp-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<relativePath>../spring-amqp-parent/pom.xml</relativePath>
</parent>
<artifactId>spring-rabbit-admin</artifactId>
<packaging>jar</packaging>
<name>Spring RabbitMQ Administrative Support</name>
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-erlang</artifactId>
</dependency>
<!-- Spring -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-oxm</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-tx</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<scope>compile</scope>
</dependency>
<!-- Other -->
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>
<!-- Logging -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>jcl-over-slf4j</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<exclusions>
<exclusion>
<groupId>javax.mail</groupId>
<artifactId>mail</artifactId>
</exclusion>
<exclusion>
<groupId>javax.jms</groupId>
<artifactId>jms</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jdmk</groupId>
<artifactId>jmxtools</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jmx</groupId>
<artifactId>jmxri</artifactId>
</exclusion>
</exclusions>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>jsr250-api</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>cglib</groupId>
<artifactId>cglib-nodep</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>compile</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>com.springsource.bundlor</groupId>
<artifactId>com.springsource.bundlor.maven</artifactId>
</plugin>
</plugins>
</build>
</project>

42
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitConfiguration.java
View File

@@ -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";
}

136
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java
View File

@@ -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 <em>default values</em> 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<RabbitConfiguration> 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);
}
}
}

78
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/SpringRabbitJUnit4ClassRunner.java
View File

@@ -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 <em>default <code>ContextLoader</code> class name</em>.
* 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<TestExecutionListener> executionListeners =
(ArrayList<TestExecutionListener>) ReflectionUtils.getField(executionListenersField, mgr);
ArrayList<TestExecutionListener> newExecutionListeners = new ArrayList<TestExecutionListener>();
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;
}
}

27
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListenerIntegrationTests.java
View File

@@ -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");
}
}

9
spring-rabbit-admin/src/test/resources/log4j.properties
View File

@@ -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

18
spring-rabbit-admin/src/test/resources/org/springframework/amqp/rabbit/test/ConfigureBrokerAnntotationTests-context.xml
View File

@@ -1,18 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
<!--
<bean id="rabbitAdminTemplate" class="org.springframework.amqp.rabbit.core.RabbitAdminTemplate">
<constructor-arg ref="connectionFactory"/>
</bean>
<bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.SingleConnectionFactory">
<property name="password" value="guest"/>
<property name="username" value="guest"/>
</bean>
-->
</beans>

13
spring-rabbit-admin/template.mf
View File

@@ -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"

5
spring-rabbit/pom.xml
View File

@@ -19,6 +19,11 @@
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-erlang</artifactId>
<optional>true</optional>
</dependency>
<!-- Spring -->
<dependency>

0
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/QueueInfo.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/QueueInfo.java Normal file → Executable file
View File

0
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminAuthException.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminAuthException.java Normal file → Executable file
View File

401
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java Normal file → Executable file
View File

@@ -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<Object>() {
public Object doInRabbit(Channel channel) throws Exception {
channel.exchangeDeclare(exchangeName, exchangeType, durable, autoDelete, new HashMap<String, Object>());
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<DeleteOk>() {
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<QueueInfo> getQueues() {
return (List<QueueInfo>) 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<Object>() {
public Object doInRabbit(Channel channel) throws Exception {
channel.queueUnbind(binding.getQueue(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
return null;
}
});
@SuppressWarnings("unchecked")
public List<QueueInfo> getQueues(String virtualHost) {
return (List<QueueInfo>) 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<Object> result = executor.submit(new Callable<Object>() {
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<String> env = new ArrayList<String>();
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<RabbitStatus> started = executor.submit(new Callable<RabbitStatus>() {
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.<Application> emptyList(), Collections.<Node> emptyList(),
Collections.<Node> 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.<Application> emptyList(), Collections.<Node> emptyList(),
Collections.<Node> 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);
}
}

38
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java Normal file → Executable file
View File

@@ -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. <p>Goal is to support full CRUD of Exchanges,
* Queues, Bindings, User, VHosts, etc. <p>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<QueueInfo> getQueues();
public List<QueueInfo> 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();

0
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitControlErlangConverter.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitControlErlangConverter.java Normal file → Executable file
View File

32
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java → spring-rabbit/src/main/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java Normal file → Executable file
View File

@@ -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<Application> runningApplications;
private List<Node> nodes;
private List<Node> runningNodes;
public RabbitStatus(List<Application> runningApplications,
List<Node> nodes, List<Node> 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<Application> getRunningApplications() {
return runningApplications;
}
@@ -63,7 +84,4 @@ public class RabbitStatus {
+ ", runningNodes=" + runningNodes + ", nodes=" + nodes + "]";
}
}

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java
View File

@@ -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
*/

10
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

@@ -177,6 +177,16 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, SmartLif
});
}
@ManagedOperation
public void removeBinding(final Binding binding) {
rabbitTemplate.execute(new ChannelCallback<Object>() {
public Object doInRabbit(Channel channel) throws Exception {
channel.queueUnbind(binding.getQueue(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
return null;
}
});
}
// Lifecycle implementation

79
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java Normal file → Executable file
View File

@@ -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();

0
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/PojoHandler.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/PojoHandler.java Normal file → Executable file
View File

116
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java Normal file → Executable file
View File

@@ -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<String> 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<QueueInfo> queues = brokerAdmin.getQueues();
assertEquals("test.queue", queues.get(0).getName());

71
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminStopIntegrationTests.java
View File

@@ -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);
}
}
}

0
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitConsumerConfiguration.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitConsumerConfiguration.java Normal file → Executable file
View File

0
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitProducerConfiguration.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitProducerConfiguration.java Normal file → Executable file
View File

0
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateConsumerExample.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateConsumerExample.java Normal file → Executable file
View File

0
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateProducerExample.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitTemplateProducerExample.java Normal file → Executable file
View File

0
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/TestConstants.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/TestConstants.java Normal file → Executable file
View File

0
spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/TestRabbitConfiguration.java → spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/TestRabbitConfiguration.java Normal file → Executable file
View File

51
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerPanic.java Executable file
View File

@@ -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;
}
}
};
}
}