Add smoke test for Spring Integration with weblux and data

Closes gh-10

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
+}

integration/integration-webfux-data/docker-compose.yml

@@ -0,0 +1,6 @@
+version: '3.1'
+services:
+  redis:
+    image: 'redis:7'
+    ports:
+      - '6379'

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"));
+	}
+
+}

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<Date, Integer> currentSeconds() {
+		return new Converter<Date, Integer>() { // 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);
+
+	}
+
+}

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}
+