diff --git a/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessor.java b/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessor.java index 7f9d484..871a425 100644 --- a/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessor.java +++ b/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessor.java @@ -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 uppercaseFunction() { + return String::toUpperCase; } public static void main(String[] args) { diff --git a/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/sink/JdbcSink.java b/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/sink/JdbcSink.java index a76bad3..1cec602 100644 --- a/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/sink/JdbcSink.java +++ b/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/sink/JdbcSink.java @@ -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); + } + +} diff --git a/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/source/FooBarSource.java b/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/source/OddEvenSource.java similarity index 54% rename from testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/source/FooBarSource.java rename to testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/source/OddEvenSource.java index b231a08..76c0e51 100644 --- a/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/source/FooBarSource.java +++ b/testing-samples/testing-demo/src/main/java/org/springframework/cloud/stream/testing/source/OddEvenSource.java @@ -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 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 oddEvenSupplier() { + return () -> this.semaphore.getAndSet(!this.semaphore.get()) ? "odd" : "even"; + } + + public static void main(String[] args) { + SpringApplication.run(OddEvenSource.class, args); + } + +} diff --git a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/NaiveToUpperCaseTests.java b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/NaiveToUpperCaseTests.java index 194b897..e543755 100644 --- a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/NaiveToUpperCaseTests.java +++ b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/NaiveToUpperCaseTests.java @@ -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")); + } + +} diff --git a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessorTests.java b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessorTests.java index 030587d..b9245ba 100644 --- a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessorTests.java +++ b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/ToUpperCaseProcessorTests.java @@ -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> 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 testMessage = - MessageBuilder.withPayload("headers") - .setHeader("foo", "bar") - .build(); - - input.send(testMessage); - - Message expected = - MessageBuilder.withPayload("HEADERS") - .copyHeaders(testMessage.getHeaders()) - .build(); - - Matcher> sameExceptIgnorableHeaders = - (Matcher>) (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> 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 testMessage = + MessageBuilder.withPayload("headers") + .setHeader("odd", "even") + .build(); + + input.send(testMessage); + + Message expected = + MessageBuilder.withPayload("HEADERS") + .copyHeaders(testMessage.getHeaders()) + .build(); + + Matcher> sameExceptIgnorableHeaders = + (Matcher>) (Matcher) sameExceptIgnorableHeaders(expected); + + assertThat(messages, receivesMessageThat(sameExceptIgnorableHeaders)); + } + +} diff --git a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/integration/ToUpperCaseProcessorIntTests.java b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/integration/ToUpperCaseProcessorIntTests.java index 8e477fa..1ea61e2 100644 --- a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/integration/ToUpperCaseProcessorIntTests.java +++ b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/processor/integration/ToUpperCaseProcessorIntTests.java @@ -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 template; @@ -71,24 +78,22 @@ public class ToUpperCaseProcessorIntTests { @Autowired private DefaultKafkaConsumerFactory 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 consumer = this.consumerFactory.createConsumer(); - kafkaEmbedded.getEmbeddedKafka().consumeFromAnEmbeddedTopic(consumer, "output"); + embeddedKafkaBroker.consumeFromAnEmbeddedTopic(consumer, TEST_TOPIC_OUT); ConsumerRecords replies = KafkaTestUtils.getRecords(consumer); assertThat(replies.count()).isEqualTo(1); Iterator> iterator = replies.iterator(); - assertThat(iterator.next().value()).isEqualTo("BAR"); + assertThat(iterator.next().value()).isEqualTo("TEST"); } } diff --git a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/sink/JdbcSinkTests.java b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/sink/JdbcSinkTests.java index 27a7749..27059ec 100644 --- a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/sink/JdbcSinkTests.java +++ b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/sink/JdbcSinkTests.java @@ -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> messageAtomicReference = new AtomicReference<>(); - final AtomicReference> 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> data = this.jdbcTemplate.queryForList("SELECT * FROM foobar"); + List> 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> messageArgumentCaptor = - (ArgumentCaptor>) (ArgumentCaptor) ArgumentCaptor.forClass(Message.class); + ArgumentCaptor> 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"); } } diff --git a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/source/FooBarSourceTests.java b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/source/FooBarSourceTests.java deleted file mode 100644 index f867131..0000000 --- a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/source/FooBarSourceTests.java +++ /dev/null @@ -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> 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"))); - } - -} diff --git a/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/source/OddEvenSourceTests.java b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/source/OddEvenSourceTests.java new file mode 100644 index 0000000..d642dc3 --- /dev/null +++ b/testing-samples/testing-demo/src/test/java/org/springframework/cloud/stream/testing/source/OddEvenSourceTests.java @@ -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> 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"))); + } + +} diff --git a/testing-samples/testing-demo/src/test/resources/schema.sql b/testing-samples/testing-demo/src/test/resources/schema.sql index fccb0a5..a4c7817 100644 --- a/testing-samples/testing-demo/src/test/resources/schema.sql +++ b/testing-samples/testing-demo/src/test/resources/schema.sql @@ -1,3 +1,3 @@ -CREATE TABLE FOOBAR ( +CREATE TABLE SINK ( value VARCHAR(256), );