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

RabbitBrokerAdmin now uses SingleConnectionFactory

Browse Source 
The self node name created appends a UUID suffix to make it unique.
Trim node names passed to JInterface
SingleConnectionFactory now implements DisposableBean
This commit is contained in:
markpollack
2010-09-29 10:34:52 -04:00
parent 1efba1df6e
commit 72ced9dd74
5 changed files with 82 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,7 +18,10 @@ 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;
@@ -75,6 +78,10 @@ import com.ericsson.otp.erlang.OtpSelf;
*/
public class SimpleConnectionFactory implements ConnectionFactory, InitializingBean {
protected final Log logger = LogFactory.getLog(getClass());
private boolean uniqueSelfNodeName = true;
private String selfNodeName;
private String cookie;
@@ -106,22 +113,36 @@ public class SimpleConnectionFactory implements ConnectionFactory, InitializingB
+ "' 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' 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(this.selfNodeName);
this.otpSelf = new OtpSelf(selfNodeNameToUse.trim());
}
else {
this.otpSelf = new OtpSelf(this.selfNodeName, this.cookie);
this.otpSelf = new OtpSelf(selfNodeNameToUse.trim(), this.cookie);
}
}
catch (IOException e) {
throw new OtpIOException(e);
}
this.otpPeer = new OtpPeer(this.peerNodeName);
this.otpPeer = new OtpPeer(this.peerNodeName.trim());
}
}

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

@@ -24,9 +24,11 @@ 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;
@@ -78,9 +80,11 @@ import com.ericsson.otp.erlang.OtpSelf;
* @author Mark Pollack
*/
public class SingleConnectionFactory implements ConnectionFactory,
InitializingBean {
InitializingBean, DisposableBean {
protected final Log logger = LogFactory.getLog(getClass());
private boolean uniqueSelfNodeName = true;
private String selfNodeName;
@@ -112,6 +116,14 @@ public class SingleConnectionFactory implements ConnectionFactory,
this.selfNodeName = selfNodeName;
this.peerNodeName = peerNodeName;
}
public boolean isUniqueSelfNodeName() {
return uniqueSelfNodeName;
}
public void setUniqueSelfNodeName(boolean uniqueSelfNodeName) {
this.uniqueSelfNodeName = uniqueSelfNodeName;
}
public Connection createConnection() throws UnknownHostException,
OtpAuthException {
@@ -144,6 +156,29 @@ public class SingleConnectionFactory implements ConnectionFactory,
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.
*/
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.
@@ -207,16 +242,21 @@ public class SingleConnectionFactory implements ConnectionFactory,
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(this.selfNodeName);
this.otpSelf = new OtpSelf(selfNodeNameToUse.trim());
} else {
this.otpSelf = new OtpSelf(this.selfNodeName, this.cookie);
this.otpSelf = new OtpSelf(selfNodeNameToUse.trim(), this.cookie);
}
} catch (IOException e) {
throw new OtpIOException(e);
}
this.otpPeer = new OtpPeer(this.peerNodeName);
this.otpPeer = new OtpPeer(this.peerNodeName.trim());
}
private class SharedConnectionInvocationHandler implements

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

@@ -38,6 +38,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.erlang.OtpAuthException;
import org.springframework.erlang.OtpIOException;
import org.springframework.erlang.connection.SimpleConnectionFactory;
import org.springframework.erlang.connection.SingleConnectionFactory;
import org.springframework.erlang.core.Application;
import org.springframework.erlang.core.ErlangTemplate;
import org.springframework.erlang.core.Node;
@@ -376,7 +377,7 @@ public class RabbitBrokerAdmin implements RabbitBrokerOperations {
}
String peerNodeName = "rabbit@" + host;
logger.debug("Creating jinterface connection with peerNodeName = [" + peerNodeName + "]");
SimpleConnectionFactory otpCf = new SimpleConnectionFactory("rabbit-spring-monitor", peerNodeName);
SingleConnectionFactory otpCf = new SingleConnectionFactory("rabbit-spring-monitor", peerNodeName);
otpCf.afterPropertiesSet();
createErlangTemplate(otpCf);
}

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

@@ -9,9 +9,8 @@ import java.net.UnknownHostException;
import junit.framework.Assert;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.erlang.connection.SimpleConnectionFactory;
import org.springframework.erlang.connection.SingleConnectionFactory;
import org.springframework.erlang.core.ErlangTemplate;
import com.ericsson.otp.erlang.OtpAuthException;
@@ -26,14 +25,14 @@ import com.ericsson.otp.erlang.OtpSelf;
//@Ignore("manual integration test only.")
public class JInterfaceIntegrationTests {
private static int counter;
@Test
public void rawApi() {
OtpConnection connection = null;
try {
OtpSelf self = new OtpSelf("rabbit-monitor");
String hostName = "rabbit@vmc-ssrc-rh82";
//+ InetAddress.getLocalHost().getHostName();
String hostName = "rabbit@" + InetAddress.getLocalHost().getHostName();
OtpPeer peer = new OtpPeer(hostName);
connection = self.connect(peer);
// connection.sendRPC("erlang","date", new OtpErlangList());
@@ -77,7 +76,7 @@ public class JInterfaceIntegrationTests {
//System.out.println("home = " + home);
//System.out.println("peerNodeName = " + peerNodeName);
SimpleConnectionFactory cf = new SimpleConnectionFactory(selfNodeName,
SingleConnectionFactory cf = new SingleConnectionFactory(selfNodeName,
peerNodeName);
cf.afterPropertiesSet();
@@ -90,6 +89,8 @@ public class JInterfaceIntegrationTests {
long number = (Long) template.executeAndConvertRpc("erlang", "abs", -161803399);
Assert.assertEquals(161803399, number);
cf.destroy();
}
@@ -117,8 +118,8 @@ public class JInterfaceIntegrationTests {
}
public OtpConnection createConnection() throws Exception {
OtpSelf self = new OtpSelf("rabbit-monitor");
OtpPeer peer = new OtpPeer("rabbit");// + InetAddress.getLocalHost().getHostName());
OtpSelf self = new OtpSelf("rabbit-monitor-" + counter++);
OtpPeer peer = new OtpPeer("rabbit@" + InetAddress.getLocalHost().getHostName());
return self.connect(peer);
}

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

@@ -23,14 +23,16 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.erlang.OtpIOException;
/**
*
* This test class assumes that you are already running the rabbitmq broker.
* @author Mark Pollack
*/
public class RabbitBrokerAdminIntegrationTests {
@@ -47,15 +49,6 @@ public class RabbitBrokerAdminIntegrationTests {
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
brokerAdmin = new RabbitBrokerAdmin(connectionFactory);
logger.info("Starting broker node");
brokerAdmin.startNode();
Thread.sleep(1000L);
}
@AfterClass
public static void tearDown() {
logger.info("Shutting down broker node");
brokerAdmin.stopNode();
}
@Test
@@ -102,7 +95,6 @@ public class RabbitBrokerAdminIntegrationTests {
for (int i = 1; i< 100; i++) {
testStatusAndBrokerLifecycle();
System.out.println("i = " + i);
//Thread.sleep(1000);
}
}
@@ -123,7 +115,7 @@ public class RabbitBrokerAdminIntegrationTests {
@Test
//@Ignore("NEEDS RABBITMQ_HOME to be set.")
@Ignore("Test Manually")
public void testStartNode() {
try {
brokerAdmin.stopNode();
@@ -132,12 +124,10 @@ public class RabbitBrokerAdminIntegrationTests {
}
brokerAdmin.startNode();
assertEquals(1,1);
brokerAdmin.stopNode();
}
@Test
public void testGetQueues() throws Exception {
Thread.sleep(1000L);
brokerAdmin.declareQueue(new Queue("test.queue"));
assertEquals("/", connectionFactory.getVirtualHost());
List<QueueInfo> queues = brokerAdmin.getQueues();