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

XD-3414: Create a new project for @RedisRule

Browse Source
This commit is contained in:
Eric Bottard
2015-08-24 11:56:48 +02:00
committed by Patrick Peralta
parent 1e4c8bd749
commit 7023e16196
23 changed files with 263 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
pom.xml
View File

@@ -30,9 +30,10 @@
<module>spring-cloud-stream-rxjava</module>
<module>spring-cloud-stream-samples</module>
<module>spring-cloud-stream-module-launcher</module>
<module>docs</module>
<module>spring-cloud-stream-test-support</module>
<module>spring-cloud-stream-test-support-internal</module>
<module>spring-cloud-stream-configuration-metadata</module>
<module>docs</module>
</modules>
<dependencyManagement>
<dependencies>
@@ -118,6 +119,16 @@
<artifactId>spring-cloud-stream-test-support</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-module-test-support</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>

6
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml
View File

@@ -14,7 +14,6 @@
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<kafka.version>0.8.2.1</kafka.version>
<spring-integration-kafka.version>1.2.0.RELEASE</spring-integration-kafka.version>
<rxjava-math.version>1.0.0</rxjava-math.version>
@@ -44,6 +43,11 @@
<artifactId>spring-cloud-stream-binder-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-kafka</artifactId>

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

@@ -36,6 +36,7 @@ import org.junit.Ignore;
import org.junit.Test;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.kafka.core.KafkaMessage;

2
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java
View File

@@ -22,6 +22,8 @@ import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Registration;
import org.springframework.cloud.stream.binder.AbstractTestBinder;
import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
import org.springframework.cloud.stream.test.junit.kafka.TestKafkaCluster;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.codec.kryo.KryoRegistrar;

5
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/pom.xml
View File

@@ -46,6 +46,11 @@
<artifactId>spring-cloud-stream-binder-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>

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

@@ -29,6 +29,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
/**

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

@@ -1,57 +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 org.springframework.http.HttpStatus;
import org.springframework.web.client.HttpClientErrorException;
import org.springframework.web.client.RestTemplate;
import org.springframework.cloud.stream.binder.AbstractExternalResourceTestSupport;
import java.util.Map;
/**
* JUnit {@link org.junit.Rule} that detects the fact that RabbitMQ is available on localhost with
* the management plugin enabled.
*
* @author Gary Russell
* @since 1.2
*/
public class RabbitAdminTestSupport extends AbstractExternalResourceTestSupport<RestTemplate> {
public RabbitAdminTestSupport() {
super("RABBITADMIN");
}
@Override
protected void obtainResource() throws Exception {
resource = new RestTemplate();
try {
resource.getForObject("http://localhost:15672/api/overview", Map.class);
}
catch (HttpClientErrorException e) {
if (e.getStatusCode() != HttpStatus.UNAUTHORIZED) {
throw e;
}
}
}
@Override
protected void cleanupResource() throws Exception {
}
}

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

@@ -35,6 +35,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
import org.springframework.cloud.stream.binder.BinderUtils;
@@ -57,10 +58,7 @@ public class RabbitBinderCleanerTests {
private static final String BINDER_PREFIX = "binder.rabbit.";
@Rule
public RabbitAdminTestSupport adminTest = new RabbitAdminTestSupport();
@Rule
public RabbitTestSupport test = new RabbitTestSupport();
public RabbitTestSupport rabbitWithMgmtEnabled = new RabbitTestSupport(true);
@Test
public void testCleanStream() {
@@ -93,7 +91,7 @@ public class RabbitBinderCleanerTests {
.buildAndExpand("/", MessageChannelBinderSupport.constructDLQName(queue1Name)).encode().toUri();
template.put(uri, new AmqpQueue(false, true));
}
CachingConnectionFactory connectionFactory = test.getResource();
CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource();
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
final FanoutExchange fanout1 = new FanoutExchange(
MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, MessageChannelBinderSupport.applyPubSub(

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

@@ -46,6 +46,7 @@ import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.binder.Spy;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;

5
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/pom.xml
View File

@@ -46,6 +46,11 @@
<artifactId>spring-cloud-stream-binder-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-redis</artifactId>

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

@@ -38,6 +38,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

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

@@ -27,6 +27,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.springframework.cloud.stream.binder.BinderTestUtils;
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

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

@@ -31,6 +31,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.springframework.cloud.stream.binder.BinderTestUtils;
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericToStringSerializer;

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

@@ -27,6 +27,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.springframework.cloud.stream.binder.BinderTestUtils;
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;

134
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractExternalResourceTestSupport.java
View File

@@ -1,134 +0,0 @@
/*
* Copyright 2013 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;
import static org.junit.Assert.fail;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.junit.Assume;
import org.junit.Rule;
import org.junit.rules.TestRule;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import org.springframework.util.Assert;
/**
* Abstract base class for JUnit {@link Rule}s that detect the presence of some external resource. If the resource is
* indeed present, it will be available during the test lifecycle through {@link #getResource()}. If it is not, tests
* will either fail or be skipped, depending on the value of system property {@value #XD_EXTERNAL_SERVERS_REQUIRED}.
*
* @author Eric Bottard
* @author Gary Russell
*/
public abstract class AbstractExternalResourceTestSupport<R> implements TestRule {
public static final String XD_EXTERNAL_SERVERS_REQUIRED = "XD_EXTERNAL_SERVERS_REQUIRED";
protected R resource;
private String resourceDescription;
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected AbstractExternalResourceTestSupport(String resourceDescription) {
Assert.hasText(resourceDescription, "resourceDescription is required");
this.resourceDescription = resourceDescription;
}
@Override
public Statement apply(final Statement base, Description description) {
try {
obtainResource();
}
catch (Exception e) {
maybeCleanup();
return failOrSkip(e);
}
return new Statement() {
@Override
public void evaluate() throws Throwable {
try {
base.evaluate();
}
finally {
try {
cleanupResource();
}
catch (Exception ignored) {
logger.warn("Exception while trying to cleanup proper resource", ignored);
}
}
}
};
}
private Statement failOrSkip(final Exception e) {
String serversRequired = System.getenv(XD_EXTERNAL_SERVERS_REQUIRED);
if ("true".equalsIgnoreCase(serversRequired)) {
logger.error(resourceDescription + " IS REQUIRED BUT NOT AVAILABLE", e);
fail(resourceDescription + " IS NOT AVAILABLE");
// Never reached, here to satisfy method signature
return null;
}
else {
logger.error(resourceDescription + " IS NOT AVAILABLE, SKIPPING TESTS", e);
return new Statement() {
@Override
public void evaluate() throws Throwable {
Assume.assumeTrue("Skipping test due to " + resourceDescription + " not being available " + e, false);
}
};
}
}
private void maybeCleanup() {
if (resource != null) {
try {
cleanupResource();
}
catch (Exception ignored) {
logger.warn("Exception while trying to cleanup failed resource", ignored);
}
}
}
public R getResource() {
return resource;
}
/**
* Perform cleanup of the {@link #resource} field, which is guaranteed to be non null.
*
* @throws Exception any exception thrown by this method will be logged and swallowed
*/
protected abstract void cleanupResource() throws Exception;
/**
* Try to obtain and validate a resource. Implementors should either set the {@link #resource} field with a valid
* resource and return normally, or throw an exception.
*/
protected abstract void obtainResource() throws Exception;
}

74
spring-cloud-stream-test-support-internal/pom.xml Normal file
View File

@@ -0,0 +1,74 @@
<?xml version="1.0"?>
<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<description>Set of classes and utility code that may assist in testing both
spring-cloud-stream itself, and also modules.</description>
<properties>
<kafka.version>0.8.2.1</kafka.version>
<curator.version>2.6.0</curator.version>
</properties>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
</dependency>
<!-- External systems supported, each should be marked optional -->
<!-- redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-redis</artifactId>
<optional>true</optional>
</dependency>
<!-- kafka -->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>${curator.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<version>${kafka.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<classifier>test</classifier>
<version>${kafka.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<version>${curator.version}</version>
<optional>true</optional>
</dependency>
<!-- rabbit -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<optional>true</optional>
</dependency>
</dependencies>
</project>

133
spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/AbstractExternalResourceTestSupport.java Normal file
View File

@@ -0,0 +1,133 @@
/*
* 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.test.junit;
import static org.junit.Assert.fail;
import org.junit.Assume;
import org.junit.Rule;
import org.junit.rules.TestRule;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
/**
* Abstract base class for JUnit {@link Rule}s that detect the presence of some external resource. If the resource is
* indeed present, it will be available during the test lifecycle through {@link #getResource()}. If it is not, tests
* will either fail or be skipped, depending on the value of system property {@value #SCS_EXTERNAL_SERVERS_REQUIRED}.
*
* @author Eric Bottard
* @author Gary Russell
*/
public abstract class AbstractExternalResourceTestSupport<R> implements TestRule {
public static final String SCS_EXTERNAL_SERVERS_REQUIRED = "SCS_EXTERNAL_SERVERS_REQUIRED";
protected R resource;
private String resourceDescription;
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected AbstractExternalResourceTestSupport(String resourceDescription) {
Assert.hasText(resourceDescription, "resourceDescription is required");
this.resourceDescription = resourceDescription;
}
@Override
public Statement apply(final Statement base, Description description) {
try {
obtainResource();
}
catch (Exception e) {
maybeCleanup();
return failOrSkip(e);
}
return new Statement() {
@Override
public void evaluate() throws Throwable {
try {
base.evaluate();
}
finally {
try {
cleanupResource();
}
catch (Exception ignored) {
logger.warn("Exception while trying to cleanup proper resource", ignored);
}
}
}
};
}
private Statement failOrSkip(final Exception e) {
String serversRequired = System.getenv(SCS_EXTERNAL_SERVERS_REQUIRED);
if ("true".equalsIgnoreCase(serversRequired)) {
logger.error(resourceDescription + " IS REQUIRED BUT NOT AVAILABLE", e);
fail(resourceDescription + " IS NOT AVAILABLE");
// Never reached, here to satisfy method signature
return null;
}
else {
logger.error(resourceDescription + " IS NOT AVAILABLE, SKIPPING TESTS", e);
return new Statement() {
@Override
public void evaluate() throws Throwable {
Assume.assumeTrue("Skipping test due to " + resourceDescription + " not being available " + e, false);
}
};
}
}
private void maybeCleanup() {
if (resource != null) {
try {
cleanupResource();
}
catch (Exception ignored) {
logger.warn("Exception while trying to cleanup failed resource", ignored);
}
}
}
public R getResource() {
return resource;
}
/**
* Perform cleanup of the {@link #resource} field, which is guaranteed to be non null.
*
* @throws Exception any exception thrown by this method will be logged and swallowed
*/
protected abstract void cleanupResource() throws Exception;
/**
* Try to obtain and validate a resource. Implementors should either set the {@link #resource} field with a valid
* resource and return normally, or throw an exception.
*/
protected abstract void obtainResource() throws Exception;
}

2
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/EmbeddedZookeeper.java → spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/EmbeddedZookeeper.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kafka;
package org.springframework.cloud.stream.test.junit.kafka;
import java.io.File;
import java.net.InetSocketAddress;

5
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestSupport.java → spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kafka;
package org.springframework.cloud.stream.test.junit.kafka;
import java.util.Properties;
@@ -34,7 +34,8 @@ import org.junit.Rule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.binder.AbstractExternalResourceTestSupport;
import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport;
/**
* JUnit {@link Rule} that starts an embedded Kafka server (with an associated Zookeeper)

80
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/TestKafkaCluster.java → spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/TestKafkaCluster.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kafka;
package org.springframework.cloud.stream.test.junit.kafka;
import kafka.admin.AdminUtils;
import kafka.consumer.Consumer;
@@ -85,87 +85,9 @@ public class TestKafkaCluster {
}
/**
* See XD-2293. This is used to reproduce Kafka rebalance issues.
*/
public static void main(String[] args) throws Exception {
TestKafkaCluster cluster = new TestKafkaCluster();
ZkClient client = new ZkClient(cluster.getZkConnectString(), 10000, 10000, KafkaMessageChannelBinder.utf8Serializer);
int partitions = 5;
int replication = 1;
AdminUtils.createTopic(client, "mytopic", partitions, replication, new Properties());
Properties props = new Properties();
props.put("zookeeper.connect", cluster.getZkConnectString());
props.put("group.id", "foo");
props.put("rebalance.backoff.ms", "2000");
props.put("rebalance.max.retries", "2000");
ConsumerConfig config = new ConsumerConfig(props);
CuratorFramework curator = CuratorFrameworkFactory.newClient(cluster.getZkConnectString(), new RetryUntilElapsed(1000, 100));
curator.start();
RebalanceListener listener = null;
for (int i = 0; i < 5; i++) {
System.out.format("%nCreating consumer #%d%n", i + 1);
ConsumerConnector connector = Consumer.createJavaConsumerConnector(config);
connector.createMessageStreams(Collections.singletonMap("mytopic", 1));
if (i == 0) {
PathChildrenCache cache = new PathChildrenCache(curator, "/consumers/foo/owners/mytopic", true);
listener = new RebalanceListener(5);
cache.getListenable().addListener(listener);
cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
}
synchronized (listener) {
System.out.println("******** Waiting for rebalance...");
listener.wait();
}
}
System.out.println();
}
public String getZkConnectString() {
return zkServer.getConnectString();
}
private static class RebalanceListener implements PathChildrenCacheListener {
private int expected;
private int actual;
private boolean ready;
public RebalanceListener(int expected) {
this.expected = expected;
}
@Override
public synchronized void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
System.out.println(event);
System.out.println(event.getData() != null ? new String(event.getData().getData()) : "no data");
switch (event.getType()) {
case CHILD_ADDED:
actual++;
if (ready && actual == expected) {
System.out.println("*** Moving on... ");
this.notify();
}
break;
case CHILD_REMOVED:
actual--;
break;
case INITIALIZED:
Assert.isTrue(actual == expected);
ready = true;
this.notify();
break;
}
}
}
}

12
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestSupport.java → spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/rabbit/RabbitTestSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2015 the original author or authors.
* 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.
@@ -14,14 +14,14 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.rabbit;
package org.springframework.cloud.stream.test.junit.rabbit;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.cloud.stream.binder.AbstractExternalResourceTestSupport;
import java.net.Socket;
import javax.net.SocketFactory;
import java.net.Socket;
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.

10
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestSupport.java → spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/redis/RedisTestSupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* 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.
@@ -14,15 +14,13 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.redis;
import org.junit.Rule;
package org.springframework.cloud.stream.test.junit.redis;
import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.cloud.stream.binder.AbstractExternalResourceTestSupport;
/**
* JUnit {@link Rule} that detects the fact that a Redis server is running on localhost.
* JUnit {@link org.junit.Rule} that detects the fact that a Redis server is running on localhost.
*
* @author Gary Russell
* @author Eric Bottard

8
spring-cloud-stream-test-support/pom.xml
View File

@@ -6,12 +6,8 @@
<artifactId>spring-cloud-stream-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<name>Spring Cloud Stream Test Support</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<artifactId>spring-cloud-stream-module-test-support</artifactId>
<description>A set of classes to ease testing of Spring Cloud Stream modules.</description>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>