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

AMQP-58 - Support for caching of OtpConnections

Browse Source
This commit is contained in:
markpollack
2010-09-27 15:07:02 -04:00
parent 2974d9fb62
commit cb9f64a5e3
9 changed files with 80 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,6 +31,6 @@ import com.ericsson.otp.erlang.OtpConnection;
*/
public interface ConnectionFactory {
OtpConnection createConnection() throws UnknownHostException, OtpAuthException, IOException;
Connection createConnection() throws UnknownHostException, OtpAuthException, IOException;
}

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

@@ -20,8 +20,6 @@ package org.springframework.erlang.connection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.ericsson.otp.erlang.OtpConnection;
/**
* @author Mark Pollack
*/
@@ -33,7 +31,7 @@ public class ConnectionFactoryUtils {
/**
* Release the given Connection by closing it.
*/
public static void releaseConnection(OtpConnection con, ConnectionFactory cf) {
public static void releaseConnection(Connection con, ConnectionFactory cf) {
if (con == null) {
return;
}

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

@@ -24,11 +24,18 @@ import org.springframework.erlang.OtpIOException;
import org.springframework.util.Assert;
import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpConnection;
import com.ericsson.otp.erlang.OtpPeer;
import com.ericsson.otp.erlang.OtpSelf;
/**
* A simple implementation of {@link ConnectionFactory} that return a new connection
* for each invocation of the createConnection method.
*
* Note that use of this ConnectionFactory with ErlangTemplate has unstable behavior when
* invoked frequently and will be deprecated. See {@link SingleConnectionFactory} for an
* alternative implementation.
*
*
* Provides a more traditional API to creating a connection to a remote erlang node than
* the JInterface API.
*
@@ -91,9 +98,9 @@ public class SimpleConnectionFactory implements ConnectionFactory, InitializingB
}
public OtpConnection createConnection() throws UnknownHostException, OtpAuthException, IOException {
public Connection createConnection() throws UnknownHostException, OtpAuthException, IOException {
try {
return otpSelf.connect(otpPeer);
return new DefaultConnection(otpSelf.connect(otpPeer));
}
catch (IOException ex) {
throw new OtpIOException("failed to connect from '" + this.selfNodeName

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

@@ -16,7 +16,7 @@
package org.springframework.erlang.core;
import com.ericsson.otp.erlang.OtpConnection;
import org.springframework.erlang.connection.Connection;
/**
* Basic callback for use in ErlangTemplate
@@ -28,6 +28,6 @@ public interface ConnectionCallback<T> {
* Execute any number of operations against the supplied OTP connection,
* possibly returning a result.
*/
T doInConnection(OtpConnection connection) throws Exception; //Not sure everything it throws
T doInConnection(Connection connection) throws Exception; //Not sure everything it throws
}

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

@@ -19,6 +19,7 @@ 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;
@@ -44,7 +45,7 @@ public class ErlangTemplate extends ErlangAccessor implements ErlangOperations {
public OtpErlangObject executeErlangRpc(final String module, final String function, final OtpErlangList args) {
return execute(new ConnectionCallback<OtpErlangObject>() {
public OtpErlangObject doInConnection(OtpConnection connection) throws Exception {
public OtpErlangObject doInConnection(Connection connection) throws Exception {
logger.debug("Sending RPC for module [" + module + "] function [" + function + "] args [" + args);
connection.sendRPC(module, function, args);
//TODO consider dedicated response object.
@@ -109,7 +110,7 @@ public class ErlangTemplate extends ErlangAccessor implements ErlangOperations {
public <T> T execute(ConnectionCallback<T> action) throws OtpException {
Assert.notNull(action, "Callback object must not be null");
OtpConnection con = null;
Connection con = null;
try {
con = createConnection();
return action.doInConnection(con);

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

@@ -24,6 +24,7 @@ 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;
@@ -39,7 +40,7 @@ public abstract class ErlangAccessor implements InitializingBean {
private ConnectionFactory connectionFactory;
protected OtpConnection createConnection() throws UnknownHostException, OtpAuthException, IOException {
protected Connection createConnection() throws UnknownHostException, OtpAuthException, IOException {
return getConnectionFactory().createConnection();
}

12
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java
View File

@@ -32,6 +32,7 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.erlang.connection.SimpleConnectionFactory;
import org.springframework.erlang.connection.SingleConnectionFactory;
import org.springframework.erlang.core.ErlangTemplate;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedOperationParameter;
@@ -303,9 +304,16 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations {
protected void initializeDefaultErlangTemplate(RabbitTemplate rabbitTemplate) {
String peerNodeName = "rabbit@" + rabbitTemplate.getConnectionFactory().getHost();
logger.debug("Creating jinterface connection with peerNodeName = [" + peerNodeName + "]");
SimpleConnectionFactory otpCf = new SimpleConnectionFactory("rabbit-spring-monitor", peerNodeName);
createErlangTemplate(createErlangConnectionFactory(peerNodeName));
}
protected org.springframework.erlang.connection.ConnectionFactory createErlangConnectionFactory(
String peerNodeName) {
logger.debug("Creating org.springframework.erlang.connection.SingleConnectionFactory.");
SingleConnectionFactory otpCf = new SingleConnectionFactory("rabbit-spring-monitor", peerNodeName);
otpCf.afterPropertiesSet();
createErlangTemplate(otpCf);
return (org.springframework.erlang.connection.ConnectionFactory) otpCf;
}
protected void createErlangTemplate(org.springframework.erlang.connection.ConnectionFactory otpCf) {

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

@@ -23,7 +23,7 @@ import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpPeer;
import com.ericsson.otp.erlang.OtpSelf;
@Ignore("manual integration test only.")
//@Ignore("manual integration test only.")
public class JInterfaceIntegrationTests {
@Test
@@ -32,8 +32,8 @@ public class JInterfaceIntegrationTests {
try {
OtpSelf self = new OtpSelf("rabbit-monitor");
String hostName = "rabbit@"
+ InetAddress.getLocalHost().getHostName();
String hostName = "rabbit@vmc-ssrc-rh82";
//+ InetAddress.getLocalHost().getHostName();
OtpPeer peer = new OtpPeer(hostName);
connection = self.connect(peer);
// connection.sendRPC("erlang","date", new OtpErlangList());
@@ -94,13 +94,44 @@ public class JInterfaceIntegrationTests {
}
@Test
public void rawOtpConnect() throws Exception {
String cookie = readCookie();
OtpSelf self = new OtpSelf("rabbit-monitor", cookie);
OtpPeer peer = new OtpPeer("rabbit@" + InetAddress.getLocalHost().getHostName());
self.connect(peer);
createConnection();
}
@Test
public void stressTest() throws Exception {
//String cookie = readCookie();
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");
System.out.println("i = " + i);
}
}
public OtpConnection createConnection() throws Exception {
OtpSelf self = new OtpSelf("rabbit-monitor");
OtpPeer peer = new OtpPeer("rabbit");// + InetAddress.getLocalHost().getHostName());
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();
//System.out.println(module + " response received = " + response.toString());
if (recycleConnection) {
con.close();
con = createConnection();
}
}
private String readCookie() throws Exception {
String cookie = null;

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

@@ -80,10 +80,18 @@ public class RabbitBrokerAdminIntegrationTests {
}
@Test
//
public void testStatusAndBrokerLifecycle() {
public void repeatLifecycle() throws Exception {
for (int i = 1; i< 100; i++) {
testStatusAndBrokerLifecycle();
System.out.println("i = " + i);
//Thread.sleep(1000);
}
}
//@Test
public void testStatusAndBrokerLifecycle() throws Exception {
RabbitStatus status = brokerAdmin.getStatus();
assertBrokerAppRunning(status);
brokerAdmin.stopBrokerApplication();
status = brokerAdmin.getStatus();
@@ -94,6 +102,8 @@ public class RabbitBrokerAdminIntegrationTests {
assertBrokerAppRunning(status);
}
@Test
//@Ignore("NEEDS RABBITMQ_HOME to be set.")
public void testStartNode() {