From cb9f64a5e33fe50faac4f8fe97e06a894dc7b75e Mon Sep 17 00:00:00 2001 From: markpollack Date: Mon, 27 Sep 2010 15:07:02 -0400 Subject: [PATCH] AMQP-58 - Support for caching of OtpConnections --- .../erlang/connection/ConnectionFactory.java | 2 +- .../connection/ConnectionFactoryUtils.java | 4 +- .../connection/SimpleConnectionFactory.java | 13 ++++-- .../erlang/core/ConnectionCallback.java | 4 +- .../erlang/core/ErlangTemplate.java | 5 ++- .../erlang/support/ErlangAccessor.java | 3 +- .../amqp/rabbit/admin/RabbitBrokerAdmin.java | 12 ++++- .../admin/JInterfaceIntegrationTests.java | 45 ++++++++++++++++--- .../RabbitBrokerAdminIntegrationTests.java | 16 +++++-- 9 files changed, 80 insertions(+), 24 deletions(-) diff --git a/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactory.java b/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactory.java index c4e98d96..168d5df7 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactory.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactory.java @@ -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; } diff --git a/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactoryUtils.java b/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactoryUtils.java index 44e3c314..2ff19956 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactoryUtils.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactoryUtils.java @@ -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; } diff --git a/spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java b/spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java index 96cddcdd..62f59c03 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/connection/SimpleConnectionFactory.java @@ -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 diff --git a/spring-erlang/src/main/java/org/springframework/erlang/core/ConnectionCallback.java b/spring-erlang/src/main/java/org/springframework/erlang/core/ConnectionCallback.java index 1e433bef..864b6159 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/core/ConnectionCallback.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/core/ConnectionCallback.java @@ -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 { * 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 } diff --git a/spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java b/spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java index 41d86965..44525667 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/core/ErlangTemplate.java @@ -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() { - 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 execute(ConnectionCallback 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); diff --git a/spring-erlang/src/main/java/org/springframework/erlang/support/ErlangAccessor.java b/spring-erlang/src/main/java/org/springframework/erlang/support/ErlangAccessor.java index 1edd2ea9..fbbaa44e 100644 --- a/spring-erlang/src/main/java/org/springframework/erlang/support/ErlangAccessor.java +++ b/spring-erlang/src/main/java/org/springframework/erlang/support/ErlangAccessor.java @@ -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(); } diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java index c5aa2c25..2274a54c 100644 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java +++ b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdmin.java @@ -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) { diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java index 1e6826b4..381407e4 100644 --- a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java +++ b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/JInterfaceIntegrationTests.java @@ -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; diff --git a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java index 105ebea5..8d75ff64 100644 --- a/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java +++ b/spring-rabbit-admin/src/test/java/org/springframework/amqp/rabbit/admin/RabbitBrokerAdminIntegrationTests.java @@ -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() {