diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactory.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactory.java deleted file mode 100644 index 7815bba3c..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactory.java +++ /dev/null @@ -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. - *

{@link #getTargetConnectionFactory(Object)} is invoked by the - * {@code SimpleMessageListenerContainer}, when establishing a connection, with the lookup key having - * the format {@code '[queueName]'}. - *

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 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 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; - } - -} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 8b7bd8fcd..adb0119b4 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -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; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java index 2548e07b4..29f00cf48 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java @@ -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; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryTests.java deleted file mode 100644 index 929e9ce52..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryTests.java +++ /dev/null @@ -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 cfs = new HashMap<>(); - - private final Map connections = new HashMap<>(); - - private final Map channels = new HashMap<>(); - - private final Map consumers = new HashMap<>(); - - private final Map 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 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 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() { - - @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; - } - -}