diff --git a/pom.xml b/pom.xml index 02c1a740d..67d87c78f 100644 --- a/pom.xml +++ b/pom.xml @@ -27,6 +27,7 @@ spring-cloud-stream-binder-rabbit spring-cloud-starter-stream-rabbit + spring-cloud-stream-binder-rabbit-test-support diff --git a/spring-cloud-stream-binder-rabbit-test-support/pom.xml b/spring-cloud-stream-binder-rabbit-test-support/pom.xml new file mode 100644 index 000000000..996f5ccec --- /dev/null +++ b/spring-cloud-stream-binder-rabbit-test-support/pom.xml @@ -0,0 +1,32 @@ + + + 4.0.0 + + org.springframework.cloud + spring-cloud-stream-binder-rabbit-parent + 1.1.0.BUILD-SNAPSHOT + + spring-cloud-stream-binder-rabbit-test-support + Rabbit related test classes + + + junit + junit + compile + + + org.springframework.cloud + spring-cloud-stream-test-support-internal + + + org.springframework.boot + spring-boot-starter-logging + + + org.springframework.boot + spring-boot-starter-amqp + true + + + \ No newline at end of file diff --git a/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java b/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java new file mode 100644 index 000000000..073f01ce0 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java @@ -0,0 +1,166 @@ +/* + * 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.test.junit.rabbit; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import javax.net.ServerSocketFactory; +import javax.net.SocketFactory; + +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport; + +/** + * JUnit {@link org.junit.Rule} that detects the fact that RabbitMQ is available on localhost. + * + * @author Mark Fisher + * @author Gary Russell + * @author Eric Bottard + */ +public class RabbitTestSupport extends AbstractExternalResourceTestSupport { + + private final boolean management; + + public RabbitTestSupport() { + this(false); + } + + public RabbitTestSupport(boolean management) { + super("RABBIT"); + this.management = management; + } + + @Override + protected void obtainResource() throws Exception { + resource = new CachingConnectionFactory("localhost"); + resource.createConnection().close(); + if (management) { + Socket socket = SocketFactory.getDefault().createSocket("localhost", 15672); + socket.close(); + } + } + + @Override + protected void cleanupResource() throws Exception { + resource.destroy(); + } + + /** + * Test class to allow testing deferred entity declarations when RabbitMQ is down. + * + */ + public static class RabbitProxy { + + private final int port; + + private final ExecutorService serverExec = Executors.newSingleThreadExecutor(); + + private final ExecutorService socketExec = Executors.newCachedThreadPool(); + + private volatile ServerSocket serverSocket; + + public RabbitProxy() throws IOException { + ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0); + this.port = serverSocket.getLocalPort(); + serverSocket.close(); + } + + public int getPort() { + return this.port; + } + + public void start() throws IOException { + this.serverSocket = ServerSocketFactory.getDefault().createServerSocket(this.port); + this.serverExec.execute(new Runnable() { + + @Override + public void run() { + try { + while (true) { + final Socket socket = serverSocket.accept(); + socketExec.execute(new Runnable() { + + @Override + public void run() { + try { + final Socket rabbitSocket = SocketFactory.getDefault().createSocket("localhost", + 5672); + socketExec.execute(new Runnable() { + + @Override + public void run() { + try { + InputStream is = rabbitSocket.getInputStream(); + OutputStream os = socket.getOutputStream(); + int c; + while ((c = is.read()) >= 0) { + os.write(c); + } + } + catch (IOException e) { + try { + socket.close(); + rabbitSocket.close(); + } + catch (IOException e1) { + } + } + } + }); + InputStream is = socket.getInputStream(); + OutputStream os = rabbitSocket.getOutputStream(); + int c; + while ((c = is.read()) >= 0) { + os.write(c); + } + } + catch (IOException e) { + try { + socket.close(); + } + catch (IOException e1) { + } + } + } + + }); + } + } + catch (IOException e) { + try { + serverSocket.close(); + } + catch (IOException e1) { + } + } + } + }); + } + + public void stop() throws IOException { + this.serverSocket.close(); + } + + } + +} diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index 9e4d6d404..d890c1f03 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -47,6 +47,12 @@ spring-cloud-stream-test-support-internal test + + org.springframework.cloud + spring-cloud-stream-binder-rabbit-test-support + ${project.version} + test + org.springframework.boot spring-boot-starter-amqp diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 0f828928d..81eac8856 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -49,7 +49,7 @@ import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy; import org.springframework.cloud.stream.binder.PartitionSelectorStrategy; import org.springframework.cloud.stream.binder.PartitionTestSupport; import org.springframework.cloud.stream.binder.Spy; -import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport; +import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; import org.springframework.context.ApplicationContext; import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.integration.channel.DirectChannel;