From 3f516b565a8d75970c03fd10f85a240e8a628f17 Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Thu, 3 Aug 2023 15:08:26 +0200 Subject: [PATCH] Add smoke test for Spring Integration with weblux and data Closes gh-10 --- .../integration-webfux-data/build.gradle | 39 +++++ .../docker-compose.yml | 6 + .../IntegrationApplicationTests.java | 55 +++++++ .../integration/IntegrationApplication.java | 150 ++++++++++++++++++ .../src/main/resources/application.properties | 5 + 5 files changed, 255 insertions(+) create mode 100644 integration/integration-webfux-data/build.gradle create mode 100644 integration/integration-webfux-data/docker-compose.yml create mode 100644 integration/integration-webfux-data/src/appTest/java/com/example/integration/IntegrationApplicationTests.java create mode 100644 integration/integration-webfux-data/src/main/java/com/example/integration/IntegrationApplication.java create mode 100644 integration/integration-webfux-data/src/main/resources/application.properties diff --git a/integration/integration-webfux-data/build.gradle b/integration/integration-webfux-data/build.gradle new file mode 100644 index 0000000..aa1633b --- /dev/null +++ b/integration/integration-webfux-data/build.gradle @@ -0,0 +1,39 @@ +plugins { + id "java" + id "org.springframework.boot" + id "org.springframework.cr.smoke-test" +} + +dependencies { + implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES)) + implementation("org.springframework.boot:spring-boot-starter-integration") + implementation("org.springframework.boot:spring-boot-starter-data-redis-reactive") + implementation("org.springframework.boot:spring-boot-starter-webflux") + //NOTE: Netty is not CR compatible at the moment + implementation("org.springframework.boot:spring-boot-starter-undertow") + modules { + module("org.springframework.boot:spring-boot-starter-reactor-netty") { + replacedBy("org.springframework.boot:spring-boot-starter-undertow", "Use Undertow instead of Netty") + } + } + + implementation("org.springframework.integration:spring-integration-webflux") + implementation("org.springframework.integration:spring-integration-jdbc") + implementation("org.springframework.integration:spring-integration-redis") + implementation("io.micrometer:micrometer-core") + implementation("com.jayway.jsonpath:json-path") + + implementation("org.crac:crac:$cracVersion") + implementation(project(":cr-listener")) + + runtimeOnly("com.h2database:h2") + + testImplementation("org.springframework.boot:spring-boot-starter-test") + + appTestImplementation(project(":cr-smoke-test-support")) + appTestImplementation("org.awaitility:awaitility:4.2.0") +} + +crSmokeTest { + webApplication = true +} diff --git a/integration/integration-webfux-data/docker-compose.yml b/integration/integration-webfux-data/docker-compose.yml new file mode 100644 index 0000000..17115c3 --- /dev/null +++ b/integration/integration-webfux-data/docker-compose.yml @@ -0,0 +1,6 @@ +version: '3.1' +services: + redis: + image: 'redis:7' + ports: + - '6379' diff --git a/integration/integration-webfux-data/src/appTest/java/com/example/integration/IntegrationApplicationTests.java b/integration/integration-webfux-data/src/appTest/java/com/example/integration/IntegrationApplicationTests.java new file mode 100644 index 0000000..bd3ae95 --- /dev/null +++ b/integration/integration-webfux-data/src/appTest/java/com/example/integration/IntegrationApplicationTests.java @@ -0,0 +1,55 @@ +/* + * Copyright 2023 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 com.example.integration; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; + +import org.springframework.cr.smoketest.support.assertj.AssertableOutput; +import org.springframework.cr.smoketest.support.junit.ApplicationTest; +import org.springframework.http.MediaType; +import org.springframework.test.web.reactive.server.WebTestClient; + +@ApplicationTest +public class IntegrationApplicationTests { + + @Test + void shouldOutputIntegrationGraph(WebTestClient client, AssertableOutput output) { + client.get().uri("/control-bus/dateSourceEndpoint").exchange().expectStatus().isOk(); + + output.assertThat().hasSingleLineContaining("Starting endpoint: dateSourceEndpoint"); + + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> output.assertThat().hasLineContaining("Current seconds:")); + + client.get() + .uri("/integration-graph") + .accept(MediaType.APPLICATION_JSON) + .exchange() + .expectStatus() + .isOk() + .expectBody(String.class) + .value(graph -> assertThat(graph).contains("null-channel") + .contains("loggingChannel") + .contains("dateSourceEndpoint")); + } + +} diff --git a/integration/integration-webfux-data/src/main/java/com/example/integration/IntegrationApplication.java b/integration/integration-webfux-data/src/main/java/com/example/integration/IntegrationApplication.java new file mode 100644 index 0000000..06b8ddb --- /dev/null +++ b/integration/integration-webfux-data/src/main/java/com/example/integration/IntegrationApplication.java @@ -0,0 +1,150 @@ +/* + * Copyright 2023 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 com.example.integration; + +import java.util.Calendar; +import java.util.Date; + +import javax.sql.DataSource; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.core.convert.converter.Converter; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer; +import org.springframework.http.HttpMethod; +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.interceptor.WireTap; +import org.springframework.integration.config.EnableIntegrationManagement; +import org.springframework.integration.config.EnableMessageHistory; +import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.config.IntegrationConverter; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlowDefinition; +import org.springframework.integration.handler.LoggingHandler; +import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice; +import org.springframework.integration.http.config.EnableIntegrationGraphController; +import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; +import org.springframework.integration.jdbc.store.channel.H2ChannelMessageStoreQueryProvider; +import org.springframework.integration.redis.store.RedisChannelMessageStore; +import org.springframework.integration.support.json.JacksonJsonUtils; +import org.springframework.integration.webflux.dsl.WebFlux; +import org.springframework.messaging.MessageHandler; + +@SpringBootApplication(proxyBeanMethods = false) +@EnableMessageHistory("dateChannel") +@EnableIntegrationManagement +@EnableIntegrationGraphController("/integration-graph") +public class IntegrationApplication { + + public static void main(String[] args) { + SpringApplication.run(IntegrationApplication.class, args); + } + + @Bean + MeterRegistry simpleMeterRegistry() { + return new SimpleMeterRegistry(); + } + + @Bean + JdbcChannelMessageStore jdbcChannelMessageStore(DataSource dataSource) { + JdbcChannelMessageStore jdbcChannelMessageStore = new JdbcChannelMessageStore(dataSource); + jdbcChannelMessageStore.setChannelMessageStoreQueryProvider(new H2ChannelMessageStoreQueryProvider()); + return jdbcChannelMessageStore; + } + + @Bean + RedisChannelMessageStore redisChannelMessageStore(RedisConnectionFactory connectionFactory) { + RedisChannelMessageStore redisChannelMessageStore = new RedisChannelMessageStore(connectionFactory); + redisChannelMessageStore + .setValueSerializer(new GenericJackson2JsonRedisSerializer(JacksonJsonUtils.messagingAwareMapper())); + return redisChannelMessageStore; + } + + @Bean + IntegrationFlow printFormattedSecondsFlow(JdbcChannelMessageStore jdbcChannelMessageStore, + RedisChannelMessageStore redisChannelMessageStore) { + + return IntegrationFlow + .fromSupplier(Date::new, e -> e.id("dateSourceEndpoint").poller(p -> p.fixedDelay(1000, 1000))) + .channel(c -> c.queue("dateChannel", jdbcChannelMessageStore, "dateChannelGroup")) + .gateway(subflow -> subflow.convert(Integer.class, e -> e.advice(new RequestHandlerRetryAdvice()))) + .channel(c -> c.queue(redisChannelMessageStore, "secondsChannelGroup")) + .handle(m -> System.out.println("Current seconds: " + m.getPayload())) + .get(); + } + + @Bean + @GlobalChannelInterceptor(patterns = "dateChannel") + WireTap loggingWireTap() { + return new WireTap("loggingChannel"); + } + + @Bean + @ServiceActivator(inputChannel = "loggingChannel") + MessageHandler loggingHandler() { + LoggingHandler loggingHandler = new LoggingHandler(LoggingHandler.Level.TRACE); + loggingHandler.setLoggerName("tracing.data"); + loggingHandler.setLogExpressionString("#jsonPath(payload.toString(), '$')"); + return loggingHandler; + } + + @Bean + @IntegrationConverter + Converter currentSeconds() { + return new Converter() { // Not lambda for generic info presence + + @Override + public Integer convert(Date date) { + Calendar calendar = Calendar.getInstance(); + calendar.setTime(date); + return calendar.get(Calendar.SECOND); + } + + }; + } + + @Bean + public IntegrationFlow controlBus() { + return IntegrationFlowDefinition::controlBus; + } + + @Bean + public IntegrationFlow controlBusControllerFlow(ControlBusGateway controlBusGateway) { + return IntegrationFlow + .from(WebFlux.inboundChannelAdapter("/control-bus/{endpointId}") + .payloadExpression("#pathVariables.endpointId") + .requestMapping(mapping -> mapping.methods(HttpMethod.GET))) + .wireTap(subflow -> subflow.handle(m -> System.out.println("Starting endpoint: " + m.getPayload()))) + .handle(controlBusGateway, "startEndpoint") + .get(); + } + + @MessagingGateway(defaultRequestChannel = "controlBus.input") + public interface ControlBusGateway { + + @Gateway(payloadExpression = "'@' + args[0] + '.start()'") + void startEndpoint(String id); + + } + +} diff --git a/integration/integration-webfux-data/src/main/resources/application.properties b/integration/integration-webfux-data/src/main/resources/application.properties new file mode 100644 index 0000000..c3ad28e --- /dev/null +++ b/integration/integration-webfux-data/src/main/resources/application.properties @@ -0,0 +1,5 @@ +spring.integration.endpoint.no-auto-startup=dateSourceEndpoint +logging.level.tracing.data=trace +spring.data.redis.host=${REDIS_HOST:localhost} +spring.data.redis.port=${REDIS_PORT_6379:6379} +