SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use LocalizedQueueConnectionFactory from Spring AMQP

Browse Source 
- Remove LocalizedQueueConnectionFactory and the unit tests
   - The unit tests are already covered in Spring AMQP and also, certain methods accessors require public access to run tests from SCS
 - Kept the integration tests as is (though this can be removed as well - it is also covered in Spring AMQP)

This resolves #279
This commit is contained in:
Ilayaperumal Gopinathan
2016-01-19 19:30:49 +05:30
committed by Mark Fisher
parent 31ae416727
commit e792b0e347
4 changed files with 2 additions and 417 deletions
Download Patch File Download Diff File Expand all files Collapse all files

231
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactory.java
View File

@@ -1,231 +0,0 @@
/*
* Copyright 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.cloud.stream.binder.rabbit;
import java.net.URI;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
import org.springframework.amqp.rabbit.connection.RoutingConnectionFactory;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
/**
* A {@link RoutingConnectionFactory} that determines the node on which a queue is located and
* returns a factory that connects directly to that node.
* The RabbitMQ management plugin is called over REST to determine the node and the corresponding
* address for that node is injected into the connection factory.
* A single instance of each connection factory is retained in a cache.
* If the location cannot be determined, the default connection factory is returned. This connection
* factory is typically configured to connect to all the servers in a fail-over mode.
* <p>{@link #getTargetConnectionFactory(Object)} is invoked by the
* {@code SimpleMessageListenerContainer}, when establishing a connection, with the lookup key having
* the format {@code '[queueName]'}.
* <p>All {@link ConnectionFactory} methods delegate to the default
*
* @author Gary Russell
* @since 1.2
*/
public class LocalizedQueueConnectionFactory implements ConnectionFactory, RoutingConnectionFactory {
private final Log logger = LogFactory.getLog(getClass());
private final Map<String, ConnectionFactory> nodeFactories = new HashMap<>();
private final ConnectionFactory defaultConnectionFactory;
private final String[] addresses;
private final String[] adminAdresses;
private final String[] nodes;
private final String vhost;
private final String username;
private final String password;
private final boolean useSSL;
private final Resource sslPropertiesLocation;
/**
*
* @param defaultConnectionFactory the fallback connection factory to use if the queue can't be located.
* @param addresses the rabbitmq server addresses (host:port, ...).
* @param adminAddresses the rabbitmq admin addresses (http://host:port, ...) must be the same length
* as addresses.
* @param nodes the rabbitmq nodes corresponding to addresses (rabbit@server1, ...).
* @param vhost the virtual host.
* @param username the user name.
* @param password the password.
*/
public LocalizedQueueConnectionFactory(ConnectionFactory defaultConnectionFactory,
String[] addresses, String[] adminAddresses, String[] nodes, String vhost,
String username, String password, boolean useSSL, Resource sslPropertiesLocation) {
Assert.isTrue(addresses.length == adminAddresses.length
&& addresses.length == nodes.length,
"'addresses', 'adminAddresses', and 'nodes' properties must have equal length");
this.defaultConnectionFactory = defaultConnectionFactory;
this.addresses = Arrays.copyOf(addresses, addresses.length);
this.adminAdresses = Arrays.copyOf(adminAddresses, adminAddresses.length);
this.nodes = Arrays.copyOf(nodes, nodes.length);
this.vhost = vhost;
this.username = username;
this.password = password;
this.useSSL = useSSL;
this.sslPropertiesLocation = sslPropertiesLocation;
}
@Override
public Connection createConnection() throws AmqpException {
return this.defaultConnectionFactory.createConnection();
}
@Override
public String getHost() {
return this.defaultConnectionFactory.getHost();
}
@Override
public int getPort() {
return this.defaultConnectionFactory.getPort();
}
@Override
public String getVirtualHost() {
return this.vhost;
}
@Override
public void addConnectionListener(ConnectionListener listener) {
this.defaultConnectionFactory.addConnectionListener(listener);
}
@Override
public boolean removeConnectionListener(ConnectionListener listener) {
return this.defaultConnectionFactory.removeConnectionListener(listener);
}
@Override
public void clearConnectionListeners() {
this.defaultConnectionFactory.clearConnectionListeners();
}
@Override
public ConnectionFactory getTargetConnectionFactory(Object key) {
String queue = ((String) key);
queue = queue.substring(1, queue.length() - 1);
ConnectionFactory connectionFactory = determineConnectionFactory(queue);
if (connectionFactory == null) {
return this.defaultConnectionFactory;
}
else {
return connectionFactory;
}
}
private ConnectionFactory determineConnectionFactory(String queue) {
for (int i = 0; i < this.adminAdresses.length; i++) {
String adminUri = this.adminAdresses[i];
RestTemplate template = createRestTemplate(adminUri);
URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("queues", "{vhost}", "{queue}")
.buildAndExpand(this.vhost, queue).encode().toUri();
try {
@SuppressWarnings("unchecked")
Map<String, Object> queueInfo = template.getForObject(uri, Map.class);
if (queueInfo != null) {
String node = (String) queueInfo.get("node");
if (node != null) {
for (int j = 0; j < this.nodes.length; j++) {
if (this.nodes[j].equals(node)) {
return nodeConnectionFactory(queue, j);
}
}
}
}
}
catch (Exception e) {
logger.error("Failed to determine queue location for: " + queue + " at: " +
uri.toString(), e);
}
}
logger.warn("Failed to determine queue location for: " + queue);
return null;
}
private synchronized ConnectionFactory nodeConnectionFactory(String queue, int index) throws Exception {
String address = this.addresses[index];
String node = this.nodes[index];
if (logger.isDebugEnabled()) {
logger.debug("Queue: " + queue + " is on node: " + node + " at: " + address);
}
ConnectionFactory cf = this.nodeFactories.get(node);
if (cf == null) {
if (logger.isDebugEnabled()) {
logger.debug("Creating new connection factory for: " + address);
}
cf = createConnectionFactory(address);
this.nodeFactories.put(node, cf);
}
return cf;
}
/**
* Create a RestTemplate for the supplied URI.
* @param adminUri the URI.
* @return the template.
*/
protected RestTemplate createRestTemplate(String adminUri) {
return RabbitManagementUtils.buildRestTemplate(adminUri, this.username, this.password);
}
/**
* Create a dedicated connection factory for the address.
* @param address the address to which the factory should connect.
* @return the connection factory.
* @throws Exception if errors occur during creation.
*/
protected ConnectionFactory createConnectionFactory(String address) throws Exception {
RabbitConnectionFactoryBean rcfb = new RabbitConnectionFactoryBean();
rcfb.setUseSSL(this.useSSL);
rcfb.setSslPropertiesLocation(this.sslPropertiesLocation);
rcfb.afterPropertiesSet();
CachingConnectionFactory ccf = new CachingConnectionFactory(rcfb.getObject());
ccf.setAddresses(address);
ccf.setUsername(this.username);
ccf.setPassword(this.password);
ccf.setVirtualHost(this.vhost);
return ccf;
}
}

1
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -42,6 +42,7 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

1
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java
View File

@@ -27,6 +27,7 @@ import org.junit.Test;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;

186
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryTests.java
View File

@@ -1,186 +0,0 @@
/*
* Copyright 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.cloud.stream.binder.rabbit;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Matchers.anyMap;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Matchers;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.http.MediaType;
import org.springframework.test.web.client.MockRestServiceServer;
import org.springframework.web.client.RestTemplate;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
/**
*
* @author Gary Russell
*/
public class LocalizedQueueConnectionFactoryTests {
private final Map<String, ConnectionFactory> cfs = new HashMap<>();
private final Map<String, Connection> connections = new HashMap<>();
private final Map<String, Channel> channels = new HashMap<>();
private final Map<String, Consumer> consumers = new HashMap<>();
private final Map<String, String> consumerTags = new HashMap<>();
private final CountDownLatch latch = new CountDownLatch(2);
@SuppressWarnings("unchecked")
@Test
public void testFailOver() throws Exception {
ConnectionFactory defaultConnectionFactory = mockCF("localhost:1234");
String rabbit1 = "localhost:1235";
String rabbit2 = "localhost:1236";
String[] addresses = new String[] { rabbit1, rabbit2 };
String[] adminAddresses = new String[] { "http://localhost:11235", "http://localhost:11236" };
String[] nodes = new String[] { "rabbit@foo", "rabbit@bar" };
String vhost = "/";
String username = "guest";
String password = "guest";
final AtomicBoolean firstServer = new AtomicBoolean(true);
LocalizedQueueConnectionFactory lqcf = new LocalizedQueueConnectionFactory(defaultConnectionFactory, addresses,
adminAddresses, nodes, vhost, username, password, false, null) {
private final String[] nodes = new String[] { "rabbit@foo", "rabbit@bar" };
@Override
protected RestTemplate createRestTemplate(String adminUri) {
return doCreateRestTemplate(adminUri, firstServer.get() ? nodes[0] : nodes[1]);
}
@Override
protected ConnectionFactory createConnectionFactory(String address) throws Exception {
return mockCF(address);
}
};
Log logger = spy(TestUtils.getPropertyValue(lqcf, "logger", Log.class));
new DirectFieldAccessor(lqcf).setPropertyValue("logger", logger);
when(logger.isDebugEnabled()).thenReturn(true);
ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(lqcf);
container.setQueueNames("q");
container.afterPropertiesSet();
container.start();
Channel channel = this.channels.get(rabbit1);
verify(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(),
anyBoolean(), anyMap(),
Matchers.any(Consumer.class));
verify(logger, atLeast(1)).debug(captor.capture());
assertTrue(assertLog(captor.getAllValues(), "Queue: q is on node: rabbit@foo at: localhost:1235"));
// Fail rabbit1 and verify the container switches to rabbit2
firstServer.set(false);
when(channel.isOpen()).thenReturn(false);
when(this.connections.get(rabbit1).isOpen()).thenReturn(false);
this.consumers.get(rabbit1).handleCancel(consumerTags.get(rabbit1));
assertTrue(latch.await(10, TimeUnit.SECONDS));
channel = this.channels.get(rabbit2);
verify(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(),
anyBoolean(), anyMap(),
Matchers.any(Consumer.class));
container.stop();
verify(logger, atLeast(1)).debug(captor.capture());
assertTrue(assertLog(captor.getAllValues(), "Queue: q is on node: rabbit@bar at: localhost:1236"));
}
private boolean assertLog(List<String> logRows, String expected) {
for (String log : logRows) {
if (log.contains(expected)) {
return true;
}
}
return false;
}
private RestTemplate doCreateRestTemplate(String uri, String node) {
RestTemplate template = new RestTemplate();
MockRestServiceServer server = MockRestServiceServer.createServer(template);
server.expect(requestTo(uri + "/api/queues/%2F/q"))
.andRespond(withSuccess("{ \"node\" : \""
+ node
+ "\" }", MediaType.APPLICATION_JSON));
return template;
}
@SuppressWarnings("unchecked")
private ConnectionFactory mockCF(final String address) throws Exception {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
Connection connection = mock(Connection.class);
Channel channel = mock(Channel.class);
when(connectionFactory.createConnection()).thenReturn(connection);
when(connection.createChannel(false)).thenReturn(channel);
when(connection.isOpen()).thenReturn(true);
when(channel.isOpen()).thenReturn(true);
doAnswer(new Answer<String>() {
@Override
public String answer(InvocationOnMock invocation) throws Throwable {
String tag = UUID.randomUUID().toString();
consumers.put(address, (Consumer) invocation.getArguments()[6]);
consumerTags.put(address, tag);
latch.countDown();
return tag;
}
}).when(channel).basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), anyMap(),
any(Consumer.class));
when(connectionFactory.getHost()).thenReturn(address);
this.cfs.put(address, connectionFactory);
this.connections.put(address, connection);
this.channels.put(address, channel);
return connectionFactory;
}
}