Add more Spring Integration native hints

artembilan · sdeleuze · commit bdb85a032d8b · 2021-07-23T09:00:56.000+02:00
* The `IntegrationFlow` interface can be proxied at runtime * The resource pattern for JDBC schemas should really be a regex pattern * The HTTP and WebFlux request mappings need a reflection access to method of their handlers to map * Process interfaces with a `@MessagingGateway` and register their proxy hints * Improve `IntegrationApplication` in the sample to cover the mentioned above cases about an `IntegrationFlow` proxy and `@MessagingGateway` * Fix `ControlBusGateway` to emit really a Control Bus command * Use `WebClient.Builder` bean from the ctx for application logic * Add reflection info for `AbstractEndpoint` to satisfy Control Bus requirements * Add some common types to satisfy common SpEL expressions * Add explicit `@TypeHint` for `Message` impls - * Add more Spring Integration types specific to `MessageStore` serialization functionality * Add programmatic scan for `IntegrationNode` impls instead of huge number of explicit types in `@TypeHint` * Add `JdbcChannelMessageStore` logic into the sample to cover its serialization functionality * Add `@SerializationHint` for messaging types * Add programmatic scan for `IntegrationNode` impls instead of huge number of explicit types in `@TypeHint` * Add `JdbcChannelMessageStore` logic into the sample to cover its serialization functionality * Add `RedisChannelMessageStore` logic into the sample to cover JSON (un)marshaling feature * Rely on Spring Integration version from parent in the sample Closes gh-894
diff --git a/samples/integration/docker-compose.yml b/samples/integration/docker-compose.yml
@@ -1,6 +1,14 @@
 version: '3.1'
 services:
+  redis:
+    image: 'redis:latest'
+    ports:
+      - '6379:6379'
   integration:
     image: integration:0.0.1-SNAPSHOT
     ports:
-      - "8080:8080"
+      - '8080:8080'
+    depends_on:
+      - redis
+    environment:
+      - SPRING_REDIS_HOST=redis
diff --git a/samples/integration/pom.xml b/samples/integration/pom.xml
@@ -17,7 +17,6 @@
 
 	<properties>
 		<java.version>11</java.version>
-		<spring-integration.version>5.5.0-SNAPSHOT</spring-integration.version>
 	</properties>
 
 	<dependencies>
@@ -29,18 +28,35 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-integration</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-redis</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-webflux</artifactId>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
-			<artifactId>spring-integration-http</artifactId>
+			<artifactId>spring-integration-webflux</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-jdbc</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-redis</artifactId>
 		</dependency>
 		<dependency>
 			<groupId>io.micrometer</groupId>
 			<artifactId>micrometer-core</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.h2database</groupId>
+			<artifactId>h2</artifactId>
+			<scope>runtime</scope>
+		</dependency>
 	</dependencies>
 
 	<build>
diff --git a/samples/integration/src/main/java/com/example/integration/ControlBusGateway.java b/samples/integration/src/main/java/com/example/integration/ControlBusGateway.java
@@ -0,0 +1,12 @@
+package com.example.integration;
+
+import org.springframework.integration.annotation.Gateway;
+import org.springframework.integration.annotation.MessagingGateway;
+
+@MessagingGateway(defaultRequestChannel = "controlBus.input")
+public interface ControlBusGateway {
+
+	@Gateway(payloadExpression = "'@' + args[0] + '.start()'")
+	void startEndpoint(String id);
+
+}
diff --git a/samples/integration/src/main/java/com/example/integration/IntegrationApplication.java b/samples/integration/src/main/java/com/example/integration/IntegrationApplication.java
@@ -4,11 +4,16 @@
 import java.util.Calendar;
 import java.util.Date;
 
+import javax.sql.DataSource;
+
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.ApplicationContext;
 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.http.MediaType;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.interceptor.WireTap;
@@ -17,12 +22,19 @@
 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.dsl.IntegrationFlows;
-import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 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.scheduling.PollerMetadata;
+import org.springframework.integration.support.json.JacksonJsonUtils;
+import org.springframework.integration.webflux.dsl.WebFlux;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.web.reactive.function.client.WebClient;
 
 import io.micrometer.core.instrument.MeterRegistry;
@@ -35,15 +47,32 @@
 public class IntegrationApplication {
 
 	public static void main(String[] args) throws InterruptedException {
-		ConfigurableApplicationContext applicationContext = SpringApplication.run(IntegrationApplication.class, args);
-		applicationContext.getBean("dateSourceEndpoint", SourcePollingChannelAdapter.class).start();
+		ApplicationContext context = SpringApplication.run(IntegrationApplication.class, args);
+
+		WebClient webClient =
+				context.getBean(WebClient.Builder.class)
+						.baseUrl("http://localhost:8080")
+						.build();
+
+		Thread.sleep(1000);
+
+		System.out.println("Starting 'dateSourceEndpoint'...");
+
+		webClient
+				.get()
+				.uri("control-bus/dateSourceEndpoint")
+				.retrieve()
+				.toBodilessEntity()
+				.block(Duration.ofSeconds(10));
 
 		Thread.sleep(1000);
 
+		System.out.println("Obtaining integration graph...");
+
 		String integrationGraph =
-				WebClient.create()
+				webClient
 						.get()
-						.uri("http://localhost:8080/integration-graph")
+						.uri("integration-graph")
 						.accept(MediaType.APPLICATION_JSON)
 						.retrieve()
 						.bodyToMono(String.class)
@@ -58,14 +87,38 @@ MeterRegistry simpleMeterRegistry() {
 	}
 
 	@Bean
-	IntegrationFlow printFormattedSecondsFlow() {
+	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(PollerMetadata.DEFAULT_POLLER)
+	PollerMetadata defaultPoller() {
+		PollerMetadata pollerMetadata = new PollerMetadata();
+		pollerMetadata.setTrigger(new PeriodicTrigger(10));
+		return pollerMetadata;
+	}
+
+	@Bean
+	IntegrationFlow printFormattedSecondsFlow(JdbcChannelMessageStore jdbcChannelMessageStore,
+			RedisChannelMessageStore redisChannelMessageStore) {
 		return IntegrationFlows
 				.fromSupplier(Date::new,
 						e -> e.id("dateSourceEndpoint")
 								.poller(p -> p.fixedDelay(1000, 1000)))
-				.channel("dateChannel")
+				.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();
 	}
@@ -99,4 +152,19 @@ public Integer convert(Date date) {
 		};
 	}
 
+	@Bean
+	public IntegrationFlow controlBus() {
+		return IntegrationFlowDefinition::controlBus;
+	}
+
+	@Bean
+	public IntegrationFlow controlBusControllerFlow(ControlBusGateway controlBusGateway) {
+		return IntegrationFlows
+				.from(WebFlux.inboundChannelAdapter("/control-bus/{endpointId}")
+						.payloadExpression("#pathVariables.endpointId")
+						.requestMapping(mapping -> mapping.methods(HttpMethod.GET)))
+				.handle(controlBusGateway, "startEndpoint")
+				.get();
+	}
+
 }
diff --git a/spring-native-configuration/pom.xml b/spring-native-configuration/pom.xml
@@ -470,6 +470,11 @@
             <artifactId>spring-integration-http</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-webflux</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework.integration</groupId>
             <artifactId>spring-integration-websocket</artifactId>
@@ -480,6 +485,11 @@
             <artifactId>spring-integration-xml</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-file</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework.ws</groupId>
             <artifactId>spring-ws-core</artifactId>
diff --git a/spring-native-configuration/src/main/java/org/springframework/integration/IntegrationHints.java b/spring-native-configuration/src/main/java/org/springframework/integration/IntegrationHints.java
