GH-163: Refresh testing-demo as functions
Fixes https://github.com/spring-cloud/spring-cloud-stream-samples/issues/163 * Rework a logic for all the Spring Cloud Stream micorservices in the `testing-demo` into functions without any `@EnableBinding` and friends * Change test to JUnit 5 * Change the logic in tests to reflect function-based configuration and modern approaches for testing this kind of applications * Use `@EmbeddedKafka` instead of rule which is not available in JUnit 5 * Exclude unnecessary auto-configurations from tests * Decrease `OddEvenSourceTests` time using `spring.cloud.stream.poller.fixed-delay=1` configuration property * Removing headerMode from test configuration
This commit is contained in:
Artem Bilan
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2017 the original author or authors.
|
||||
* Copyright 2017-2019 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.
|
||||
@@ -16,12 +16,11 @@
|
||||
|
||||
package org.springframework.cloud.stream.testing.processor;
|
||||
|
||||
import java.util.function.Function;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.annotation.StreamListener;
|
||||
import org.springframework.cloud.stream.messaging.Processor;
|
||||
import org.springframework.messaging.handler.annotation.SendTo;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
/**
|
||||
* The Spring Cloud Stream Processor application,
|
||||
@@ -31,13 +30,11 @@ import org.springframework.messaging.handler.annotation.SendTo;
|
||||
*
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableBinding(Processor.class)
|
||||
public class ToUpperCaseProcessor {
|
||||
|
||||
@StreamListener(Processor.INPUT)
|
||||
@SendTo(Processor.OUTPUT)
|
||||
public String transform(String payload) {
|
||||
return payload.toUpperCase();
|
||||
@Bean
|
||||
public Function<String, String> uppercaseFunction() {
|
||||
return String::toUpperCase;
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
|
||||
@@ -1,51 +1,57 @@
|
||||
/*
|
||||
* Copyright 2017 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
|
||||
*
|
||||
* https://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.testing.sink;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.messaging.Sink;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.jdbc.JdbcMessageHandler;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
/**
|
||||
* The Spring Cloud Stream Sink application,
|
||||
* which insert payloads of incoming messages to the FOOBAR table in the RDBMS.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableBinding(Sink.class)
|
||||
public class JdbcSink {
|
||||
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = Sink.INPUT)
|
||||
public MessageHandler jdbcHandler(DataSource dataSource) {
|
||||
return new JdbcMessageHandler(dataSource, "INSERT INTO foobar (value) VALUES (:payload)");
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(JdbcSink.class, args);
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2017-2019 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
|
||||
*
|
||||
* https://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.testing.sink;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.integration.dsl.IntegrationFlow;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.integration.jdbc.JdbcMessageHandler;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
|
||||
/**
|
||||
* The Spring Cloud Stream Sink application,
|
||||
* which insert payloads of incoming messages to the SINK table in the RDBMS.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@SpringBootApplication
|
||||
public class JdbcSink {
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow jdbcConsumerFlow() {
|
||||
return IntegrationFlows.from(Consumer.class, (gateway) -> gateway.beanName("jdbcConsumer"))
|
||||
.handle(jdbcHandler(null))
|
||||
.get();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public MessageHandler jdbcHandler(DataSource dataSource) {
|
||||
return new JdbcMessageHandler(dataSource, "INSERT INTO sink (value) VALUES (:payload)");
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(JdbcSink.class, args);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,55 +1,47 @@
|
||||
/*
|
||||
* Copyright 2017 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
|
||||
*
|
||||
* https://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.testing.source;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.messaging.Source;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.integration.annotation.InboundChannelAdapter;
|
||||
import org.springframework.integration.annotation.Poller;
|
||||
import org.springframework.integration.core.MessageSource;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
/**
|
||||
* The Spring Cloud Stream Source application,
|
||||
* which generates each 100 milliseconds "foo" or "bar" string in round-robin manner.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableBinding(Source.class)
|
||||
public class FooBarSource {
|
||||
|
||||
private AtomicBoolean semaphore = new AtomicBoolean(true);
|
||||
|
||||
@Bean
|
||||
@InboundChannelAdapter(channel = Source.OUTPUT, poller = @Poller(fixedDelay = "100"))
|
||||
public MessageSource<String> fooBarStrings() {
|
||||
return () ->
|
||||
new GenericMessage<>(this.semaphore.getAndSet(!this.semaphore.get()) ? "foo" : "bar");
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(FooBarSource.class, args);
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2017-2019 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
|
||||
*
|
||||
* https://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.testing.source;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
/**
|
||||
* The Spring Cloud Stream Source application,
|
||||
* which generates every second "odd" or "even" string in round-robin manner.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@SpringBootApplication
|
||||
public class OddEvenSource {
|
||||
|
||||
private AtomicBoolean semaphore = new AtomicBoolean(true);
|
||||
|
||||
@Bean
|
||||
public Supplier<String> oddEvenSupplier() {
|
||||
return () -> this.semaphore.getAndSet(!this.semaphore.get()) ? "odd" : "even";
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(OddEvenSource.class, args);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,37 +1,37 @@
|
||||
/*
|
||||
* Copyright 2017 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
|
||||
*
|
||||
* https://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.testing.processor;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
* The simple test-case to demonstrate how it is easy to apply unit testing assertion
|
||||
* for te Spring Cloud Stream applications.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
public class NaiveToUpperCaseTests {
|
||||
|
||||
@Test
|
||||
public void testUpperCase() {
|
||||
assertEquals("FOO", new ToUpperCaseProcessor().transform("foo"));
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2017-2019 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
|
||||
*
|
||||
* https://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.testing.processor;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* The simple test-case to demonstrate how it is easy to apply unit testing assertion
|
||||
* for te Spring Cloud Stream applications.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
class NaiveToUpperCaseTests {
|
||||
|
||||
@Test
|
||||
void testUpperCase() {
|
||||
assertEquals("TEST", new ToUpperCaseProcessor().uppercaseFunction().apply("test"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,106 +1,108 @@
|
||||
/*
|
||||
* Copyright 2017 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
|
||||
*
|
||||
* https://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.testing.processor;
|
||||
|
||||
import org.hamcrest.Matcher;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.boot.test.mock.mockito.SpyBean;
|
||||
import org.springframework.cloud.stream.messaging.Processor;
|
||||
import org.springframework.cloud.stream.test.binder.MessageCollector;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.hamcrest.CoreMatchers.not;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.mockito.Matchers.anyString;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesMessageThat;
|
||||
import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesPayloadThat;
|
||||
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
|
||||
|
||||
/**
|
||||
* The Spring Boot-base test-case to demonstrate how can we test Spring Cloud Stream applications
|
||||
* with available testing tools.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
@DirtiesContext
|
||||
@Ignore
|
||||
public class ToUpperCaseProcessorTests {
|
||||
|
||||
@Autowired
|
||||
private Processor channels;
|
||||
|
||||
@Autowired
|
||||
private MessageCollector collector;
|
||||
|
||||
@SpyBean
|
||||
private ToUpperCaseProcessor toUpperCaseProcessor;
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testMessages() {
|
||||
SubscribableChannel input = this.channels.input();
|
||||
|
||||
input.send(new GenericMessage<>("foo"));
|
||||
input.send(new GenericMessage<>("bar"));
|
||||
input.send(new GenericMessage<>("foo meets bar"));
|
||||
input.send(new GenericMessage<>("nothing but the best test"));
|
||||
|
||||
BlockingQueue<Message<?>> messages = this.collector.forChannel(channels.output());
|
||||
|
||||
assertThat(messages, receivesPayloadThat(is("FOO")));
|
||||
assertThat(messages, receivesPayloadThat(is("BAR")));
|
||||
assertThat(messages, receivesPayloadThat(is("FOO MEETS BAR")));
|
||||
assertThat(messages, receivesPayloadThat(not("nothing but the best test")));
|
||||
|
||||
Message<String> testMessage =
|
||||
MessageBuilder.withPayload("headers")
|
||||
.setHeader("foo", "bar")
|
||||
.build();
|
||||
|
||||
input.send(testMessage);
|
||||
|
||||
Message<String> expected =
|
||||
MessageBuilder.withPayload("HEADERS")
|
||||
.copyHeaders(testMessage.getHeaders())
|
||||
.build();
|
||||
|
||||
Matcher<Message<Object>> sameExceptIgnorableHeaders =
|
||||
(Matcher<Message<Object>>) (Matcher<?>) sameExceptIgnorableHeaders(expected);
|
||||
|
||||
assertThat(messages, receivesMessageThat(sameExceptIgnorableHeaders));
|
||||
|
||||
verify(this.toUpperCaseProcessor, times(5)).transform(anyString());
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2017 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
|
||||
*
|
||||
* https://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.testing.processor;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.hamcrest.CoreMatchers.not;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesMessageThat;
|
||||
import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesPayloadThat;
|
||||
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
|
||||
import org.hamcrest.Matcher;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.boot.actuate.autoconfigure.metrics.KafkaMetricsAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.jdbc.DataSourceTransactionManagerAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.transaction.TransactionAutoConfiguration;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.cloud.stream.test.binder.MessageCollector;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
|
||||
/**
|
||||
* The Spring Boot-base test-case to demonstrate how can we test Spring Cloud Stream applications
|
||||
* with available testing tools.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
@ImportAutoConfiguration(exclude = {
|
||||
KafkaAutoConfiguration.class,
|
||||
KafkaMetricsAutoConfiguration.class,
|
||||
DataSourceAutoConfiguration.class,
|
||||
TransactionAutoConfiguration.class,
|
||||
DataSourceTransactionManagerAutoConfiguration.class })
|
||||
@DirtiesContext
|
||||
class ToUpperCaseProcessorTests {
|
||||
|
||||
@Autowired
|
||||
@Qualifier("uppercaseFunction-in-0")
|
||||
private MessageChannel input;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("uppercaseFunction-out-0")
|
||||
private MessageChannel output;
|
||||
|
||||
@Autowired
|
||||
private MessageCollector collector;
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void testMessages() {
|
||||
this.input.send(new GenericMessage<>("odd"));
|
||||
this.input.send(new GenericMessage<>("even"));
|
||||
this.input.send(new GenericMessage<>("odd meets even"));
|
||||
this.input.send(new GenericMessage<>("nothing but the best test"));
|
||||
|
||||
BlockingQueue<Message<?>> messages = this.collector.forChannel(this.output);
|
||||
|
||||
assertThat(messages, receivesPayloadThat(is("ODD")));
|
||||
assertThat(messages, receivesPayloadThat(is("EVEN")));
|
||||
assertThat(messages, receivesPayloadThat(is("ODD MEETS EVEN")));
|
||||
assertThat(messages, receivesPayloadThat(not("nothing but the best test")));
|
||||
|
||||
Message<String> testMessage =
|
||||
MessageBuilder.withPayload("headers")
|
||||
.setHeader("odd", "even")
|
||||
.build();
|
||||
|
||||
input.send(testMessage);
|
||||
|
||||
Message<String> expected =
|
||||
MessageBuilder.withPayload("HEADERS")
|
||||
.copyHeaders(testMessage.getHeaders())
|
||||
.build();
|
||||
|
||||
Matcher<Message<Object>> sameExceptIgnorableHeaders =
|
||||
(Matcher<Message<Object>>) (Matcher<?>) sameExceptIgnorableHeaders(expected);
|
||||
|
||||
assertThat(messages, receivesMessageThat(sameExceptIgnorableHeaders));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2017 the original author or authors.
|
||||
* Copyright 2017-2019 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.
|
||||
@@ -16,28 +16,29 @@
|
||||
|
||||
package org.springframework.cloud.stream.testing.processor.integration;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import java.util.Iterator;
|
||||
|
||||
import org.apache.kafka.clients.consumer.Consumer;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecords;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.jdbc.DataSourceTransactionManagerAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.transaction.TransactionAutoConfiguration;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration;
|
||||
import org.springframework.cloud.stream.testing.processor.ToUpperCaseProcessor;
|
||||
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
|
||||
import org.springframework.kafka.core.KafkaTemplate;
|
||||
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
|
||||
import org.springframework.kafka.test.EmbeddedKafkaBroker;
|
||||
import org.springframework.kafka.test.context.EmbeddedKafka;
|
||||
import org.springframework.kafka.test.utils.KafkaTestUtils;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* A Spring Boot integration test for the Spring Cloud Stream Processor application
|
||||
@@ -46,24 +47,30 @@ import static org.assertj.core.api.Assertions.assertThat;
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest(
|
||||
properties = {
|
||||
"spring.autoconfigure.exclude=org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration",
|
||||
"spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer",
|
||||
"spring.cloud.stream.bindings.output.producer.headerMode=raw",
|
||||
"spring.cloud.stream.bindings.input.consumer.headerMode=raw",
|
||||
"spring.cloud.stream.bindings.input.group=embeddedKafkaApplication",
|
||||
"spring.cloud.stream.bindings.uppercaseFunction-out-0.destination=" + ToUpperCaseProcessorIntTests.TEST_TOPIC_OUT,
|
||||
"spring.cloud.stream.bindings.uppercaseFunction-in-0.group=embeddedKafkaApplication",
|
||||
"spring.cloud.stream.bindings.uppercaseFunction-in-0.destination=" + ToUpperCaseProcessorIntTests.TEST_TOPIC_IN,
|
||||
"spring.kafka.consumer.group-id=EmbeddedKafkaIntTest"
|
||||
},
|
||||
classes = ToUpperCaseProcessor.class,
|
||||
webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
@ImportAutoConfiguration(exclude = {
|
||||
TestSupportBinderAutoConfiguration.class,
|
||||
DataSourceAutoConfiguration.class,
|
||||
TransactionAutoConfiguration.class,
|
||||
DataSourceTransactionManagerAutoConfiguration.class })
|
||||
@EmbeddedKafka(controlledShutdown = true,
|
||||
bootstrapServersProperty = "spring.kafka.bootstrap-servers",
|
||||
topics = ToUpperCaseProcessorIntTests.TEST_TOPIC_OUT)
|
||||
@DirtiesContext
|
||||
@Ignore
|
||||
public class ToUpperCaseProcessorIntTests {
|
||||
class ToUpperCaseProcessorIntTests {
|
||||
|
||||
@ClassRule
|
||||
public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true, "output");
|
||||
static final String TEST_TOPIC_IN = "test_topic_in";
|
||||
|
||||
static final String TEST_TOPIC_OUT = "test_topic_out";
|
||||
|
||||
@Autowired
|
||||
private KafkaTemplate<byte[], byte[]> template;
|
||||
@@ -71,24 +78,22 @@ public class ToUpperCaseProcessorIntTests {
|
||||
@Autowired
|
||||
private DefaultKafkaConsumerFactory<byte[], String> consumerFactory;
|
||||
|
||||
@BeforeClass
|
||||
public static void setup() {
|
||||
System.setProperty("spring.kafka.bootstrap-servers", kafkaEmbedded.getEmbeddedKafka().getBrokersAsString());
|
||||
}
|
||||
@Autowired
|
||||
private EmbeddedKafkaBroker embeddedKafkaBroker;
|
||||
|
||||
@Test
|
||||
public void testMessagesOverKafka() throws Exception {
|
||||
this.template.send("input", "bar".getBytes());
|
||||
void testMessagesOverKafka() {
|
||||
this.template.send(TEST_TOPIC_IN, "test".getBytes());
|
||||
|
||||
Consumer<byte[], String> consumer = this.consumerFactory.createConsumer();
|
||||
|
||||
kafkaEmbedded.getEmbeddedKafka().consumeFromAnEmbeddedTopic(consumer, "output");
|
||||
embeddedKafkaBroker.consumeFromAnEmbeddedTopic(consumer, TEST_TOPIC_OUT);
|
||||
|
||||
ConsumerRecords<byte[], String> replies = KafkaTestUtils.getRecords(consumer);
|
||||
assertThat(replies.count()).isEqualTo(1);
|
||||
|
||||
Iterator<ConsumerRecord<byte[], String>> iterator = replies.iterator();
|
||||
assertThat(iterator.next().value()).isEqualTo("BAR");
|
||||
assertThat(iterator.next().value()).isEqualTo("TEST");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -25,23 +25,25 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.boot.actuate.autoconfigure.metrics.KafkaMetricsAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.boot.test.mock.mockito.SpyBean;
|
||||
import org.springframework.cloud.stream.messaging.Sink;
|
||||
import org.springframework.integration.channel.AbstractMessageChannel;
|
||||
import org.springframework.integration.test.mock.MockIntegration;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.support.ChannelInterceptorAdapter;
|
||||
import org.springframework.messaging.support.ChannelInterceptor;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
/**
|
||||
* The Spring Boot-base test-case to demonstrate how can we test Spring Cloud Stream applications
|
||||
@@ -50,13 +52,16 @@ import org.springframework.test.context.junit4.SpringRunner;
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
@ImportAutoConfiguration(exclude = {
|
||||
KafkaAutoConfiguration.class,
|
||||
KafkaMetricsAutoConfiguration.class })
|
||||
@DirtiesContext
|
||||
public class JdbcSinkTests {
|
||||
class JdbcSinkTests {
|
||||
|
||||
@Autowired
|
||||
private Sink channels;
|
||||
@Qualifier("jdbcConsumer-in-0")
|
||||
private AbstractMessageChannel input;
|
||||
|
||||
@Autowired
|
||||
private JdbcTemplate jdbcTemplate;
|
||||
@@ -65,47 +70,45 @@ public class JdbcSinkTests {
|
||||
private MessageHandler jdbcMessageHandler;
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testMessages() {
|
||||
AbstractMessageChannel input = (AbstractMessageChannel) this.channels.input();
|
||||
void testMessages() {
|
||||
AtomicReference<Message<?>> messageAtomicReference = new AtomicReference<>();
|
||||
|
||||
final AtomicReference<Message<?>> messageAtomicReference = new AtomicReference<>();
|
||||
ChannelInterceptor assertionInterceptor =
|
||||
new ChannelInterceptor() {
|
||||
|
||||
ChannelInterceptorAdapter assertionInterceptor = new ChannelInterceptorAdapter() {
|
||||
@Override
|
||||
public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent,
|
||||
Exception ex) {
|
||||
|
||||
@Override
|
||||
public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
|
||||
messageAtomicReference.set(message);
|
||||
super.afterSendCompletion(message, channel, sent, ex);
|
||||
}
|
||||
messageAtomicReference.set(message);
|
||||
}
|
||||
|
||||
};
|
||||
};
|
||||
|
||||
input.addInterceptor(assertionInterceptor);
|
||||
this.input.addInterceptor(assertionInterceptor);
|
||||
|
||||
input.send(new GenericMessage<>("foo"));
|
||||
this.input.send(new GenericMessage<>("odd"));
|
||||
|
||||
input.removeInterceptor(assertionInterceptor);
|
||||
this.input.removeInterceptor(assertionInterceptor);
|
||||
|
||||
input.send(new GenericMessage<>("bar"));
|
||||
this.input.send(new GenericMessage<>("even"));
|
||||
|
||||
List<Map<String, Object>> data = this.jdbcTemplate.queryForList("SELECT * FROM foobar");
|
||||
List<Map<String, Object>> data = this.jdbcTemplate.queryForList("SELECT * FROM SINK");
|
||||
|
||||
assertThat(data.size()).isEqualTo(2);
|
||||
assertThat(data.get(0).get("value")).isEqualTo("foo");
|
||||
assertThat(data.get(1).get("value")).isEqualTo("bar");
|
||||
assertThat(data.get(0).get("value")).isEqualTo("odd");
|
||||
assertThat(data.get(1).get("value")).isEqualTo("even");
|
||||
|
||||
Message<?> message1 = messageAtomicReference.get();
|
||||
assertThat(message1).isNotNull();
|
||||
assertThat(message1).hasFieldOrPropertyWithValue("payload", "foo");
|
||||
assertThat(message1).hasFieldOrPropertyWithValue("payload", "odd");
|
||||
|
||||
ArgumentCaptor<Message<?>> messageArgumentCaptor =
|
||||
(ArgumentCaptor<Message<?>>) (ArgumentCaptor<?>) ArgumentCaptor.forClass(Message.class);
|
||||
ArgumentCaptor<Message<?>> messageArgumentCaptor = MockIntegration.messageArgumentCaptor();
|
||||
|
||||
verify(this.jdbcMessageHandler, times(2)).handleMessage(messageArgumentCaptor.capture());
|
||||
|
||||
Message<?> message = messageArgumentCaptor.getValue();
|
||||
assertThat(message).hasFieldOrPropertyWithValue("payload", "bar");
|
||||
assertThat(message).hasFieldOrPropertyWithValue("payload", "even");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,65 +0,0 @@
|
||||
/*
|
||||
* Copyright 2017 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
|
||||
*
|
||||
* https://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.testing.source;
|
||||
|
||||
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesPayloadThat;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.cloud.stream.messaging.Source;
|
||||
import org.springframework.cloud.stream.test.binder.MessageCollector;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
/**
|
||||
* The Spring Boot-base test-case to demonstrate how can we test Spring Cloud Stream applications
|
||||
* with available testing tools.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
@DirtiesContext
|
||||
public class FooBarSourceTests {
|
||||
|
||||
@Autowired
|
||||
private Source channels;
|
||||
|
||||
@Autowired
|
||||
private MessageCollector collector;
|
||||
|
||||
@Test
|
||||
public void testMessages() {
|
||||
BlockingQueue<Message<?>> messages = this.collector.forChannel(channels.output());
|
||||
|
||||
assertThat(messages, receivesPayloadThat(is("foo")));
|
||||
assertThat(messages, receivesPayloadThat(is("bar")));
|
||||
assertThat(messages, receivesPayloadThat(is("foo")));
|
||||
assertThat(messages, receivesPayloadThat(is("bar")));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
/*
|
||||
* Copyright 2017-2019 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
|
||||
*
|
||||
* https://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.testing.source;
|
||||
|
||||
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesPayloadThat;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.boot.actuate.autoconfigure.metrics.KafkaMetricsAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.jdbc.DataSourceTransactionManagerAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.transaction.TransactionAutoConfiguration;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.cloud.stream.test.binder.MessageCollector;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
|
||||
/**
|
||||
* The Spring Boot-base test-case to demonstrate how can we test Spring Cloud Stream applications
|
||||
* with available testing tools.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
*/
|
||||
@SpringBootTest(
|
||||
webEnvironment = SpringBootTest.WebEnvironment.NONE,
|
||||
properties = "spring.cloud.stream.poller.fixed-delay=1")
|
||||
@ImportAutoConfiguration(exclude = {
|
||||
KafkaAutoConfiguration.class,
|
||||
KafkaMetricsAutoConfiguration.class,
|
||||
DataSourceAutoConfiguration.class,
|
||||
TransactionAutoConfiguration.class,
|
||||
DataSourceTransactionManagerAutoConfiguration.class })
|
||||
@DirtiesContext
|
||||
class OddEvenSourceTests {
|
||||
|
||||
@Autowired
|
||||
@Qualifier("oddEvenSupplier-out-0")
|
||||
MessageChannel outputDestination;
|
||||
|
||||
@Autowired
|
||||
MessageCollector collector;
|
||||
|
||||
@Test
|
||||
void testMessages() {
|
||||
BlockingQueue<Message<?>> messages = this.collector.forChannel(this.outputDestination);
|
||||
|
||||
assertThat(messages, receivesPayloadThat(is("odd")));
|
||||
assertThat(messages, receivesPayloadThat(is("even")));
|
||||
assertThat(messages, receivesPayloadThat(is("odd")));
|
||||
assertThat(messages, receivesPayloadThat(is("even")));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,3 +1,3 @@
|
||||
CREATE TABLE FOOBAR (
|
||||
CREATE TABLE SINK (
|
||||
value VARCHAR(256),
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user