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

AMQP-495: Remove spring-erlang

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-495
This commit is contained in:
Gary Russell
2015-05-06 16:15:32 +01:00
parent cdcc95b2d0
commit 23e6dc2ad6
53 changed files with 10 additions and 5167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
build.gradle
View File

@@ -75,7 +75,6 @@ subprojects { subproject ->
ext {
cglibVersion = '3.1'
commonsIoVersion = '2.4'
erlangOtpVersion = '1.5.6'
hamcrestVersion = '1.3'
jacksonVersion = '1.9.13'
jackson2Version = '2.3.2'
@@ -190,26 +189,11 @@ project('spring-amqp') {
}
project('spring-erlang') {
description = 'Spring Erlang Support'
dependencies {
compile "org.springframework:spring-beans:$springVersion"
compile "commons-io:commons-io:$commonsIoVersion"
compile ("javax.annotation:jsr250-api:1.0", optional)
compile "org.erlang.otp:jinterface:$erlangOtpVersion"
}
}
project('spring-rabbit') {
description = 'Spring RabbitMQ Support'
dependencies {
compile project(":spring-amqp")
testCompile project(":spring-erlang")
compile "com.rabbitmq:amqp-client:$rabbitmqVersion"
compile ("com.rabbitmq:http-client:$rabbitmqHttpClientVersion", optional)

1
settings.gradle
View File

@@ -1,6 +1,5 @@
rootProject.name = 'spring-amqp-dist'
include 'spring-amqp'
include 'spring-erlang'
include 'spring-rabbit'

49
spring-erlang/src/main/java/org/springframework/erlang/ErlangBadRpcException.java
View File

@@ -1,49 +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.erlang;
import com.ericsson.otp.erlang.OtpErlangTuple;
/**
* Exception thrown when an 'badrpc' is received from an Erlang RPC call.
* @author Mark Pollack
*
*/
@SuppressWarnings("serial")
public class ErlangBadRpcException extends OtpException {
private OtpErlangTuple reasonTuple;
public ErlangBadRpcException(String reason) {
super(reason);
}
public ErlangBadRpcException(OtpErlangTuple tuple) {
super(tuple.toString());
this.reasonTuple = tuple;
}
public OtpErlangTuple getReasonTuple() {
return reasonTuple;
}
}

44
spring-erlang/src/main/java/org/springframework/erlang/ErlangErrorRpcException.java
View File

@@ -1,44 +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.erlang;
import com.ericsson.otp.erlang.OtpErlangTuple;
/**
* Exception thrown when an 'error' is received from an Erlang RPC call
*
* @author Mark Pollack
*/
@SuppressWarnings("serial")
public class ErlangErrorRpcException extends OtpException {
private OtpErlangTuple reasonTuple;
public ErlangErrorRpcException(String message) {
super(message);
}
public ErlangErrorRpcException(OtpErlangTuple tuple) {
super(tuple.toString());
this.reasonTuple = tuple;
}
public OtpErlangTuple getReasonTuple() {
return reasonTuple;
}
}

35
spring-erlang/src/main/java/org/springframework/erlang/OtpAuthException.java
View File

@@ -1,35 +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.erlang;
/**
* Runtime exception mirroring the OTP OtpAuthException.
*
* @author Mark Pollack
*/
@SuppressWarnings("serial")
public class OtpAuthException extends OtpException {
public OtpAuthException(com.ericsson.otp.erlang.OtpAuthException cause) {
super(cause);
}
public OtpAuthException(String msg, com.ericsson.otp.erlang.OtpAuthException cause) {
super(msg, cause);
}
}

39
spring-erlang/src/main/java/org/springframework/erlang/OtpException.java
View File

@@ -1,39 +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.erlang;
/**
* Base RuntimeException for errors that occur when executing OTP operations.
*
* @author Mark Pollack
*/
@SuppressWarnings("serial")
public class OtpException extends RuntimeException {
public OtpException(String message) {
super(message);
}
public OtpException(Throwable cause) {
super(cause);
}
public OtpException(String message, Throwable cause) {
super(message, cause);
}
}

39
spring-erlang/src/main/java/org/springframework/erlang/OtpIOException.java
View File

@@ -1,39 +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.erlang;
import java.io.IOException;
/**
* RuntimeException wrapper for an {@link IOException} which
* can be commonly thrown from OTP operations.
*
* @author Mark Pollack
* @author Mark Fisher
*/
@SuppressWarnings("serial")
public class OtpIOException extends OtpException {
public OtpIOException(IOException cause) {
super(cause);
}
public OtpIOException(String message, IOException cause) {
super(message, cause);
}
}

36
spring-erlang/src/main/java/org/springframework/erlang/UncategorizedOtpException.java
View File

@@ -1,36 +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.erlang;
/**
* A "catch-all" exception type within the OtpException hierarchy
* when no more specific cause is known.
*
* @author Mark Pollack
*/
@SuppressWarnings("serial")
public class UncategorizedOtpException extends OtpException {
public UncategorizedOtpException(Throwable cause) {
super(cause);
}
public UncategorizedOtpException(String message, Throwable cause) {
super(message, cause);
}
}

102
spring-erlang/src/main/java/org/springframework/erlang/connection/Connection.java
View File

@@ -1,102 +0,0 @@
/*
* Copyright 2002-2015 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.erlang.connection;
import java.io.IOException;
import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpErlangExit;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangObject;
/**
* A simple interface that is used to wrap access to the {@code OtpConnection} class
* in order to support caching of {@code OtpConnection}s via method interception.
*
* Note: The surface area of the API is all that is required to implement administrative
* functionality for the Spring AMQP admin project.
* To access the underlying {@code OtpConnection}, use the method {@code getTargetConnection}
* on the interface {@code ConnectionProxy} that is implemented by {@code DefaultConnection}.
*
* @author Mark Pollack
* @author aArtem Bilan
*/
public interface Connection {
/**
* Close the connection to the remote node.
*/
void close();
/**
* Send an RPC request to the remote Erlang node. This convenience function
* creates the following message and sends it to 'rex' on the remote node:
* <pre class="code">
* { self, { call, Mod, Fun, Args, user } }
* </pre>
* <p>
* Note that this method has unpredictable results if the remote node is not
* an Erlang node.
* @param mod the name of the Erlang module containing the function to be called.
* @param fun the name of the function to call.
* @param args a list of Erlang terms, to be used as arguments to the function.
* @exception java.io.IOException if the connection is not active or a communication
* error occurs.
*/
void sendRPC(final String mod, final String fun, final OtpErlangList args) throws IOException;
/**
* Receive an RPC reply from the remote Erlang node. This convenience
* function receives a message from the remote node, and expects it to have
* the following format:
* <pre class="code">
* { rex, Term }
* </pre>
* @return the second element of the tuple if the received message is a
* two-tuple, otherwise null. No further error checking is
* performed.
* @exception java.io.IOException if the connection is not active or a communication
* error occurs.
* @exception OtpErlangExit if an exit signal is received from a process on the
* peer node.
* @exception OtpAuthException if the remote node sends a message containing an
* invalid cookie.
*/
OtpErlangObject receiveRPC() throws IOException, OtpErlangExit, OtpAuthException;
/**
* Send an RPC request to the remote Erlang node and receive result.
* The implementation must ensure {@code synchronized} mode of this method since
* the underlying {@code OtpConnection} isn't thread-safe.
* @param mod the name of the Erlang module containing the function to be called.
* @param fun the name of the function to call.
* @param args a list of Erlang terms, to be used as arguments to the function.
* @return the second element of the tuple if the received message is a
* two-tuple, otherwise null. No further error checking is
* performed.
* @exception java.io.IOException if the connection is not active or a communication
* error occurs.
* @exception OtpErlangExit if an exit signal is received from a process on the
* peer node.
* @exception OtpAuthException if the remote node sends a message containing an
* invalid cookie.
* @since 1.5
*/
OtpErlangObject sendAndReceiveRPC(final String mod, final String fun, final OtpErlangList args)
throws IOException, OtpErlangExit, OtpAuthException;
}

36
spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactory.java
View File

@@ -1,36 +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.erlang.connection;
import java.io.IOException;
import java.net.UnknownHostException;
import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpConnection;
/**
* An interface based ConnectionFactory for creating {@link OtpConnection}s.
*
* <p>NOTE: The Rabbit API contains a ConnectionFactory class (same name).
*
* @author Mark Pollack
*/
public interface ConnectionFactory {
Connection createConnection() throws UnknownHostException, OtpAuthException, IOException;
}

50
spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactoryUtils.java
View File

@@ -1,50 +0,0 @@
/*
* Copyright 2002-2015 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.erlang.connection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* @author Mark Pollack
* @author Gary Russell
*/
public class ConnectionFactoryUtils {
private static final Log logger = LogFactory.getLog(ConnectionFactoryUtils.class);
/**
* Release the given Connection by closing it.
*
* @param con The connection.
* @param cf The connection factory.
*/
public static void releaseConnection(Connection con, ConnectionFactory cf) {
if (con == null) {
return;
}
try {
con.close();
}
catch (Exception ex) {
logger.debug("Could not close Otp Connection", ex);
}
}
}

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

@@ -1,51 +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.erlang.connection;
import org.springframework.util.Assert;
import com.ericsson.otp.erlang.OtpPeer;
import com.ericsson.otp.erlang.OtpSelf;
/**
* Encapsulate properties to create a OtpConnection
* @author Mark Pollack
*
*/
public class ConnectionParameters {
private OtpSelf otpSelf;
private OtpPeer otpPeer;
public ConnectionParameters(OtpSelf otpSelf, OtpPeer otpPeer) {
Assert.notNull(otpSelf, "OtpSelf must be non-null");
Assert.notNull(otpPeer, "OtpPeer must be non-null");
this.otpSelf = otpSelf;
this.otpPeer = otpPeer;
}
public OtpSelf getOtpSelf() {
return otpSelf;
}
public OtpPeer getOtpPeer() {
return otpPeer;
}
}

31
spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionProxy.java
View File

@@ -1,31 +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.erlang.connection;
import com.ericsson.otp.erlang.OtpConnection;
/**
* Subinterface of {@link Connection} to be implemented by Connection proxies.
* Allows access to the underlying target Connection
*
* @author Mark Pollack
*
*/
public interface ConnectionProxy extends Connection {
OtpConnection getTargetConnection();
}

68
spring-erlang/src/main/java/org/springframework/erlang/connection/DefaultConnection.java
View File

@@ -1,68 +0,0 @@
/*
* Copyright 2002-2015 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.erlang.connection;
import java.io.IOException;
import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpConnection;
import com.ericsson.otp.erlang.OtpErlangExit;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangObject;
/**
* Basic implementation of {@link ConnectionProxy} that delegates
* to an underlying {@link OtpConnection}.
* @author Mark Pollack
* @author Artem Bilan
*/
public class DefaultConnection implements ConnectionProxy {
private OtpConnection otpConnection;
public DefaultConnection(OtpConnection otpConnection) {
this.otpConnection = otpConnection;
}
@Override
public void close() {
this.otpConnection.close();
}
@Override
public void sendRPC(String mod, String fun, OtpErlangList args) throws IOException {
this.otpConnection.sendRPC(mod, fun, args);
}
@Override
public OtpErlangObject receiveRPC() throws IOException, OtpErlangExit, OtpAuthException {
return this.otpConnection.receiveRPC();
}
@Override
public synchronized OtpErlangObject sendAndReceiveRPC(String mod, String fun, OtpErlangList args)
throws IOException, OtpErlangExit, OtpAuthException {
sendRPC(mod, fun, args);
return receiveRPC();
}
public OtpConnection getTargetConnection() {
return this.otpConnection;
}
}

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

@@ -1,138 +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.erlang.connection;
import java.io.IOException;
import java.net.UnknownHostException;
import java.util.UUID;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.erlang.OtpIOException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpPeer;
import com.ericsson.otp.erlang.OtpSelf;
/**
* <p>
* 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.
* </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.
* </p>
* @author Mark Pollack
* @author Mark Fisher
* @author Dave Syer
*/
public class SimpleConnectionFactory implements ConnectionFactory, InitializingBean {
protected final Log logger = LogFactory.getLog(getClass());
private boolean uniqueSelfNodeName = true;
private final String selfNodeName;
private final String peerNodeName;
private final String cookie;
private OtpSelf otpSelf;
private OtpPeer otpPeer;
public SimpleConnectionFactory(String selfNodeName, String peerNodeName, String cookie) {
this.selfNodeName = selfNodeName;
this.peerNodeName = peerNodeName;
this.cookie = cookie;
}
public SimpleConnectionFactory(String selfNodeName, String peerNodeName) {
this(selfNodeName, peerNodeName, null);
}
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);
}
}
public boolean isUniqueSelfNodeName() {
return uniqueSelfNodeName;
}
public void setUniqueSelfNodeName(boolean uniqueSelfNodeName) {
this.uniqueSelfNodeName = uniqueSelfNodeName;
}
public void afterPropertiesSet() {
Assert.isTrue(this.selfNodeName != null && this.peerNodeName != null,
"'selfNodeName' and 'peerNodeName' are required");
String selfNodeNameToUse = this.selfNodeName;
if (isUniqueSelfNodeName()) {
selfNodeNameToUse = this.selfNodeName + "-" + UUID.randomUUID().toString();
logger.debug("Creating OtpSelf with node name = [" + selfNodeNameToUse + "]");
}
try {
if (StringUtils.hasText(cookie)) {
this.otpSelf = new OtpSelf(selfNodeNameToUse.trim(), this.cookie);
} else {
this.otpSelf = new OtpSelf(selfNodeNameToUse.trim());
}
} catch (IOException e) {
throw new OtpIOException(e);
}
this.otpPeer = new OtpPeer(this.peerNodeName.trim());
}
}

300
spring-erlang/src/main/java/org/springframework/erlang/connection/SingleConnectionFactory.java
View File

@@ -1,300 +0,0 @@
/*
* Copyright 2002-2015 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.erlang.connection;
import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.erlang.OtpIOException;
import org.springframework.util.Assert;
import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpPeer;
import com.ericsson.otp.erlang.OtpSelf;
/**
* A {@link ConnectionFactory} implementation that returns the same Connections from all
* {@link #createConnection()} calls, and ignores calls to {@link Connection#close()}.
*
* 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.
* </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.
* </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.
* </p>
*
* @author Mark Pollack
*/
public class SingleConnectionFactory implements ConnectionFactory,
InitializingBean, DisposableBean {
protected final Log logger = LogFactory.getLog(getClass());
private boolean uniqueSelfNodeName = true;
private final String selfNodeName;
private String cookie;
private final String peerNodeName;
private OtpSelf otpSelf;
private OtpPeer otpPeer;
/** Raw JInterface Connection */
private Connection targetConnection;
/** Proxy Connection */
private Connection connection;
/** Synchronization monitor for the shared Connection */
private final Object connectionMonitor = new Object();
public SingleConnectionFactory(String selfNodeName, String cookie,
String peerNodeName) {
this.selfNodeName = selfNodeName;
this.cookie = cookie;
this.peerNodeName = peerNodeName;
}
public SingleConnectionFactory(String selfNodeName, String peerNodeName) {
this.selfNodeName = selfNodeName;
this.peerNodeName = peerNodeName;
}
public boolean isUniqueSelfNodeName() {
return uniqueSelfNodeName;
}
public void setUniqueSelfNodeName(boolean uniqueSelfNodeName) {
this.uniqueSelfNodeName = uniqueSelfNodeName;
}
@Override
public Connection createConnection() throws UnknownHostException,
OtpAuthException {
synchronized (this.connectionMonitor) {
if (this.connection == null) {
try {
initConnection();
} catch (IOException e) {
throw new OtpIOException("failed to connect from '"
+ this.selfNodeName + "' to peer node '"
+ this.peerNodeName + "'", e);
}
}
return this.connection;
}
}
public void initConnection() throws IOException, OtpAuthException {
synchronized (this.connectionMonitor) {
if (this.targetConnection != null) {
closeConnection(this.targetConnection);
}
this.targetConnection = doCreateConnection();
prepareConnection(this.targetConnection);
if (logger.isInfoEnabled()) {
logger.info("Established shared Rabbit Connection: "
+ this.targetConnection);
}
this.connection = getSharedConnectionProxy(this.targetConnection);
}
}
/**
* Close the underlying shared connection.
* The provider of this ConnectionFactory needs to care for proper shutdown.
* <p>As this bean implements DisposableBean, a bean factory will
* automatically invoke this on destruction of its cached singletons.
*/
@Override
public void destroy() {
resetConnection();
}
/**
* Reset the underlying shared Connection, to be reinitialized on next access.
*/
public void resetConnection() {
synchronized (this.connectionMonitor) {
if (this.targetConnection != null) {
closeConnection(this.targetConnection);
}
this.targetConnection = null;
this.connection = null;
}
}
/**
* Close the given Connection.
*
* @param connection
* the Connection to close
*/
protected void closeConnection(Connection connection) {
if (logger.isDebugEnabled()) {
logger.debug("Closing shared Rabbit Connection: "
+ this.targetConnection);
}
try {
// TODO there are other close overloads close(int closeCode,
// java.lang.String closeMessage, int timeout)
connection.close();
}
catch (Exception ex) {
logger.debug("Could not close shared Rabbit Connection", ex);
}
}
/**
* Create a JInterface Connection via this class's ConnectionFactory.
*
* @return the new Otp Connection
* @throws OtpAuthException Any.
* @throws IOException Any.
*/
protected Connection doCreateConnection() throws IOException,
OtpAuthException {
return new DefaultConnection(otpSelf.connect(otpPeer));
}
protected void prepareConnection(Connection con) throws IOException {
}
/**
* Wrap the given OtpConnection with a proxy that delegates every method
* call to it but suppresses close calls. This is useful for allowing
* application code to handle a special framework Connection just like an
* ordinary Connection from a Rabbit ConnectionFactory.
*
* @param target
* the original Connection to wrap
* @return the wrapped Connection
*/
protected Connection getSharedConnectionProxy(Connection target) {
List<Class<?>> classes = new ArrayList<Class<?>>(1);
classes.add(Connection.class);
return (Connection) Proxy.newProxyInstance(
Connection.class.getClassLoader(),
classes.toArray(new Class<?>[classes.size()]),
new SharedConnectionInvocationHandler(target));
}
/*
* (non-Javadoc)
*
* @see
* org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
@Override
public void afterPropertiesSet() {
Assert.isTrue(this.selfNodeName != null || this.peerNodeName != null,
"'selfNodeName' or 'peerNodeName' is required");
String selfNodeNameToUse = this.selfNodeName;
if (isUniqueSelfNodeName()) {
selfNodeNameToUse = this.selfNodeName + "-" + UUID.randomUUID().toString();
logger.debug("Creating OtpSelf with node name = [" + selfNodeNameToUse + "]");
}
try {
if (this.cookie == null) {
this.otpSelf = new OtpSelf(selfNodeNameToUse.trim());
} else {
this.otpSelf = new OtpSelf(selfNodeNameToUse.trim(), this.cookie);
}
} catch (IOException e) {
throw new OtpIOException(e);
}
this.otpPeer = new OtpPeer(this.peerNodeName.trim());
}
private static class SharedConnectionInvocationHandler implements
InvocationHandler {
private final Connection target;
public SharedConnectionInvocationHandler(Connection target) {
this.target = target;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
if (method.getName().equals("equals")) {
// Only consider equal when proxies are identical.
return (proxy == args[0]);
} else if (method.getName().equals("hashCode")) {
// Use hashCode of Connection proxy.
return System.identityHashCode(proxy);
} else if (method.getName().equals("toString")) {
return "Shared Otp Connection: " + this.target;
} else if (method.getName().equals("close")) {
// Handle close method: don't pass the call on.
return null;
}
try {
return method.invoke(this.target, args);
} catch (InvocationTargetException ex) {
throw ex.getTargetException();
}
}
}
}

4
spring-erlang/src/main/java/org/springframework/erlang/connection/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Provides classes supporting connections.
*/
package org.springframework.erlang.connection;

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

@@ -1,67 +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.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.
*
* See http://www.erlang.org/doc/man/app.html for full details
*
* @author Mark Pollack
*
*/
@SuppressWarnings("serial")
public class Application implements Serializable {
private String description;
private String id;
private String version;
public Application(String description, String id, String version) {
super();
this.description = description;
this.id = id;
this.version = version;
}
public String getDescription() {
return description;
}
public String getId() {
return id;
}
public String getVersion() {
return version;
}
@Override
public String toString() {
return "Application [description=" + description + ", id=" + id
+ ", version=" + version + "]";
}
}

37
spring-erlang/src/main/java/org/springframework/erlang/core/ConnectionCallback.java
View File

@@ -1,37 +0,0 @@
/*
* Copyright 2002-2014 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.erlang.core;
import org.springframework.erlang.connection.Connection;
/**
* Basic callback for use in ErlangTemplate
* @author Mark Pollack
*/
public interface ConnectionCallback<T> {
/**
* Execute any number of operations against the supplied OTP connection,
* possibly returning a result.
*
* @param connection The connection.
* @return The result.
* @throws Exception We are not sure everything it throws.
*/
T doInConnection(Connection connection) throws Exception;
}

46
spring-erlang/src/main/java/org/springframework/erlang/core/ErlangOperations.java
View File

@@ -1,46 +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.erlang.core;
import org.springframework.erlang.OtpException;
import org.springframework.erlang.support.converter.ErlangConverter;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangObject;
/**
* Operations to perform against OTP/Erlang
* @author Mark Pollack
*
*/
public interface ErlangOperations {
<T> T execute(ConnectionCallback<T> action) throws OtpException;
OtpErlangObject executeErlangRpc(String module, String function, OtpErlangList args) throws OtpException;
OtpErlangObject executeErlangRpc(String module, String function, OtpErlangObject... args) throws OtpException;
OtpErlangObject executeRpc(String module, String function, Object... args) throws OtpException;
Object executeAndConvertRpc(String module, String function, ErlangConverter converterToUse, Object... args) throws OtpException;
Object executeAndConvertRpc(String module, String function, Object... args) throws OtpException;
}

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

@@ -1,152 +0,0 @@
/*
* Copyright 2002-2015 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.erlang.core;
import org.springframework.erlang.ErlangBadRpcException;
import org.springframework.erlang.ErlangErrorRpcException;
import org.springframework.erlang.OtpException;
import org.springframework.erlang.connection.Connection;
import org.springframework.erlang.connection.ConnectionFactory;
import org.springframework.erlang.support.ErlangAccessor;
import org.springframework.erlang.support.ErlangUtils;
import org.springframework.erlang.support.converter.ErlangConverter;
import org.springframework.erlang.support.converter.SimpleErlangConverter;
import org.springframework.util.Assert;
import com.ericsson.otp.erlang.*;
/**
* @author Mark Pollack
* @author Artem Bilan
*/
public class ErlangTemplate extends ErlangAccessor implements ErlangOperations {
private volatile ErlangConverter erlangConverter = new SimpleErlangConverter();
public ErlangTemplate(ConnectionFactory connectionFactory) {
setConnectionFactory(connectionFactory);
afterPropertiesSet();
}
public OtpErlangObject executeErlangRpc(final String module, final String function, final OtpErlangList args) {
return execute(new ConnectionCallback<OtpErlangObject>() {
public OtpErlangObject doInConnection(Connection connection) throws Exception {
logger.debug("Sending RPC for module [" + module + "] function [" + function + "] args [" + args);
OtpErlangObject response = connection.sendAndReceiveRPC(module, function, args);
logger.debug("Response received = " + response.toString());
handleResponseError(module, function, response);
return response;
}
});
}
public void handleResponseError(String module, String function, OtpErlangObject result) {
//{badrpc,{'EXIT',{undef,[{rabbit_access_control,list_users,[[]]},{rpc,'-handle_call/3-fun-0-',5}]}}}
if (result instanceof OtpErlangTuple) {
OtpErlangTuple msg = (OtpErlangTuple)result;
if (msg.elementAt(0) instanceof OtpErlangAtom)
{
OtpErlangAtom responseAtom = (OtpErlangAtom)msg.elementAt(0);
//TODO consider error handler strategy.
if (responseAtom.atomValue().equals("badrpc")) {
if (msg.elementAt(1) instanceof OtpErlangTuple) {
throw new ErlangBadRpcException( (OtpErlangTuple)msg.elementAt(1));
} else {
throw new ErlangBadRpcException( msg.elementAt(1).toString());
}
} else if (responseAtom.atomValue().equals("error")) {
if (msg.elementAt(1) instanceof OtpErlangTuple) {
throw new ErlangErrorRpcException( (OtpErlangTuple)msg.elementAt(1));
} else {
throw new ErlangErrorRpcException( msg.elementAt(1).toString());
}
}
}
}
}
public OtpErlangObject executeErlangRpc(String module, String function, OtpErlangObject... args) {
return executeRpc(module, function, new OtpErlangList(args));
}
public OtpErlangObject executeRpc(String module, String function, Object... args) {
return executeErlangRpc(module, function, (OtpErlangList) erlangConverter.toErlang(args));
}
public Object executeAndConvertRpc(String module, String function, ErlangConverter converterToUse, Object... args) {
return converterToUse.fromErlang(executeRpc(module, function, converterToUse.toErlang(args)));
}
public Object executeAndConvertRpc(String module, String function, Object... args) {
return erlangConverter.fromErlangRpc(module, function, executeErlangRpc(module, function, (OtpErlangList)erlangConverter.toErlang(args)));
}
public ErlangConverter getErlangConverter() {
return erlangConverter;
}
public void setErlangConverter(ErlangConverter erlangConverter) {
this.erlangConverter = erlangConverter;
}
public <T> T execute(ConnectionCallback<T> action) throws OtpException {
Assert.notNull(action, "Callback object must not be null");
Connection con = null;
try {
con = createConnection();
return action.doInConnection(con);
}
catch (OtpException ex) {
throw ex;
}
catch (Exception ex) {
throw convertOtpAccessException(ex);
}
finally {
org.springframework.erlang.connection.ConnectionFactoryUtils.releaseConnection(con, getConnectionFactory());
}
// TODO: physically close and reopen the connection if there is an exception
}
/**
* Convert the specified checked exception to
* a Spring runtime exception equivalent.
* <p>The default implementation delegates to the
* {@link org.springframework.erlang.support.ErlangUtils#convertOtpAccessException} method.
* @param ex the original checked {@link Exception} to convert
* @return the Spring runtime wrapping <code>ex</code>
* @see org.springframework.erlang.support.ErlangUtils#convertOtpAccessException
*/
protected OtpException convertOtpAccessException(Exception ex) {
return ErlangUtils.convertOtpAccessException(ex);
}
}

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

@@ -1,48 +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.
*/
/**
* Describes an Erlang node.
*/
package org.springframework.erlang.core;
import java.io.Serializable;
/**
* Simple description class for an Erlang node.
*
* @author Mark Pollack
*
*/
@SuppressWarnings("serial")
public class Node implements Serializable {
private String name;
public Node(String name) {
this.name = name;
}
public String getName() {
return name;
}
@Override
public String toString() {
return "Node [name=" + name + "]";
}
}

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

@@ -1,4 +0,0 @@
/**
* Provides core classes for Spring Erlang.
*/
package org.springframework.erlang.core;

4
spring-erlang/src/main/java/org/springframework/erlang/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Base package for Spring Erlang.
*/
package org.springframework.erlang;

64
spring-erlang/src/main/java/org/springframework/erlang/support/ErlangAccessor.java
View File

@@ -1,64 +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.erlang.support;
import java.io.IOException;
import java.net.UnknownHostException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.erlang.OtpException;
import org.springframework.erlang.connection.Connection;
import org.springframework.erlang.connection.ConnectionFactory;
import com.ericsson.otp.erlang.OtpAuthException;
/**
* @author Mark Pollack
*/
public abstract class ErlangAccessor implements InitializingBean {
/** Logger available to subclasses */
protected final Log logger = LogFactory.getLog(getClass());
private ConnectionFactory connectionFactory;
protected Connection createConnection() throws UnknownHostException, OtpAuthException, IOException {
return getConnectionFactory().createConnection();
}
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
public ConnectionFactory getConnectionFactory() {
return this.connectionFactory;
}
public void afterPropertiesSet() {
if (getConnectionFactory() == null) {
throw new IllegalArgumentException("Property 'connectionFactory' is required");
}
}
protected OtpException convertOtpAccessException(Exception ex) {
return ErlangUtils.convertOtpAccessException(ex);
}
}

57
spring-erlang/src/main/java/org/springframework/erlang/support/ErlangUtils.java
View File

@@ -1,57 +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.erlang.support;
import java.io.IOException;
import org.springframework.erlang.OtpException;
import org.springframework.erlang.OtpIOException;
import org.springframework.erlang.UncategorizedOtpException;
import org.springframework.util.Assert;
import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpConnection;
/**
* @author Mark Pollack
*/
public class ErlangUtils {
/**
* Close the given Connection.
* @param con the Connection to close if necessary
* (if this is <code>null</code>, the call will be ignored)
*/
public static void releaseConnection(OtpConnection con) {
if (con == null) {
return;
}
con.close();
}
public static OtpException convertOtpAccessException(Exception ex) {
Assert.notNull(ex, "Exception must not be null");
if (ex instanceof IOException) {
return new OtpIOException((IOException) ex);
}
if (ex instanceof OtpAuthException) {
return new org.springframework.erlang.OtpAuthException((OtpAuthException) ex);
}
//fallback
return new UncategorizedOtpException(ex);
}
}

35
spring-erlang/src/main/java/org/springframework/erlang/support/converter/ErlangConversionException.java
View File

@@ -1,35 +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.erlang.support.converter;
import org.springframework.erlang.OtpException;
/**
* @author Mark Pollack
*/
@SuppressWarnings("serial")
public class ErlangConversionException extends OtpException {
public ErlangConversionException(String message, Throwable cause) {
super(message, cause);
}
public ErlangConversionException(String message) {
super(message);
}
}

45
spring-erlang/src/main/java/org/springframework/erlang/support/converter/ErlangConverter.java
View File

@@ -1,45 +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.erlang.support.converter;
import com.ericsson.otp.erlang.OtpErlangObject;
/**
* Converter between Java and Erlang Types. Additional support for converting results from RPC calls.
*
* @author Mark Pollack
*/
public interface ErlangConverter {
/**
* Convert a Java object to a Erlang data type.
* @param object the object to convert
* @return the Erlang data type
* @throws ErlangConversionException in case of conversion failure
*/
OtpErlangObject toErlang(Object object) throws ErlangConversionException;
/**
* Convert from a Erlang data type to a Java object.
* @param erlangObject the Elang object to convert
* @return the converted Java object
* @throws ErlangConversionException in case of conversion failure
*/
Object fromErlang(OtpErlangObject erlangObject) throws ErlangConversionException;
Object fromErlangRpc(String module, String function, OtpErlangObject erlangObject) throws ErlangConversionException;
}

158
spring-erlang/src/main/java/org/springframework/erlang/support/converter/SimpleErlangConverter.java
View File

@@ -1,158 +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.erlang.support.converter;
import java.util.ArrayList;
import com.ericsson.otp.erlang.OtpErlangAtom;
import com.ericsson.otp.erlang.OtpErlangBinary;
import com.ericsson.otp.erlang.OtpErlangBoolean;
import com.ericsson.otp.erlang.OtpErlangByte;
import com.ericsson.otp.erlang.OtpErlangChar;
import com.ericsson.otp.erlang.OtpErlangDouble;
import com.ericsson.otp.erlang.OtpErlangFloat;
import com.ericsson.otp.erlang.OtpErlangInt;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangLong;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpErlangPid;
import com.ericsson.otp.erlang.OtpErlangRangeException;
import com.ericsson.otp.erlang.OtpErlangShort;
import com.ericsson.otp.erlang.OtpErlangString;
/**
* Converter that supports the basic types and arrays.
* @author Mark Pollack
*
*/
public class SimpleErlangConverter implements ErlangConverter {
public Object fromErlang(OtpErlangObject erlangObject)
throws ErlangConversionException {
//TODO support arrays.
return convertErlangToBasicType(erlangObject);
}
public Object fromErlangRpc(String module, String function,
OtpErlangObject erlangObject) throws ErlangConversionException {
return this.fromErlang(erlangObject);
}
public OtpErlangObject toErlang(Object obj)
throws ErlangConversionException {
if (obj instanceof OtpErlangObject) {
return (OtpErlangObject) obj;
}
if (obj instanceof Object[]) {
Object[] objectsToConvert = (Object[]) obj;
if (objectsToConvert.length != 0) {
ArrayList<OtpErlangObject> tempList = new ArrayList<OtpErlangObject>();
for (Object objectToConvert : objectsToConvert) {
OtpErlangObject erlangObject = convertBasicTypeToErlang(objectToConvert);
tempList.add(erlangObject);
}
OtpErlangObject ia[] = new OtpErlangObject[tempList.size()];
return new OtpErlangList(tempList.toArray(ia));
} else {
return new OtpErlangList();
}
} else {
return convertBasicTypeToErlang(obj);
}
}
protected OtpErlangObject convertBasicTypeToErlang(Object obj) {
if (obj instanceof byte[]) {
return new OtpErlangBinary((byte[]) obj);
} else if (obj instanceof Boolean) {
return new OtpErlangBoolean((Boolean) obj);
} else if (obj instanceof Byte) {
return new OtpErlangByte((Byte) obj);
} else if (obj instanceof Character) {
return new OtpErlangChar((Character) obj);
} else if (obj instanceof Double) {
return new OtpErlangDouble((Double) obj);
} else if (obj instanceof Float) {
return new OtpErlangFloat((Float) obj);
} else if (obj instanceof Integer) {
return new OtpErlangInt((Integer) obj);
} else if (obj instanceof Long) {
return new OtpErlangLong((Long) obj);
} else if (obj instanceof Short) {
return new OtpErlangShort((Short) obj);
} else if (obj instanceof String) {
return new OtpErlangString((String) obj);
} else {
throw new ErlangConversionException(
"Could not convert Java object of type [" + obj.getClass()
+ "] to an Erlang data type.");
}
}
protected Object convertErlangToBasicType(OtpErlangObject erlangObject) {
try {
if (erlangObject instanceof OtpErlangBinary) {
return ((OtpErlangBinary) erlangObject).binaryValue();
} else if (erlangObject instanceof OtpErlangAtom) {
return ((OtpErlangAtom) erlangObject).atomValue();
} else if (erlangObject instanceof OtpErlangBinary) {
return ((OtpErlangBinary) erlangObject).binaryValue();
} else if (erlangObject instanceof OtpErlangBoolean) {
return extractBoolean(erlangObject);
} else if (erlangObject instanceof OtpErlangByte) {
return ((OtpErlangByte) erlangObject).byteValue();
} else if (erlangObject instanceof OtpErlangChar) {
return ((OtpErlangChar) erlangObject).charValue();
} else if (erlangObject instanceof OtpErlangDouble) {
return ((OtpErlangDouble) erlangObject).doubleValue();
} else if (erlangObject instanceof OtpErlangFloat) {
return ((OtpErlangFloat) erlangObject).floatValue();
} else if (erlangObject instanceof OtpErlangInt) {
return ((OtpErlangInt) erlangObject).intValue();
} else if (erlangObject instanceof OtpErlangLong) {
return ((OtpErlangLong) erlangObject).longValue();
} else if (erlangObject instanceof OtpErlangShort) {
return ((OtpErlangShort) erlangObject).shortValue();
} else if (erlangObject instanceof OtpErlangString) {
return ((OtpErlangString) erlangObject).stringValue();
} else if (erlangObject instanceof OtpErlangPid) {
return ((OtpErlangPid) erlangObject).toString();
} else {
throw new ErlangConversionException(
"Could not convert Erlang object ["
+ erlangObject.getClass() + "] to Java type.");
}
} catch (OtpErlangRangeException e) {
throw new ErlangConversionException(
"Could not convert Erlang object ["
+ erlangObject.getClass() + "] to Java type.", e);
}
}
public static boolean extractBoolean(OtpErlangObject erlangObject) {
return ((OtpErlangBoolean) erlangObject).booleanValue();
}
public static String extractPid(OtpErlangObject value) {
return ((OtpErlangPid)value).toString();
}
public static long extractLong(OtpErlangObject value) {
return ((OtpErlangLong)value).longValue();
}
}

4
spring-erlang/src/main/java/org/springframework/erlang/support/converter/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Provides classes conversion between Java and Erlang types.
*/
package org.springframework.erlang.support.converter;

4
spring-erlang/src/main/java/org/springframework/erlang/support/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Provides support classes for Spring Erlang.
*/
package org.springframework.erlang.support;

143
spring-erlang/src/main/java/org/springframework/util/exec/Background.java
View File

@@ -1,143 +0,0 @@
/*
* Copyright 2002-2014 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.util.exec;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
/**
* Class to execute processes in the background. These processes
* will not exit once the controlling Java process exits.
*/
public class Background {
/**
* Execute the command (and its args, ala Runtime.exec), sending the
* output and error streams to the void.
*
* @param cmd The command and args.
* @throws IOException Any.
*/
public static void exec(String[] cmd)
throws IOException
{
File devNull;
if(Os.isFamily("unix")) {
devNull = new File("/dev/null");
}
else if (Os.isFamily("windows")) {
devNull = new File("NUL");
}
else {
throw new IllegalStateException("Unhandled Java environment");
}
exec(cmd, devNull, false, devNull, false);
}
/**
* Execute a command (and its args, ala Runtime.exec)
*
* @param cmd The command and args.
* @param outFile File to send standard out from the process to
* @param appendOut If true, append the file with standard out,
* else truncate or create a new file
* @param errFile File to send standard err from the process to
* @param appendErr If true, append the file with standard error,
* else truncate or create a new file
*
* @throws IOException Any.
*/
public static void exec(String[] cmd,
File outFile, boolean appendOut,
File errFile, boolean appendErr)
throws IOException
{
if(Os.isFamily("unix")) {
execUnix(cmd, outFile, appendOut, errFile, appendErr);
}
else if (Os.isFamily("windows")) {
execWin(cmd, outFile, appendOut, errFile, appendErr);
}
else {
throw new IllegalStateException("Unhandled Java environment");
}
}
private static void execUnix(String[] cmd,
File outFile, boolean appendOut,
File errFile, boolean appendErr)
throws IOException
{
StringBuffer escaped;
String[] execCmd;
escaped = new StringBuffer();
for(int i=0; i<cmd.length; i++){
escaped.append(Escape.escape(cmd[i]));
escaped.append(" ");
}
execCmd = new String[] {
"/bin/sh",
"-c",
escaped.toString() +
(appendOut == true ? ">>" : ">") +
Escape.escape(outFile.getAbsolutePath()) +
" 2" + (appendErr == true ? ">>" : " >") +
Escape.escape(errFile.getAbsolutePath()) +
" </dev/null &"
};
Process p = Runtime.getRuntime().exec(execCmd);
try {
p.waitFor();
} catch(Exception exc){
throw new IOException("Unable to properly background process: " +
exc.getMessage(), exc);
}
}
private static void execWin(String[] cmd,
File outFile, boolean appendOut,
File errFile, boolean appendErr)
throws IOException
{
ArrayList<String> tmpCmd = new ArrayList<String>();
tmpCmd.add("cmd");
tmpCmd.add("/c");
tmpCmd.add("start");
tmpCmd.add("/b");
tmpCmd.add("\"\"");
tmpCmd.add("/MIN");
for(int i=0; i<cmd.length; i++){
tmpCmd.add(cmd[i]);
}
tmpCmd.add((appendOut == true ? ">>" : ">") +
Escape.escape(outFile.getAbsolutePath()));
tmpCmd.add((outFile.equals(errFile) ? " 2&" : " 2") +
(appendErr == true ? ">>" : " >") +
Escape.escape(errFile.getAbsolutePath()));
Runtime.getRuntime().exec(tmpCmd.toArray(cmd));
}
public static void main(String[] args) throws Exception {
Background.exec(new String[] {"javaq", "foo bar", "bar" },
new File("garfo"), true, new File("barfo"), true);
}
}

70
spring-erlang/src/main/java/org/springframework/util/exec/Escape.java
View File

@@ -1,70 +0,0 @@
/*
* Copyright 2002-2014 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.util.exec;
public class Escape {
private static char[] enlargeArray(char[] in){
char[] res;
res = new char[in.length * 2];
System.arraycopy(in, 0, res, 0, in.length);
return res;
}
/**
* Escape a string by quoting the magical elements
* (such as whitespace, quotes, slashes, etc.)
*
* @param in The string to escape.
* @return The escaped string.
*/
public static String escape(String in){
char[] inChars, outChars;
int numOut;
inChars = new char[in.length()];
outChars = new char[inChars.length];
in.getChars(0, inChars.length, inChars, 0);
numOut = 0;
for(int i=0; i<inChars.length; i++){
if(outChars.length - numOut < 5){
outChars = enlargeArray(outChars);
}
if(Character.isWhitespace(inChars[i]) ||
inChars[i] == '\\' ||
inChars[i] == '\'' ||
inChars[i] == '\"' ||
inChars[i] == '&' ||
inChars[i] == ';')
{
outChars[numOut++] = '\\';
outChars[numOut++] = inChars[i];
} else {
outChars[numOut++] = inChars[i];
}
}
return new String(outChars, 0, numOut);
}
public static void main(String[] args){
System.out.println(Escape.escape("foo bar"));
System.out.println(Escape.escape("\\\"foo' bar\""));
}
}

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

@@ -1,408 +0,0 @@
/*
* Copyright 2002-2015 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.util.exec;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.StringReader;
import java.util.Arrays;
import java.util.StringTokenizer;
import java.util.Vector;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* 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
* @author Gary Russell
*/
public class Execute {
private static Log log = LogFactory.getLog(Execute.class);
/** Invalid exit code. **/
public static final int INVALID = Integer.MAX_VALUE;
private String[] cmdl = null;
private String[] env = null;
private int exitValue = INVALID;
private final ExecuteStreamHandler streamHandler;
private final ExecuteWatchdog watchdog;
private File workingDirectory = null;
private boolean newEnvironment = false;
private Process process;
private static Vector<String> procEnvironment = null;
/**
* Find the list of environment variables for this process.
*
* @return The environment.
*/
public static synchronized Vector<String> getProcEnvironment() {
if (procEnvironment != null) {
return procEnvironment;
}
procEnvironment = new Vector<String>();
try {
ByteArrayOutputStream out = new ByteArrayOutputStream();
Execute exe = new Execute(new PumpStreamHandler(out));
exe.setCommandline(getProcEnvCommand());
// Make sure we do not recurse forever
exe.setNewenvironment(true);
exe.execute();
BufferedReader in = new BufferedReader(new StringReader(out.toString()));//NOSONAR (default charset)
String var = null;
String line, lineSep = System.getProperty("line.separator");
while ((line = in.readLine()) != null) {
if (line.indexOf('=') == -1) {
// Chunk part of previous env var (UNIX env vars can
// contain embedded new lines).
if (var == null) {
var = lineSep + line;
} else {
var += lineSep + line;//NOSONAR
}
} else {
// New env var...append the previous one if we have it.
if (var != null) {
procEnvironment.addElement(var);
}
var = line;
}
}
// Since we "look ahead" before adding, there's one last env var.
procEnvironment.addElement(var);
}
catch (Exception exc) {
exc.printStackTrace();
// Just try to see how much we got
}
return procEnvironment;
}
private static String[] getProcEnvCommand() {
if (Os.isFamily("os/2")) {
// OS/2 - use same mechanism as Windows 2000
// Not sure
String[] cmd = { "cmd", "/c", "set" };
return cmd;
} else if (Os.isFamily("windows")) {
String[] cmd = { "cmd", "/c", "set" };
return cmd;
} else if (Os.isFamily("unix")) {
// Generic UNIX
// Alternatively one could use: /bin/sh -c env
String[] cmd = { "/usr/bin/env" };
return cmd;
} else if (Os.isFamily("netware")) {
String[] cmd = { "env" };
return cmd;
} else {
// MAC OS 9 and previous
// TODO: I have no idea how to get it, someone must fix it
String[] cmd = null;
return cmd;
}
}
/**
* Creates a new execute object using <code>PumpStreamHandler</code> for stream handling.
*/
public Execute() {
this(new PumpStreamHandler(), null);
}
/**
* Creates a new execute object.
*
* @param streamHandler the stream handler used to handle the input and output streams of the subprocess.
*/
public Execute(ExecuteStreamHandler streamHandler) {
this(streamHandler, null);
}
/**
* 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.
*/
public Execute(ExecuteStreamHandler streamHandler, ExecuteWatchdog watchdog) {
this.streamHandler = streamHandler;
this.watchdog = watchdog;
}
/**
* Returns the commandline used to create a subprocess.
*
* @return the commandline used to create a subprocess
*/
public String[] getCommandline() {
return cmdl;//NOSONAR
}
public String getCommandLineString() {
return array2string(getCommandline());
}
/**
* Sets the commandline of the subprocess to launch.
*
* @param commandline the commandline of the subprocess to launch
*/
public void setCommandline(String[] commandline) {
cmdl = Arrays.copyOf(commandline, commandline.length);
}
/**
* Set whether to propagate the default environment or not.
*
* @param newenv whether to propagate the process environment.
*/
public void setNewenvironment(boolean newenv) {
newEnvironment = newenv;
}
/**
* Returns the environment used to create a subprocess.
*
* @return the environment used to create a subprocess
*/
public String[] getEnvironment() {
if (env == null || newEnvironment) {
return env;//NOSONAR
}
return patchEnvironment();
}
/**
* 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>
*/
public void setEnvironment(String[] env) {
this.env = Arrays.copyOf(env, env.length);
}
/**
* 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.
*
* @param wd the working directory of the process.
*/
public void setWorkingDirectory(File wd) {
workingDirectory = wd;
}
/**
* 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
*/
public int execute() throws Exception {
process = Runtime.getRuntime().exec(getCommandline(), getEnvironment(), workingDirectory);
try {
streamHandler.setProcessInputStream(process.getOutputStream());
streamHandler.setProcessOutputStream(process.getInputStream());
streamHandler.setProcessErrorStream(process.getErrorStream());
} catch (IOException e) {
process.destroy();
throw e;
}
streamHandler.start();
if (watchdog != null) {
watchdog.start(process, Thread.currentThread());
}
if (log.isTraceEnabled()) {
log.trace("Waiting process ");
}
waitFor(process);
process = null;
if (log.isTraceEnabled()) {
log.trace("End waiting, stop threads ");
}
if (watchdog != null) {
watchdog.stop();
}
if (log.isTraceEnabled()) {
log.trace("Watchdog stopped ");
}
streamHandler.stop();
if (log.isTraceEnabled()) {
log.trace("Stream handler stopped ");
}
if (watchdog != null) {
Exception ex = watchdog.getException();
if (ex != null) {
throw ex;
}
}
int exit = getExitValue();
if (log.isDebugEnabled()) {
log.debug("Done exit=" + exit + " " + getCommandLineString());
}
return exit;
}
public void kill() {
if (process != null) {
process.destroy();
}
}
private String array2string(String sa[]) {
if (sa == null) {
return "null";
}
StringBuffer sb = new StringBuffer();
for (int i = 0; i < sa.length; i++) {
sb.append(sa[i]).append(" ");
}
return sb.toString();
}
protected void waitFor(Process process) {
try {
process.waitFor();
setExitValue(process.exitValue());
} catch (InterruptedException e) {
log.info("waitFor() interrupted ");
Thread.currentThread().interrupt();
}
}
protected void setExitValue(int value) {
exitValue = value;
}
/**
* 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
*/
public int getExitValue() {
return exitValue;
}
/**
* Patch the current environment with the new values from the user.
*
* @return the patched environment
*/
private String[] patchEnvironment() {
@SuppressWarnings("unchecked")
Vector<String> osEnv = (Vector<String>) getProcEnvironment().clone();
for (int i = 0; i < env.length; i++) {
int pos = env[i].indexOf('=');
// Get key including "="
String key = env[i].substring(0, pos + 1);
int size = osEnv.size();
for (int j = 0; j < size; j++) {
if ((osEnv.elementAt(j)).startsWith(key)) {
osEnv.removeElementAt(j);
break;
}
}
osEnv.addElement(env[i]);
}
String[] result = new String[osEnv.size()];
osEnv.copyInto(result);
return result;
}
public static int execute(Vector<String> envVars, String cmd, File baseDir) {
Vector<String> v = new Vector<String>();
StringTokenizer st = new StringTokenizer(cmd, " ");
while (st.hasMoreTokens()) {
v.addElement(st.nextToken());
}
return execute(envVars, v, 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
*
* @return The result.
*/
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);
watchdog.setDontkill(true);
PumpStreamHandler out = new PumpStreamHandler();
Execute exec = new Execute(out, watchdog);
String cmdA[] = new String[cmd.size()];
cmd.toArray(cmdA);
if (log.isDebugEnabled()) {
StringBuffer sb = new StringBuffer();
for (int i = 0; i < cmdA.length; i++) {
sb.append(cmdA[i] + " ");
}
log.debug("Exec: " + sb.toString());
}
exec.setCommandline(cmdA);
if (envVars != null) {
String env[] = new String[envVars.size()];
envVars.toArray(env);
exec.setEnvironment(env);
}
exec.setNewenvironment(false);
if (baseDir != null) {
exec.setWorkingDirectory(baseDir);
}
exec.execute();
int status = exec.getExitValue();
log.debug("Exit value " + status);
return status;
} catch (Exception ex) {
System.err.println("An error has occurred in Execute.");
return -1;
}
}
}

67
spring-erlang/src/main/java/org/springframework/util/exec/ExecuteStreamHandler.java
View File

@@ -1,67 +0,0 @@
/*
* Copyright 2002-2014 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.util.exec;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* Used by <code>Execute</code> to handle input and output stream of
* subprocesses.
*
* @author thomas.haas@softwired-inc.com
*/
public interface ExecuteStreamHandler {
/**
* Install a handler for the input stream of the subprocess.
*
* @param os output stream to write to the standard input stream of the
* subprocess
* @throws IOException Any.
*/
void setProcessInputStream(OutputStream os) throws IOException;
/**
* Install a handler for the error stream of the subprocess.
*
* @param is input stream to read from the error stream from the subprocess
* @throws IOException Any.
*/
void setProcessErrorStream(InputStream is) throws IOException;
/**
* Install a handler for the output stream of the subprocess.
*
* @param is input stream to read from the error stream from the subprocess
* @throws IOException Any.
*/
void setProcessOutputStream(InputStream is) throws IOException;
/**
* Start handling of the streams.
*
* @throws IOException Any.
*/
void start() throws IOException;
/**
* Stop handling of the streams - will not be restarted.
*/
void stop();
}

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

@@ -1,180 +0,0 @@
/*
* Copyright 2002-2015 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.util.exec;
/**
* Destroys a process running for too long.
* For example:
* <pre class="code">
* {@code
* ExecuteWatchdog watchdog = new ExecuteWatchdog(30000);
* Execute exec = new Execute(myloghandler, watchdog);
* exec.setCommandLine(mycmdline);
* int exitvalue = exec.execute();
* if (exitvalue != SUCCESS && watchdog.killedProcess()){
* // it was killed on purpose by the watchdog
* }
* }
* </pre>
* @author thomas.haas@softwired-inc.com
* @author <a href="mailto:sbailliez@imediation.com">Stephane Bailliez</a>
* @see Execute
*/
public class ExecuteWatchdog implements Runnable {
/** the process to execute and watch for duration */
private Process process;
/** timeout duration. Once the process running time exceeds this it should be killed */
private final int timeout;
/** say whether or not the watchog is currently monitoring a process */
private boolean watch = false;
/** exception that might be thrown during the process execution */
private Exception caught = null;
/** say whether or not the process was killed due to running overtime */
private boolean killedProcess = false;
Thread execThread;
private boolean dontkill=false;
/**
* Creates a new watchdog with a given timeout.
*
* @param timeout the timeout for the process in milliseconds. It must be greather than 0.
*/
public ExecuteWatchdog(int timeout) {
if (timeout < 1) {
throw new IllegalArgumentException("timeout lesser than 1.");
}
this.timeout = timeout;
}
public void setDontkill( boolean b ) {
dontkill=b;//NOSONAR
}
/**
* Watches the given process and terminates it, if it runs for too long.
* All information from the previous run are reset.
* @param process the process to monitor. It cannot be <tt>null</tt>
* @param execThread The thread.
* @throws IllegalStateException thrown if a process is still being monitored.
*/
public synchronized void start(Process process, Thread execThread) {
if (process == null) {
throw new NullPointerException("process is null.");
}
if (this.process != null) {
throw new IllegalStateException("Already running.");
}
this.caught = null;
this.killedProcess = false;
this.watch = true;
this.process = process;
final Thread thread = new Thread(this, "WATCHDOG");
this.execThread=execThread;
thread.setDaemon(true);
thread.start();
}
/**
* Stops the watcher. It will notify all threads possibly waiting on this object.
*/
public synchronized void stop() {
watch = false;
notifyAll();
}
/**
* Watches the process and terminates it, if it runs for too long.
*/
@Override
public synchronized void run() {
try {
// This isn't a Task, don't have a Project object to log.
// project.log("ExecuteWatchdog: timeout = "+timeout+" msec", Project.MSG_VERBOSE);
final long until = System.currentTimeMillis() + timeout;
long now;
while (watch && until > (now = System.currentTimeMillis())) {
try {
wait(until - now);
} catch (InterruptedException e) {}
}
// 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 {
// We must check if the process was not stopped
// before being here
process.exitValue();
} catch (IllegalThreadStateException e){
// The process is not terminated, if this is really
// a timeout and not a manual stop then kill it.
if (watch){
killedProcess = true;
if( ! dontkill ) {
process.destroy();
}
if( execThread != null ) {
execThread.interrupt();
}
}
}
} catch(Exception e) {
caught = e;
} finally {
cleanUp();
}
}
/**
* reset the monitor flag and the process.
*/
protected void cleanUp() {
watch = false;
process = null;
}
public Exception getException() {
return caught;
}
/**
* Indicates whether or not the watchdog is still monitoring the process.
* @return <tt>true</tt> if the process is still running, otherwise <tt>false</tt>.
*/
public boolean isWatching(){
return watch;
}
/**
* Indicates whether the last process run was killed on timeout or not.
* @return <tt>true</tt> if the process was killed otherwise <tt>false</tt>.
*/
public boolean killedProcess(){
return killedProcess;
}
}

148
spring-erlang/src/main/java/org/springframework/util/exec/Os.java
View File

@@ -1,148 +0,0 @@
/*
* Copyright 2002-2014 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.util.exec;
import java.util.Locale;
/**
* Condition that tests the OS type.
*
* @author <a href="mailto:stefan.bodewig@epost.de">Stefan Bodewig</a>
* @author <a href="mailto:umagesh@apache.org">Magesh Umasankar</a>
*/
public class Os {
private static final String osName =
System.getProperty("os.name").toLowerCase(Locale.US);
private static final String osArch =
System.getProperty("os.arch").toLowerCase(Locale.US);
private static final String osVersion =
System.getProperty("os.version").toLowerCase(Locale.US);
private static final String pathSep = System.getProperty("path.separator");
/**
* Determines if the OS on which Ant is executing matches the
* given OS family.
*
* @param family The OS family type desired<br>
* Possible values:<br>
* <ul><li>dos</li>
* <li>mac</li>
* <li>netware</li>
* <li>os/2</li>
* <li>unix</li>
* <li>windows</li></ul>
* @return true if the OS is in the family.
* @since 1.5
*/
public static boolean isFamily(String family) {
return isOs(family, null, null, null);
}
/**
* Determines if the OS on which Ant is executing matches the
* given OS name.
*
* @param name The name.
* @return The result.
* @since 1.7
*/
public static boolean isName(String name) {
return isOs(null, name, null, null);
}
/**
* Determines if the OS on which Ant is executing matches the
* given OS architecture.
*
* @param arch The arch.
* @return The result.
* @since 1.7
*/
public static boolean isArch(String arch) {
return isOs(null, null, arch, null);
}
/**
* Determines if the OS on which Ant is executing matches the
* given OS version.
*
* @param version The version.
* @return The result.
* @since 1.7
*/
public static boolean isVersion(String version) {
return isOs(null, null, null, version);
}
/**
* Determines if the OS on which Ant is executing matches the
* given OS family, name, architecture and version
*
* @param family The OS family
* @param name The OS name
* @param arch The OS architecture
* @param version The OS version
*
* @return The result.
* @since 1.7
*/
public static boolean isOs(String family, String name, String arch,
String version) {
boolean retValue = false;
if (family != null || name != null || arch != null
|| version != null) {
boolean isFamily = true;
boolean isName = true;
boolean isArch = true;
boolean isVersion = true;
if (family != null) {
if (family.equals("windows")) {
isFamily = osName.indexOf("windows") > -1;
} else if (family.equals("os/2")) {
isFamily = osName.indexOf("os/2") > -1;
} else if (family.equals("netware")) {
isFamily = osName.indexOf("netware") > -1;
} else if (family.equals("dos")) {
isFamily = pathSep.equals(";") && !isFamily("netware");
} else if (family.equals("mac")) {
isFamily = osName.indexOf("mac") > -1;
} else if (family.equals("unix")) {
isFamily = pathSep.equals(":")
&& (!isFamily("mac") || osName.endsWith("x"));
} else {
throw new RuntimeException(
"Don\'t know how to detect os family \""
+ family + "\"");
}
}
if (name != null) {
isName = name.equals(osName);
}
if (arch != null) {
isArch = arch.equals(osArch);
}
if (version != null) {
isVersion = version.equals(osVersion);
}
retValue = isFamily && isName && isArch && isVersion;
}
return retValue;
}
}

126
spring-erlang/src/main/java/org/springframework/util/exec/PumpStreamHandler.java
View File

@@ -1,126 +0,0 @@
/*
* Copyright 2002-2014 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.util.exec;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* Copies standard output and error of subprocesses to standard output and
* error of the parent process.
*
* TODO: standard input of the subprocess is not implemented.
*
* @author thomas.haas@softwired-inc.com
*/
public class PumpStreamHandler implements ExecuteStreamHandler {
private Thread inputThread;
private Thread errorThread;
private final OutputStream out, err;
boolean running=false;
public PumpStreamHandler(OutputStream out, OutputStream err) {
this.out = out;
this.err = err;
}
public PumpStreamHandler(OutputStream outAndErr) {
this(outAndErr, outAndErr);
}
public PumpStreamHandler() {
this(System.out, System.err);
}
@Override
public void setProcessOutputStream(InputStream is) {
createProcessOutputPump(is, out);
}
@Override
public void setProcessErrorStream(InputStream is) {
createProcessErrorPump(is, err);
}
@Override
public void setProcessInputStream(OutputStream os) {
}
@Override
public void start() {
inputThread.start();
errorThread.start();
running=true;
}
@Override
public void stop() {
if( !running ) {
return;
}
try {
inputThread.join(1000);
} catch(InterruptedException e) {}
try {
errorThread.join(1000);
} catch(InterruptedException e) {}
try {
err.flush();
} catch (IOException e) {}
try {
out.flush();
} catch (IOException e) {}
running=false;
}
protected OutputStream getErr() {
return err;
}
protected OutputStream getOut() {
return out;
}
protected void createProcessOutputPump(InputStream is, OutputStream os) {
inputThread = createPump(is, os);
}
protected void createProcessErrorPump(InputStream is, OutputStream os) {
errorThread = createPump(is, os);
}
/**
* Creates a stream pumper to copy the given input stream to the given output stream.
*
* @param is The input stream.
* @param os The output stream.
* @return The thread.
*/
protected Thread createPump(InputStream is, OutputStream os) {
final Thread result = new Thread(new StreamPumper(is, os));
result.setDaemon(true);
return result;
}
}

69
spring-erlang/src/main/java/org/springframework/util/exec/StreamPumper.java
View File

@@ -1,69 +0,0 @@
/*
* Copyright 2002-2013 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.util.exec;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* Copies all data from an input stream to an output stream.
*
* @author thomas.haas@softwired-inc.com
*/
public class StreamPumper implements Runnable {
// TODO: make SIZE and SLEEP instance variables.
// TODO: add a status flag to note if an error occured in run.
private static final int SLEEP = 5;
private static final int SIZE = 128;
private InputStream is;
private OutputStream os;
/**
* Create a new stream pumper.
*
* @param is input stream to read data from
* @param os output stream to write data to.
*/
public StreamPumper(InputStream is, OutputStream os) {
this.is = is;
this.os = os;
}
/**
* Copies data from the input stream to the output stream.
*
* Terminates as soon as the input stream is closed or an error occurs.
*/
public void run() {
final byte[] buf = new byte[SIZE];
int length;
try {
while ((length = is.read(buf)) > 0) {
os.write(buf, 0, length);
try {
Thread.sleep(SLEEP);
} catch (InterruptedException e) {}
}
} catch(IOException e) {}
}
}

4
spring-erlang/src/main/java/org/springframework/util/exec/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Provides classes supporting execution.
*/
package org.springframework.util.exec;

219
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java
View File

@@ -1,219 +0,0 @@
/*
* Copyright 2002-2015 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.admin;
import static org.junit.Assert.assertTrue;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.UnknownHostException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.amqp.rabbit.test.BrokerTestUtils;
import org.springframework.amqp.rabbit.test.EnvironmentAvailable;
import org.springframework.erlang.connection.SingleConnectionFactory;
import org.springframework.erlang.core.ErlangTemplate;
import com.ericsson.otp.erlang.OtpConnection;
import com.ericsson.otp.erlang.OtpErlangBinary;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpPeer;
import com.ericsson.otp.erlang.OtpSelf;
/**
* @author Mark Pollack
* @author Mark Fisher
* @author Chris Beams
* @author Dave Syer
* @author Gary Russell
* @author Artem Bilan
*/
public class JInterfaceIntegrationTests {
private static Log logger = LogFactory.getLog(JInterfaceIntegrationTests.class);
private static int counter;
private static final String NODE_NAME = "spring@localhost";
private OtpConnection connection = null;
private RabbitBrokerAdmin brokerAdmin;
@ClassRule
public static EnvironmentAvailable environment = new EnvironmentAvailable("BROKER_INTEGRATION_TEST");
@Before
public void init() {
brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin(NODE_NAME);
RabbitStatus status = brokerAdmin.getStatus();
if (!status.isRunning()) {
brokerAdmin.startBrokerApplication();
}
}
@After
public void close() {
if (connection != null) {
connection.close();
}
if (brokerAdmin != null) {
brokerAdmin.stopNode();
}
}
@Test
public void testRawApi() throws Exception {
OtpSelf self = new OtpSelf("rabbit-monitor");
String hostName = NODE_NAME;
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 = NODE_NAME;
SingleConnectionFactory cf = new SingleConnectionFactory(selfNodeName, peerNodeName);
cf.afterPropertiesSet();
ErlangTemplate template = new ErlangTemplate(cf);
template.afterPropertiesSet();
long number = (Long) template.executeAndConvertRpc("erlang", "abs", -161803399);
Assert.assertEquals(161803399, number);
cf.destroy();
}
@Test
public void testRawOtpConnect() throws Exception {
createConnection();
}
@Test
public void stressTest() throws Exception {
String cookie = readCookie();
logger.info("Cookie: " + cookie);
OtpConnection con = createConnection();
boolean recycleConnection = false;
for (int i = 0; i < 100; i++) {
executeRpc(con, recycleConnection, "rabbit", "status");
executeRpc(con, recycleConnection, "rabbit", "stop");
executeRpc(con, recycleConnection, "rabbit", "status");
executeRpc(con, recycleConnection, "rabbit", "start");
executeRpc(con, recycleConnection, "rabbit", "status");
if (i % 10 == 0) {
logger.debug("i = " + i);
}
}
}
@Test
public void testConcurrency() throws Exception {
SingleConnectionFactory cf = new SingleConnectionFactory("rabbit-monitor", NODE_NAME);
cf.afterPropertiesSet();
final ErlangTemplate template = new ErlangTemplate(cf);
template.afterPropertiesSet();
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (int i = 0; i < 10; i++) {
final int j = i;
executorService.execute(new Runnable() {
@Override
public void run() {
Assert.assertEquals((long) j, template.executeAndConvertRpc("erlang", "abs", -j));
}
});
}
executorService.shutdown();
assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
cf.destroy();
}
public OtpConnection createConnection() throws Exception {
OtpSelf self = new OtpSelf("rabbit-monitor-" + counter++);
OtpPeer peer = new OtpPeer(NODE_NAME);
return self.connect(peer);
}
private void executeRpc(OtpConnection con, boolean recycleConnection, String module, String function)
throws Exception, UnknownHostException {
con.sendRPC(module, function, new OtpErlangList());
OtpErlangObject response = con.receiveRPC();
logger.debug(module + " response received = " + response.toString());
if (recycleConnection) {
con.close();
con = createConnection();
}
}
private String readCookie() throws Exception {
String cookie = null;
final String dotCookieFilename = System.getProperty("user.home") + File.separator + ".erlang.cookie";
BufferedReader br = null;
try {
final File dotCookieFile = new File(dotCookieFilename);
br = new BufferedReader(new FileReader(dotCookieFile));
cookie = br.readLine().trim();
return cookie;
} finally {
try {
if (br != null) {
br.close();
}
} catch (final IOException e) {
}
}
}
}

176
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/QueueInfo.java
View File

@@ -1,176 +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.admin;
/**
* This class represents a Queue that is configured on the RabbitMQ broker
*
* @author Mark Pollack
*
*/
public class QueueInfo {
/*{app.stock.request={transactions=0, acks_uncommitted=0,
* consumers=0, pid=#Pid<rabbit@MARK6500.150.0>,
* durable=false, messages=0, memory=2320, auto_delete=false,
* messages_ready=0, arguments=[], name=app.stock.request,
* messages_unacknowledged=0, messages_uncommitted=0},
*/
private long transactions;
private long acksUncommitted;
private long consumers;
private String pid;
private boolean durable;
private long messages;
private long memory;
private boolean autoDelete;
private long messagesReady;
private String[] arguments;
private String name;
private long messagesUnacknowledged;
private long messageUncommitted;
public long getTransactions() {
return transactions;
}
public void setTransactions(long transations) {
this.transactions = transations;
}
public long getAcksUncommitted() {
return acksUncommitted;
}
public void setAcksUncommitted(long acksUncommitted) {
this.acksUncommitted = acksUncommitted;
}
public long getConsumers() {
return consumers;
}
public void setConsumers(long consumers) {
this.consumers = consumers;
}
public String getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
public boolean isDurable() {
return durable;
}
public void setDurable(boolean durable) {
this.durable = durable;
}
public long getMessages() {
return messages;
}
public void setMessages(long messages) {
this.messages = messages;
}
public long getMemory() {
return memory;
}
public void setMemory(long memory) {
this.memory = memory;
}
public boolean isAutoDelete() {
return autoDelete;
}
public void setAutoDelete(boolean autoDelete) {
this.autoDelete = autoDelete;
}
public long getMessagesReady() {
return messagesReady;
}
public void setMessagesReady(long messagesReady) {
this.messagesReady = messagesReady;
}
public String[] getArguments() {
return arguments;
}
public void setArguments(String[] args) {
this.arguments = args;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public long getMessagesUnacknowledged() {
return messagesUnacknowledged;
}
public void setMessagesUnacknowledged(long messagesUnacknowledged) {
this.messagesUnacknowledged = messagesUnacknowledged;
}
public long getMessageUncommitted() {
return messageUncommitted;
}
public void setMessageUncommitted(long messageUncommitted) {
this.messageUncommitted = messageUncommitted;
}
@Override
public String toString() {
return "QueueInfo [name=" + name + ", durable=" + durable
+ ", autoDelete=" + autoDelete + ", arguments=" + arguments
+ ", memory=" + memory + ", messages=" + messages
+ ", consumers=" + consumers + ", transations=" + transactions
+ ", acksUncommitted=" + acksUncommitted + ", messagesReady="
+ messagesReady + ", messageUncommitted=" + messageUncommitted
+ ", messagesUnacknowledged=" + messagesUnacknowledged
+ ", pid=" + pid + "]";
}
}

12
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitAdminAuthException.java
View File

@@ -1,12 +0,0 @@
package org.springframework.amqp.rabbit.admin;
import org.springframework.erlang.OtpAuthException;
@SuppressWarnings("serial")
public class RabbitAdminAuthException extends OtpAuthException {
public RabbitAdminAuthException(String message, OtpAuthException cause) {
super(message, (com.ericsson.otp.erlang.OtpAuthException) cause.getCause());
}
}

713
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java
View File

@@ -1,713 +0,0 @@
/*
* 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.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.Map;
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.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.erlang.OtpAuthException;
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.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.exec.Execute;
import org.springframework.util.exec.Os;
/**
* Rabbit broker administration. Features:
*
* <ul>
* <li>Basic AMQP admin commands are provided, like declaring queues, exchanges and bindings.</li>
* <li>Manage user accounts and virtual hosts.</li>
* <li>Start and stop the broker process.</li>
* <li>Start and stop the broker application (in a running process).</li>
* <li>Inspect and manage the queues, e.g. listing message counts etc.</li>
* </ul>
*
* Depending on your platform, to {@link #startNode() start the broker} you might need to set some environment
* properties. The most common are available via constructors or setters in this class (e.g.
* {@link #setRabbitLogBaseDirectory(String) RABBITMQ_LOG_BASE}). All others you can set via the OS (any setting that
* RabbtMQ allows in its startup script), and some work via System properties as special convenience cases (
* <code>ERLANG_HOME</code> and <code>RABBITMQ_HOME</code> ).
*
* @author Mark Pollack
* @author Dave Syer
* @author Helena Edelson
*/
public class RabbitBrokerAdmin implements RabbitBrokerOperations {
private static final String DEFAULT_VHOST = "/";
private static String DEFAULT_NODE_NAME;
private static final int DEFAULT_PORT = 5672;
private static final String DEFAULT_ENCODING = "UTF-8";
/** Logger available to subclasses */
protected final Log logger = LogFactory.getLog(getClass());
private ErlangTemplate erlangTemplate;
private String encoding = DEFAULT_ENCODING;
private long timeout = 0;
private AsyncTaskExecutor executor;
private final String nodeName;
private final String cookie;
private final int port;
private String rabbitLogBaseDirectory;
private String rabbitMnesiaBaseDirectory;
private Map<String, String> moduleAdapter = new HashMap<String, String>();
static {
try {
String hostName = InetAddress.getLocalHost().getHostName();
DEFAULT_NODE_NAME = "rabbit@" + hostName;
} catch (UnknownHostException e) {
DEFAULT_NODE_NAME = "rabbit@localhost";
}
}
public RabbitBrokerAdmin() {
this(DEFAULT_NODE_NAME);
}
/**
* Create an instance by supplying the erlang node name (e.g. "rabbit@myserver"), or simply the hostname (if the
* alive name is "rabbit").
*
* @param nodeName the node name or hostname to use
*/
public RabbitBrokerAdmin(String nodeName) {
this(nodeName, null);
}
/**
* Create an instance by supplying the erlang node name and cookie (unique string).
*
* @param nodeName the node name or hostname to use
*
* @param cookie the cookie value to use
*/
public RabbitBrokerAdmin(String nodeName, String cookie) {
this(nodeName, DEFAULT_PORT, cookie);
}
/**
* Create an instance by supplying the erlang node name and port number. Use this on a UN*X system if you want to
* run the broker as a user without root privileges, supplying values that do not clash with the default broker
* (usually "rabbit@&lt;servername&gt;" and 5672). If, as well as managing an existing broker, you need to start the
* broker process, you will also need to set {@link #setRabbitLogBaseDirectory(String) RABBITMQ_LOG_BASE} and
* {@link #setRabbitMnesiaBaseDirectory(String) RABBITMQ_MNESIA_BASE} to point to writable directories).
*
* @param nodeName the node name or hostname to use
* @param port the port number (overriding the default which is 5672)
*/
public RabbitBrokerAdmin(String nodeName, int port) {
this(nodeName, port, null);
}
/**
* Create an instance by supplying the erlang node name, port number and cookie (unique string). If the node name
* does not contain an <code>@</code> character, it will be prepended with an alivename <code>rabbit@</code>
* (interpreting the supplied value as just the hostname).
*
* @param nodeName the node name or hostname to use
* @param port the port number (overriding the default which is 5672)
* @param cookie the cookie value to use
*/
public RabbitBrokerAdmin(String nodeName, int port, String cookie) {
if (!nodeName.contains("@")) {
nodeName = "rabbit@" + nodeName; // it was just the host
}
String[] parts = nodeName.split("@");
Assert.state(parts.length == 2, "The node name should be in the form alivename@host, e.g. rabbit@myserver");
if (Os.isFamily("windows") && !DEFAULT_NODE_NAME.equals(nodeName)) {
nodeName = parts[0] + "@" + parts[1].toUpperCase();
}
this.port = port;
this.cookie = cookie;
this.nodeName = nodeName;
SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
executor.setDaemon(true);
this.executor = executor;
}
/**
* An async task executor for launching background processing when starting or stopping the broker.
*
* @param executor the executor to set
*/
public void setExecutor(AsyncTaskExecutor executor) {
this.executor = executor;
}
/**
* The location of <code>RABBITMQ_LOG_BASE</code> to override the system default (which may be owned by another
* user). Only needed for launching the broker process. Can also be set as a system property.
*
* @param rabbitLogBaseDirectory the rabbit log base directory to set
*/
public void setRabbitLogBaseDirectory(String rabbitLogBaseDirectory) {
this.rabbitLogBaseDirectory = rabbitLogBaseDirectory;
}
/**
* The location of <code>RABBITMQ_MNESIA_BASE</code> to override the system default (which may be owned by another
* user). Only needed for launching the broker process. Can also be set as a system property.
*
* @param rabbitMnesiaBaseDirectory the rabbit Mnesia base directory to set
*/
public void setRabbitMnesiaBaseDirectory(String rabbitMnesiaBaseDirectory) {
this.rabbitMnesiaBaseDirectory = rabbitMnesiaBaseDirectory;
}
/**
* 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)
*/
public void setEncoding(String encoding) {
this.encoding = encoding;
}
/**
* 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;
}
/**
* Allows users to adapt Erlang RPC <code>(module, function)</code> pairs to older, or different, versions of the
* broker than the current target. The map is from String to String in the form
* <code>input_module%input_function -> output_module%output_function</code> (using a <code>%</code> separator).
*
* @param moduleAdapter the module adapter to set
*/
public void setModuleAdapter(Map<String, String> moduleAdapter) {
this.moduleAdapter = moduleAdapter;
}
@SuppressWarnings("unchecked")
public List<QueueInfo> getQueues() {
return (List<QueueInfo>) executeAndConvertRpc("rabbit_amqqueue", "info_all", getBytes(DEFAULT_VHOST));
}
@SuppressWarnings("unchecked")
public List<QueueInfo> getQueues(String virtualHost) {
return (List<QueueInfo>) executeAndConvertRpc("rabbit_amqqueue", "info_all", getBytes(virtualHost));
}
// User management
@ManagedOperation()
public void addUser(String username, String password) {
executeAndConvertRpc("rabbit_auth_backend_internal", "add_user", getBytes(username), getBytes(password));
}
@ManagedOperation
public void deleteUser(String username) {
executeAndConvertRpc("rabbit_auth_backend_internal", "delete_user", getBytes(username));
}
@ManagedOperation
public void changeUserPassword(String username, String newPassword) {
executeAndConvertRpc("rabbit_auth_backend_internal", "change_password", getBytes(username),
getBytes(newPassword));
}
@SuppressWarnings("unchecked")
@ManagedOperation
public List<String> listUsers() {
return (List<String>) executeAndConvertRpc("rabbit_auth_backend_internal", "list_users");
}
public int addVhost(String vhostPath) {
// TODO Auto-generated method stub
return 0;
}
public int deleteVhost(String vhostPath) {
// TODO Auto-generated method stub
return 0;
}
public void setPermissions(String username, Pattern configure, Pattern read, Pattern write) {
// TODO Auto-generated method stub
}
public void setPermissions(String username, Pattern configure, Pattern read, Pattern write, String vhostPath) {
// TODO Auto-generated method stub
}
public void clearPermissions(String username) {
// TODO Auto-generated method stub
}
public void clearPermissions(String username, String vhostPath) {
// TODO Auto-generated method stub
}
public List<String> listPermissions() {
// TODO Auto-generated method stub
return null;
}
public List<String> listPermissions(String vhostPath) {
// TODO Auto-generated method stub
return null;
}
public List<String> listUserPermissions(String username) {
// TODO Auto-generated method stub
return null;
}
@ManagedOperation
public void startBrokerApplication() {
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 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.info("Stopping Rabbit Application.");
executeAndConvertRpc("rabbit", "stop");
if (timeout > 0) {
waitForUnreadyState();
}
}
@ManagedOperation
public void startNode() {
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 = "sbin/rabbitmq-server.bat";
} else if (Os.isFamily("unix") || Os.isFamily("mac")) {
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"));
if (rabbitHome == null) {
if (Os.isFamily("windows") || Os.isFamily("dos")) {
rabbitHome = findDirectoryName("c:/Program Files", "rabbitmq");
} else if (Os.isFamily("unix") || Os.isFamily("mac")) {
rabbitHome = "/usr/lib/rabbitmq";
}
}
Assert.notNull(rabbitHome, "RABBITMQ_HOME system property (or environment variable) not set.");
rabbitHome = StringUtils.cleanPath(rabbitHome);
String rabbitStartCommand = rabbitHome + "/" + rabbitStartScript;
String[] commandline = new String[] { rabbitStartCommand };
List<String> env = new ArrayList<String>();
if (rabbitLogBaseDirectory != null) {
env.add("RABBITMQ_LOG_BASE=" + rabbitLogBaseDirectory);
} else {
addEnvironment(env, "RABBITMQ_LOG_BASE");
}
if (rabbitMnesiaBaseDirectory != null) {
env.add("RABBITMQ_MNESIA_BASE=" + rabbitMnesiaBaseDirectory);
} else {
addEnvironment(env, "RABBITMQ_MNESIA_BASE");
}
addEnvironment(env, "ERLANG_HOME");
// Make the nodename explicitly the same so the erl process knows who we are
env.add("RABBITMQ_NODENAME=" + nodeName);
// Set the port number for the new process
env.add("RABBITMQ_NODE_PORT=" + port);
// Ask for a detached erl process so stdout doesn't get diverted to a black hole when the JVM dies (without this
// you can start the Rabbit broker form Java but if you forget to stop it, the erl process is hosed).
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);
final String errorHint = hint;
executor.execute(new Runnable() {
public void run() {
try {
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);
}
}
});
try {
logger.info("Waiting for Rabbit process to be started");
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.");
}
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;
}
/**
* Find a directory whose name starts with a substring in a given parent directory. If there is none return null,
* otherwise sort the results and return the best match (an exact match if there is one or the last one in a lexical
* sort).
*
* @param parent
* @param child
* @return the full name of a directory
*/
private String findDirectoryName(String parent, String child) {
String result = null;
String[] names = new File(parent).list(new FilenameFilter() {
public boolean accept(File dir, String name) {
return name.equals("rabbitmq") && new File(dir, name).isDirectory();
}
});
if (names.length == 1) {
result = new File(parent, names[0]).getAbsolutePath();
return result;
}
List<String> sorted = Arrays.asList(new File(parent).list(new FilenameFilter() {
public boolean accept(File dir, String name) {
return name.startsWith("rabbitmq") && new File(dir, name).isDirectory();
}
}));
Collections.sort(sorted, Collections.reverseOrder());
if (!sorted.isEmpty()) {
result = new File(parent, sorted.get(0)).getAbsolutePath();
}
return result;
}
private void addEnvironment(List<String> env, String key) {
String value = System.getProperty(key);
if (value != null) {
logger.debug("Adding environment variable: " + key + "=" + value);
env.add(key + "=" + value);
}
}
@ManagedOperation
public void stopNode() {
logger.info("Stopping RabbitMQ node.");
try {
executeAndConvertRpc("rabbit", "stop_and_halt");
} catch (Exception e) {
logger.error("Failed to send stop signal", e);
}
if (timeout >= 0) {
waitForStoppedState();
}
}
@ManagedOperation
public void resetNode() {
executeAndConvertRpc("rabbit_mnesia", "reset");
}
@ManagedOperation
public void forceResetNode() {
executeAndConvertRpc("rabbit_mnesia", "force_reset");
}
@ManagedOperation
public RabbitStatus getStatus() {
try {
return (RabbitStatus) executeAndConvertRpc("rabbit", "status");
} 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());
}
}
protected void initializeDefaultErlangTemplate() {
String peerNodeName = nodeName;
logger.debug("Creating jinterface connection with peerNodeName = [" + peerNodeName + "]");
SimpleConnectionFactory otpConnectionFactory = new SimpleConnectionFactory("rabbit-spring-monitor",
peerNodeName, this.cookie);
otpConnectionFactory.afterPropertiesSet();
createErlangTemplate(otpConnectionFactory);
}
protected void createErlangTemplate(ConnectionFactory otpConnectionFactory) {
erlangTemplate = new ErlangTemplate(otpConnectionFactory);
erlangTemplate.setErlangConverter(new RabbitControlErlangConverter(moduleAdapter));
erlangTemplate.afterPropertiesSet();
}
/**
* Convenience method for lazy initialization of the {@link ErlangTemplate} and associated trimmings. All RPC calls
* should go through this method.
*
* @param <T> the type of the result
* @param module the module to address remotely
* @param function the function to call
* @param args the arguments to pass
*
* @return the result from the remote erl process converted to the correct type
*/
@SuppressWarnings("unchecked")
private <T> T executeAndConvertRpc(String module, String function, Object... args) {
if (erlangTemplate == null) {
synchronized (this) {
if (erlangTemplate == null) {
initializeDefaultErlangTemplate();
}
}
}
String key = module + "%" + function;
if (moduleAdapter.containsKey(key)) {
String adapter = moduleAdapter.get(key);
String[] values = adapter.split("%");
Assert.state(values.length == 2,
"The module adapter should be a map from 'module%function' to 'module%function'. "
+ "This one contained [" + adapter + "] which cannot be parsed to a module, function pair.");
module = values[0];
function = values[1];
}
return (T) erlangTemplate.executeAndConvertRpc(module, function, args);
}
/**
* 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);
}
}

138
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java
View File

@@ -1,138 +0,0 @@
/*
* Copyright 2002-2014 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.admin;
import static org.junit.Assert.assertEquals;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.log4j.Level;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.core.Queue;
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.BrokerTestUtils;
import org.springframework.amqp.rabbit.test.EnvironmentAvailable;
import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
/**
*
* @author Mark Pollack
* @author Dave Syer
* @author Helena Edelson
* @author Gunnar Hillert
*/
public class RabbitBrokerAdminIntegrationTests {
@Rule
public Log4jLevelAdjuster logLevel = new Log4jLevelAdjuster(Level.INFO, RabbitBrokerAdmin.class);
@ClassRule
public static EnvironmentAvailable environment = new EnvironmentAvailable("BROKER_INTEGRATION_TEST");
/*
* Ensure broker dies if a test fails (otherwise the erl process might have to be killed manually)
*/
@ClassRule
public static BrokerPanic panic = new BrokerPanic();
private static RabbitBrokerAdmin brokerAdmin;
@BeforeClass
public static void start() throws Exception {
if (environment.isActive()) {
// Set up broker admin for non-root user
brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin();
brokerAdmin.startNode();
panic.setBrokerAdmin(brokerAdmin);
}
}
@AfterClass
public static void stop() throws Exception {
if (environment.isActive()) {
brokerAdmin.stopNode();
}
}
@Test
public void integrationTestsUserCrud() throws Exception {
List<String> users = brokerAdmin.listUsers();
if (users.contains("joe")) {
brokerAdmin.deleteUser("joe");
}
Thread.sleep(200L);
brokerAdmin.addUser("joe", "trader");
Thread.sleep(200L);
brokerAdmin.changeUserPassword("joe", "sales");
Thread.sleep(200L);
users = brokerAdmin.listUsers();
if (users.contains("joe")) {
Thread.sleep(200L);
brokerAdmin.deleteUser("joe");
}
}
@Test
public void integrationTestsUserCrudWithModuleAdapter() throws Exception {
Map<String, String> adapter = new HashMap<String, String>();
// Switch two functions with identical inputs!
adapter.put("rabbit_auth_backend_internal%add_user", "rabbit_auth_backend_internal%change_password");
adapter.put("rabbit_auth_backend_internal%change_password", "rabbit_auth_backend_internal%add_user");
brokerAdmin.setModuleAdapter(adapter);
List<String> users = brokerAdmin.listUsers();
if (users.contains("joe")) {
brokerAdmin.deleteUser("joe");
}
Thread.sleep(200L);
brokerAdmin.changeUserPassword("joe", "sales");
Thread.sleep(200L);
brokerAdmin.addUser("joe", "trader");
Thread.sleep(200L);
users = brokerAdmin.listUsers();
if (users.contains("joe")) {
Thread.sleep(200L);
brokerAdmin.deleteUser("joe");
}
}
@Test
public void testGetEmptyQueues() throws Exception {
List<QueueInfo> queues = brokerAdmin.getQueues();
assertEquals(0, queues.size());
}
@Test
public void testGetQueues() throws Exception {
SingleConnectionFactory connectionFactory = new SingleConnectionFactory();
connectionFactory.setPort(BrokerTestUtils.getAdminPort());
Queue queue = new RabbitAdmin(connectionFactory).declareQueue();
assertEquals("/", connectionFactory.getVirtualHost());
List<QueueInfo> queues = brokerAdmin.getQueues();
assertEquals(queue.getName(), queues.get(0).getName());
connectionFactory.destroy();
}
}

138
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminLifecycleIntegrationTests.java
View File

@@ -1,138 +0,0 @@
/*
* Copyright 2002-2014 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.admin;
import static org.junit.Assert.assertEquals;
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.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.log4j.Level;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.rabbit.test.BrokerTestUtils;
import org.springframework.amqp.rabbit.test.EnvironmentAvailable;
import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster;
import org.springframework.erlang.OtpException;
/**
* @author Mark Pollack
* @author Dave Syer
* @author Gary Russell
*/
public class RabbitBrokerAdminLifecycleIntegrationTests {
private static Log logger = LogFactory.getLog(RabbitBrokerAdminLifecycleIntegrationTests.class);
private static final String NODE_NAME = "spring@localhost";
@Rule
public Log4jLevelAdjuster logLevel = new Log4jLevelAdjuster(Level.INFO, RabbitBrokerAdmin.class);
@ClassRule
public static EnvironmentAvailable environment = new EnvironmentAvailable("BROKER_INTEGRATION_TEST");
@Before
public void init() throws Exception {
FileUtils.deleteDirectory(new File("target/rabbitmq"));
}
@Test
public void testStartNode() throws Exception {
// Set up broker admin for non-root user
final RabbitBrokerAdmin brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin(NODE_NAME);
RabbitStatus status = brokerAdmin.getStatus();
try {
// Stop it if it is already running
if (status.isReady()) {
brokerAdmin.stopBrokerApplication();
Thread.sleep(1000L);
}
} catch (OtpException e) {
// Not useful for test.
}
status = brokerAdmin.getStatus();
if (!status.isRunning()) {
brokerAdmin.startBrokerApplication();
}
status = brokerAdmin.getStatus();
try {
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();
}
}
@Test
public void testStopAndStartBroker() throws Exception {
// Set up broker admin for non-root user
final RabbitBrokerAdmin brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin(NODE_NAME);
RabbitStatus status = brokerAdmin.getStatus();
status = brokerAdmin.getStatus();
if (!status.isRunning()) {
brokerAdmin.startBrokerApplication();
}
brokerAdmin.stopBrokerApplication();
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++) {
testStopAndStartBroker();
Thread.sleep(200);
if (i % 5 == 0) {
logger.debug("i = " + i);
}
}
}
/**
* Asserts that the named-node is running.
* @param status
*/
private void assertBrokerAppRunning(RabbitStatus status) {
assertEquals(1, status.getRunningNodes().size());
assertTrue(status.getRunningNodes().get(0).getName().contains(NODE_NAME));
}
}

112
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerOperations.java
View File

@@ -1,112 +0,0 @@
/*
* 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.admin;
import java.util.List;
import java.util.regex.Pattern;
/**
* 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.
*
* @author Mark Pollack
*/
public interface RabbitBrokerOperations {
// Queue operations
public List<QueueInfo> getQueues();
public List<QueueInfo> getQueues(String virtualHost);
// User management
void addUser(String username, String password);
void deleteUser(String username);
void changeUserPassword(String username, String newPassword);
List<String> listUsers();
// VHost management
int addVhost(String vhostPath);
int deleteVhost(String vhostPath);
// permissions
void setPermissions(String username, Pattern configure, Pattern read, Pattern write);
void setPermissions(String username, Pattern configure, Pattern read, Pattern write, String vhostPath);
void clearPermissions(String username);
void clearPermissions(String username, String vhostPath);
List<String> listPermissions();
List<String> listPermissions(String vhostPath);
List<String> listUserPermissions(String username);
// Start/Stop/Reset broker
/**
* 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.
*/
void startBrokerApplication();
/**
* Stops the RabbitMQ application, leaving the Erlang node running.
*/
void stopBrokerApplication();
/**
* Starts the Erlang node where RabbitMQ is running by shelling out to the directory specified by RABBITMQ_HOME and
* executing the standard named start script. It spawns the shell command execution into its own thread.
*/
void startNode();
/**
* Stops the halts the Erlang node on which RabbitMQ is running. To restart the node you will need to execute the
* start script from a command line or via other means.
*/
void stopNode();
/**
* Removes the node from any cluster it belongs to, removes all data from the management database, such as
* configured users and vhosts, and deletes all persistent messages. <p> For {@link #resetNode} and
* {@link #forceResetNode} to succeed the RabbitMQ application must have been stopped, e.g.
* {@link #stopBrokerApplication}
*/
void resetNode();
/**
* The forceResetNode command differs from {@link #resetNode} in that it resets the node unconditionally, regardless
* of the current management database state and cluster configuration. It should only be used as a last resort if
* the database or cluster configuration has been corrupted. <p> For {@link #resetNode} and {@link #forceResetNode}
* to succeed the RabbitMQ application must have been stopped, e.g. {@link #stopBrokerApplication}
*/
void forceResetNode();
/**
* Returns the status of the node.
* @return status of the node.
*/
RabbitStatus getStatus();
}

269
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitControlErlangConverter.java
View File

@@ -1,269 +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.admin;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.erlang.core.Application;
import org.springframework.erlang.core.Node;
import org.springframework.erlang.support.converter.ErlangConversionException;
import org.springframework.erlang.support.converter.ErlangConverter;
import org.springframework.erlang.support.converter.SimpleErlangConverter;
import org.springframework.util.Assert;
import com.ericsson.otp.erlang.OtpErlangAtom;
import com.ericsson.otp.erlang.OtpErlangBinary;
import com.ericsson.otp.erlang.OtpErlangList;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpErlangTuple;
/***
* Converter that understands the responses from the rabbit control module and related functionality.
*
* @author Mark Pollack
* @author Mark Fisher
* @author Helena Edelson
*/
public class RabbitControlErlangConverter extends SimpleErlangConverter implements ErlangConverter {
protected final Log logger = LogFactory.getLog(getClass());
private final Map<String, ErlangConverter> converterMap = new HashMap<String, ErlangConverter>();
private final Map<String, String> moduleAdapter;
public RabbitControlErlangConverter(Map<String, String> moduleAdapter) {
this.moduleAdapter = moduleAdapter;
initializeConverterMap();
}
public Object fromErlangRpc(String module, String function, OtpErlangObject erlangObject)
throws ErlangConversionException {
ErlangConverter converter = getConverter(module, function);
if (converter != null) {
return converter.fromErlang(erlangObject);
} else {
return super.fromErlangRpc(module, function, erlangObject);
}
}
protected ErlangConverter getConverter(String module, String function) {
return converterMap.get(generateKey(module, function));
}
protected void initializeConverterMap() {
registerConverter("rabbit_auth_backend_internal", "list_users", new ListUsersConverter());
registerConverter("rabbit", "status", new StatusConverter());
registerConverter("rabbit_amqqueue", "info_all", new QueueInfoAllConverter());
}
protected void registerConverter(String module, String function, ErlangConverter listUsersConverter) {
String key = generateKey(module, function);
if (moduleAdapter.containsKey(key)) {
String adapter = moduleAdapter.get(key);
String[] values = adapter.split("%");
Assert.state(values.length == 2,
"The module adapter should be a map from 'module%function' to 'module%function'. "
+ "This one contained [" + adapter + "] which cannot be parsed to a module, function pair.");
module = values[0];
function = values[1];
}
converterMap.put(generateKey(module, function), listUsersConverter);
}
protected String generateKey(String module, String function) {
return module + "%" + function;
}
public class ListUsersConverter extends SimpleErlangConverter {
public Object fromErlang(OtpErlangObject erlangObject) throws ErlangConversionException {
List<String> users = new ArrayList<String>();
if (erlangObject instanceof OtpErlangList) {
OtpErlangList erlangList = (OtpErlangList) erlangObject;
for (OtpErlangObject obj : erlangList) {
String value = extractString(obj);
if (value != null) {
users.add(value);
}
}
}
return users;
}
private String extractString(OtpErlangObject obj) {
if (obj instanceof OtpErlangBinary) {
OtpErlangBinary binary = (OtpErlangBinary) obj;
return new String(binary.binaryValue());
} else if (obj instanceof OtpErlangTuple) {
OtpErlangTuple tuple = (OtpErlangTuple) obj;
return extractString(tuple.elementAt(0));
}
return null;
}
}
public class StatusConverter extends SimpleErlangConverter {
public Object fromErlang(OtpErlangObject erlangObject) throws ErlangConversionException {
List<Application> applications = new ArrayList<Application>();
List<Node> nodes = new ArrayList<Node>();
List<Node> runningNodes = new ArrayList<Node>();
if (erlangObject instanceof OtpErlangList) {
OtpErlangList erlangList = (OtpErlangList) erlangObject;
OtpErlangTuple runningAppTuple = (OtpErlangTuple) erlangList.elementAt(0);
OtpErlangList appList = (OtpErlangList) runningAppTuple.elementAt(1);
extractApplications(applications, appList);
OtpErlangTuple nodesTuple = (OtpErlangTuple) erlangList.elementAt(1);
OtpErlangList nodesList = (OtpErlangList) nodesTuple.elementAt(1);
extractNodes(nodes, nodesList);
OtpErlangTuple runningNodesTuple = (OtpErlangTuple) erlangList.elementAt(2);
nodesList = (OtpErlangList) runningNodesTuple.elementAt(1);
extractNodes(runningNodes, nodesList);
/*
* for (OtpErlangObject obj : erlangList) { if (obj instanceof OtpErlangBinary) { OtpErlangBinary binary
* = (OtpErlangBinary) obj; users.add(new String(binary.binaryValue())); } }
*/
}
return new RabbitStatus(applications, nodes, runningNodes);
}
private void extractNodes(List<Node> nodes, OtpErlangList nodesList) {
for (OtpErlangObject erlangNodeName : nodesList) {
String nodeName = erlangNodeName.toString();
nodes.add(new Node(nodeName));
}
}
private void extractApplications(List<Application> applications, OtpErlangList appList) {
for (OtpErlangObject appDescription : appList) {
OtpErlangTuple appDescriptionTuple = (OtpErlangTuple) appDescription;
String name = appDescriptionTuple.elementAt(0).toString();
String description = appDescriptionTuple.elementAt(1).toString();
String version = appDescriptionTuple.elementAt(2).toString();
applications.add(new Application(name, description, version));
}
}
}
public enum QueueInfoField {
transactions, acks_uncommitted, consumers, pid, durable, messages, memory, auto_delete, messages_ready, arguments, name, messages_unacknowledged, messages_uncommitted, NOVALUE;
public static QueueInfoField toQueueInfoField(String str) {
try {
return valueOf(str);
} catch (Exception ex) {
return NOVALUE;
}
}
}
public class QueueInfoAllConverter extends SimpleErlangConverter {
@Override
public Object fromErlang(OtpErlangObject erlangObject) throws ErlangConversionException {
List<QueueInfo> queueInfoList = new ArrayList<QueueInfo>();
if (erlangObject instanceof OtpErlangList) {
OtpErlangList erlangList = (OtpErlangList) erlangObject;
for (OtpErlangObject element : erlangList.elements()) {
QueueInfo queueInfo = new QueueInfo();
OtpErlangList itemList = (OtpErlangList) element;
for (OtpErlangObject item : itemList.elements()) {
OtpErlangTuple tuple = (OtpErlangTuple) item;
if (tuple.arity() == 2) {
String key = tuple.elementAt(0).toString();
OtpErlangObject value = tuple.elementAt(1);
switch (QueueInfoField.toQueueInfoField(key)) {
case name:
queueInfo.setName(extractNameValueFromTuple((OtpErlangTuple) value));
break;
case transactions:
queueInfo.setTransactions(extractLong(value));
break;
case acks_uncommitted:
queueInfo.setAcksUncommitted(extractLong(value));
break;
case consumers:
queueInfo.setConsumers(extractLong(value));
break;
case pid:
queueInfo.setPid(extractPid(value));
break;
case durable:
queueInfo.setDurable(extractAtomBoolean(value));
break;
case messages:
queueInfo.setMessages(extractLong(value));
break;
case memory:
queueInfo.setMemory(extractLong(value));
break;
case auto_delete:
queueInfo.setAutoDelete(extractAtomBoolean(value));
break;
case messages_ready:
queueInfo.setMessagesReady(extractLong(value));
break;
case arguments:
OtpErlangList list = (OtpErlangList) value;
if (list != null) {
String[] args = new String[list.arity()];
for (int i = 0; i < list.arity(); i++) {
OtpErlangObject obj = list.elementAt(i);
args[i] = obj.toString();
}
queueInfo.setArguments(args);
}
break;
case messages_unacknowledged:
queueInfo.setMessagesUnacknowledged(extractLong(value));
break;
case messages_uncommitted:
queueInfo.setMessageUncommitted(extractLong(value));
break;
default:
break;
}
}
}
queueInfoList.add(queueInfo);
}
}
return queueInfoList;
}
private boolean extractAtomBoolean(OtpErlangObject value) {
return ((OtpErlangAtom) value).booleanValue();
}
private String extractNameValueFromTuple(OtpErlangTuple value) {
Object nameElement = value.elementAt(3);
return new String(((OtpErlangBinary) nameElement).binaryValue());
}
}
}

87
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/admin/RabbitStatus.java
View File

@@ -1,87 +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.admin;
import java.io.Serializable;
import java.util.List;
import org.springframework.erlang.core.Application;
import org.springframework.erlang.core.Node;
/**
* The status object returned from querying the broker
*
* @author Mark Pollack
*
*/
@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();
this.runningApplications = runningApplications;
this.nodes = nodes;
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;
}
public List<Node> getNodes() {
return nodes;
}
public List<Node> getRunningNodes() {
return runningNodes;
}
@Override
public String toString() {
return "RabbitStatus [runningApplications=" + runningApplications
+ ", runningNodes=" + runningNodes + ", nodes=" + nodes + "]";
}
}

205
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerBrokerInterruptionIntegrationTests.java
View File

@@ -1,205 +0,0 @@
/*
* Copyright 2010-2014 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.listener;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.admin.QueueInfo;
import org.springframework.amqp.rabbit.admin.RabbitBrokerAdmin;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.test.BrokerPanic;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.amqp.rabbit.test.BrokerTestUtils;
import org.springframework.amqp.rabbit.test.EnvironmentAvailable;
import com.rabbitmq.client.Channel;
/**
* @author Dave Syer
* @author Gunnar Hillert
* @author Gary Russell
* @since 1.0
*
*/
public class MessageListenerBrokerInterruptionIntegrationTests {
private static Log logger = LogFactory.getLog(MessageListenerBrokerInterruptionIntegrationTests.class);
// Ensure queue is durable, or it won't survive the broker restart
private final Queue queue = new Queue("test.queue", true);
private final int concurrentConsumers = 2;
private final int messageCount = 60;
private final int txSize = 1;
private final boolean transactional = false;
private final AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
private SimpleMessageListenerContainer container;
@ClassRule
public static EnvironmentAvailable environment = new EnvironmentAvailable("BROKER_INTEGRATION_TEST");
/*
* Ensure broker dies if a test fails (otherwise the erl process might have to be killed manually)
*/
@ClassRule
public static BrokerPanic panic = new BrokerPanic();
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue);
private CachingConnectionFactory connectionFactory;
private final RabbitBrokerAdmin brokerAdmin;
public MessageListenerBrokerInterruptionIntegrationTests() throws Exception {
FileUtils.deleteDirectory(new File("target/rabbitmq"));
brokerIsRunning.setPort(BrokerTestUtils.getAdminPort());
logger.debug("Setting up broker");
brokerAdmin = BrokerTestUtils.getRabbitBrokerAdmin();
panic.setBrokerAdmin(brokerAdmin);
if (environment.isActive()) {
brokerAdmin.startNode();
}
}
@Before
public void createConnectionFactory() {
if (environment.isActive()) {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
connectionFactory.setChannelCacheSize(concurrentConsumers);
connectionFactory.setPort(BrokerTestUtils.getAdminPort());
this.connectionFactory = connectionFactory;
}
}
@After
public void clear() throws Exception {
if (environment.isActive()) {
// Wait for broker communication to finish before trying to stop container
Thread.sleep(300L);
logger.debug("Shutting down at end of test");
if (container != null) {
container.shutdown();
}
brokerAdmin.stopNode();
// Remove all trace of the durable queue...
FileUtils.deleteDirectory(new File("target/rabbitmq"));
if (this.connectionFactory != null) {
this.connectionFactory.destroy();
}
}
}
@Test
public void testListenerRecoversFromDeadBroker() throws Exception {
List<QueueInfo> queues = brokerAdmin.getQueues();
logger.info("Queues: " + queues);
assertEquals(1, queues.size());
assertTrue(queues.get(0).isDurable());
RabbitTemplate template = new RabbitTemplate(connectionFactory);
CountDownLatch latch = new CountDownLatch(messageCount);
assertEquals("No more messages to receive before even sent!", messageCount, latch.getCount());
container = createContainer(queue.getName(), new VanillaListener(latch), connectionFactory);
for (int i = 0; i < messageCount; i++) {
template.convertAndSend(queue.getName(), i + "foo");
}
assertTrue("No more messages to receive before broker stopped", latch.getCount() > 0);
brokerAdmin.stopBrokerApplication();
assertTrue("No more messages to receive after broker stopped", latch.getCount() > 0);
boolean waited = latch.await(500, TimeUnit.MILLISECONDS);
assertFalse("Did not time out waiting for message", waited);
container.stop();
assertEquals(0, container.getActiveConsumerCount());
brokerAdmin.startBrokerApplication();
queues = brokerAdmin.getQueues();
logger.info("Queues: " + queues);
container.start();
assertEquals(concurrentConsumers, container.getActiveConsumerCount());
int timeout = Math.min(4 + messageCount / (4 * concurrentConsumers), 30);
logger.debug("Waiting for messages with timeout = " + timeout + " (s)");
waited = latch.await(timeout, TimeUnit.SECONDS);
assertTrue("Timed out waiting for message", waited);
assertNull(template.receiveAndConvert(queue.getName()));
}
private SimpleMessageListenerContainer createContainer(String queueName, Object listener,
ConnectionFactory connectionFactory) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setMessageListener(new MessageListenerAdapter(listener));
container.setQueueNames(queueName);
container.setTxSize(txSize);
container.setPrefetchCount(txSize);
container.setConcurrentConsumers(concurrentConsumers);
container.setChannelTransacted(transactional);
container.setAcknowledgeMode(acknowledgeMode);
container.afterPropertiesSet();
container.start();
return container;
}
public static class VanillaListener implements ChannelAwareMessageListener {
private final CountDownLatch latch;
public VanillaListener(CountDownLatch latch) {
this.latch = latch;
}
@Override
public void onMessage(Message message, Channel channel) throws Exception {
String value = new String(message.getBody());
logger.debug("Receiving: " + value);
latch.countDown();
}
}
}

54
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerPanic.java
View File

@@ -1,54 +0,0 @@
/*
* Copyright 2002-2014 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.TestWatcher;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import org.springframework.amqp.rabbit.admin.RabbitBrokerAdmin;
public class BrokerPanic extends TestWatcher {
private RabbitBrokerAdmin brokerAdmin;
/**
* @param brokerAdmin the brokerAdmin to set
*/
public void setBrokerAdmin(RabbitBrokerAdmin brokerAdmin) {
this.brokerAdmin = brokerAdmin;
}
@Override
public Statement apply(final Statement base, Description description) {
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;
}
}
};
}
}

67
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2015 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
@@ -12,24 +12,19 @@
*/
package org.springframework.amqp.rabbit.test;
import org.springframework.amqp.rabbit.admin.RabbitBrokerAdmin;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
/**
* Global convenience class for all integration tests, carrying constants and other utilities for broker set up.
*
* @author Dave Syer
* @author Gary Russell
*
*/
public class BrokerTestUtils {
public static final int DEFAULT_PORT = 5672;
public static final int TRACER_PORT = 5673;
public static final String ADMIN_NODE_NAME = "spring@localhost";
/**
* The port that the broker is listening on (e.g. as input for a {@link ConnectionFactory}).
*
@@ -39,62 +34,4 @@ public class BrokerTestUtils {
return DEFAULT_PORT;
}
/**
* The port that the tracer is listening on (e.g. as input for a {@link ConnectionFactory}).
*
* @return a port number
*/
public static int getTracerPort() {
return TRACER_PORT;
}
/**
* An alternative port number than can safely be used to stop and start a broker, even when one is already running
* on the standard port as a privileged user. Useful for tests involving {@link RabbitBrokerAdmin} on UN*X.
*
* @return a port number
*/
public static int getAdminPort() {
return 15672;
}
/**
* Convenience factory for a {@link RabbitBrokerAdmin} instance that will usually start and stop cleanly on all
* systems.
*
* @return a {@link RabbitBrokerAdmin} instance
*/
public static RabbitBrokerAdmin getRabbitBrokerAdmin() {
return getRabbitBrokerAdmin(ADMIN_NODE_NAME, getAdminPort());
}
/**
* Convenience factory for a {@link RabbitBrokerAdmin} instance that will usually start and stop cleanly on all
* systems.
*
* @param nodeName the name of the node
*
* @return a {@link RabbitBrokerAdmin} instance
*/
public static RabbitBrokerAdmin getRabbitBrokerAdmin(String nodeName) {
return getRabbitBrokerAdmin(nodeName, getAdminPort());
}
/**
* Convenience factory for a {@link RabbitBrokerAdmin} instance that will usually start and stop cleanly on all
* systems.
*
* @param nodeName the name of the node
* @param port the port to listen on
*
* @return a {@link RabbitBrokerAdmin} instance
*/
public static RabbitBrokerAdmin getRabbitBrokerAdmin(String nodeName, int port) {
RabbitBrokerAdmin brokerAdmin = new RabbitBrokerAdmin(nodeName, port);
brokerAdmin.setRabbitLogBaseDirectory("target/rabbitmq/log");
brokerAdmin.setRabbitMnesiaBaseDirectory("target/rabbitmq/mnesia");
brokerAdmin.setStartupTimeout(10000L);
return brokerAdmin;
}
}

8
src/reference/asciidoc/whats-new.adoc
View File

@@ -3,6 +3,12 @@
==== Changes in 1.5 Since 1.4
===== spring-erlang is No Longer Supported
The `spring-erlang` jar is no longer included in the distribution.
Use <<management-template>> instead.
===== Empty Addresses Property in CachingConnectionFactory
Previously, if the connection factory was configured with a host/port, but also an empty String was supplied for `addresses`, the host and port were ignored.
@@ -68,6 +74,8 @@ Previously, bean names were composed from the ids of the `<listener-container/>`
The `@RabbitListener` annotation can now be applied at the class level.
Together with the new `@RabbitHandler` method annotation, this allows the handler method to be selected based on payload type. See <<annotation-method-selection>> for more information.==== Changes in 1.4 Since 1.3
==== Changes in 1.4 Since 1.3
===== @RabbitListener Annotation
POJO listeners can be annotated with `@RabbitListener`, enabled by `@EnableRabbit` or `<rabbit:annotation-driven />`.