initial commit for graphql sample

build.gradle

@@ -258,9 +258,11 @@ subprojects { subproject ->
 			openJpaVersion = '2.4.0'
 			oracleDriverVersion = '19.3.0.0'
 			postgresVersion = '42.2.14'
+			projectreactorTestVersion = '3.4.15'
 			slf4jVersion = '1.7.30'
 			springCloudVersion = '2022.0.0-SNAPSHOT'
-			springIntegrationVersion = '6.0.0-M1'
+			springGraphqlVersion = '1.0.0-M5'
+			springIntegrationVersion = '6.0.0-SNAPSHOT'
 			springIntegrationSocialTwiterVersion = '1.0.1.BUILD-SNAPSHOT'
 			springIntegrationSplunkVersion = '1.2.0.BUILD-SNAPSHOT'
 			springVersion = '6.0.0-M2'
@@ -1133,6 +1135,26 @@ project('file-processing') {
 	}
 }
 
+project('graphql') {
+	description = 'GraphQL Sample'
+
+	apply plugin: 'application'
+
+	mainClassName = 'org.springframework.integration.samples.graphql.Main'
+
+	dependencies {
+		api 'org.springframework:spring-webflux'
+		api "org.springframework.integration:spring-integration-graphql:$springIntegrationVersion"
+
+		testImplementation "io.projectreactor:reactor-test:$projectreactorTestVersion"
+	}
+
+	test {
+		useJUnitPlatform()
+	}
+
+}
+
 project('mail-attachments') {
 	description = 'Mail Attachment Sample'
 

intermediate/graphql/src/main/java/org/springframework/integration/samples/graphql/OrderController.java

@@ -0,0 +1,62 @@
+package org.springframework.integration.samples.graphql;
+
+import org.springframework.graphql.data.method.annotation.Argument;
+import org.springframework.graphql.data.method.annotation.QueryMapping;
+import org.springframework.graphql.data.method.annotation.SchemaMapping;
+import org.springframework.stereotype.Controller;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Controller
+public class OrderController {
+
+    static final Map<String, Order> orders = new ConcurrentHashMap<>();
+    static final Map<String, Customer> customers = new ConcurrentHashMap<>();
+
+    static {
+
+        customers.put("0", new Customer("0", "Dan"));
+        customers.put("1", new Customer("1", "Artem"));
+
+        for(var orderId = 1; orderId <= 10; orderId++) {
+            var customerId = String.valueOf(orderId % 2);
+            var order = new Order(String.valueOf(orderId), (orderId * 10), customerId);
+            orders.put(String.valueOf(orderId),order);
+            System.out.println(order);
+        }
+    }
+
+    @QueryMapping
+    Flux<Order> orders() {
+
+        return Mono.just(orders.values())
+                .flatMapIterable(values -> values);
+    }
+
+    @QueryMapping
+    Mono<Order> orderById(@Argument String orderId) {
+
+        return Mono.just(orders.get(orderId));
+    }
+
+    @SchemaMapping(typeName = "Customer")
+    Flux<Order> ordersByCustomer(Customer customer) {
+
+        return Flux.fromIterable(orders.values())
+                .filter(order -> order.customerId().equals(customer.customerId()));
+    }
+
+    @QueryMapping
+    Flux<Customer> customers() {
+
+        return Mono.just(customers.values())
+                .flatMapIterable(values -> values);
+    }
+
+}
+
+record Order(String orderId, double amount, String customerId) { }
+record Customer(String customerId, String name) { }

intermediate/graphql/src/main/resources/graphql/schema.graphqls

@@ -0,0 +1,17 @@
+type Query {
+    orders: [Order]
+    orderById(orderId: ID): Order
+    customers: [Customer]
+}
+
+type Order {
+    orderId: ID
+    amount: Float
+    customerId: ID
+}
+
+type Customer {
+    customerId: ID
+    name: String
+    ordersByCustomer: [Order]
+}

intermediate/graphql/src/test/java/org/springframework/integration/samples/graphql/GraphqlIntegrationFlowTests.java

@@ -0,0 +1,194 @@
+package org.springframework.integration.samples.graphql;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.graphql.GraphQlService;
+import org.springframework.graphql.RequestInput;
+import org.springframework.graphql.RequestOutput;
+import org.springframework.graphql.data.method.annotation.support.AnnotatedControllerConfigurer;
+import org.springframework.graphql.execution.ExecutionGraphQlService;
+import org.springframework.graphql.execution.GraphQlSource;
+import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.MessageChannels;
+import org.springframework.integration.graphql.outbound.GraphQlMessageHandler;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import java.time.Duration;
+import java.util.*;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.in;
+
+@SpringJUnitConfig
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
+public class GraphqlIntegrationFlowTests {
+
+    @Autowired
+    private FluxMessageChannel inputChannel;
+
+    @Autowired
+    private FluxMessageChannel resultChannel;
+
+    @Autowired
+    private PollableChannel errorChannel;
+
+    @Test
+    public void testQuery() {
+
+        StepVerifier verifier = StepVerifier.create(
+                        Flux.from(this.resultChannel)
+                                .map(Message::getPayload)
+                                .cast(RequestOutput.class)
+                )
+                .consumeNextWith(result -> {
+                    assertThat(result).isInstanceOf(RequestOutput.class);
+                    Map<String, Object> data = result.getData();
+                    Map<String, Object> query = (Map<String, Object>) data.get("orderById");
+                    assertThat((List) query.get("orders"))
+                            .filteredOn("orderId", in("1", "2", "3", "4", "5", "6", "7", "8", "9", "10"))
+                            .hasSize(10);
+                })
+                .thenCancel()
+                .verifyLater();
+
+        this.inputChannel.send(
+                MessageBuilder
+                        .withPayload(new RequestInput("{ orders { orderId, amount } }", null, Collections.emptyMap(), null, UUID.randomUUID().toString()))
+                        .build()
+        );
+    }
+
+    @Test
+    public void testQueryWithArgument() {
+
+        String fakeId = "1";
+        double fakeAmount = 10.00;
+
+        StepVerifier verifier = StepVerifier.create(
+                        Flux.from(this.resultChannel)
+                                .map(Message::getPayload)
+                                .cast(RequestOutput.class)
+                )
+                .consumeNextWith(result -> {
+                    assertThat(result).isInstanceOf(RequestOutput.class);
+                    Map<String, Object> data = result.getData();
+                    Map<String, Object> query = (Map<String, Object>) data.get("orderById");
+                    assertThat(query.get("orderId")).isEqualTo(fakeId);
+                    assertThat(query.get("amount")).isEqualTo(fakeAmount);
+                })
+                .thenCancel()
+                .verifyLater();
+
+        this.inputChannel.send(
+                MessageBuilder
+                        .withPayload(new RequestInput("{ orderById(orderId: \"" + fakeId + "\") { orderId, amount } }", null, Collections.emptyMap(), null, UUID.randomUUID().toString()))
+                        .build()
+        );
+
+        verifier.verify(Duration.ofSeconds(10));
+
+    }
+
+    @Test
+    public void testQueryWithSchemaMapping() {
+
+        StepVerifier verifier = StepVerifier.create(
+                        Flux.from(this.resultChannel)
+                                .map(Message::getPayload)
+                                .cast(RequestOutput.class)
+                )
+                .consumeNextWith(result -> {
+                    assertThat(result).isInstanceOf(RequestOutput.class);
+                    Map<String, Object> data = result.getData();
+                    List<Map<String, Object>> customers = (List<Map<String, Object>>) data.get("customers");
+                    assertThat(customers)
+                            .filteredOn("customerId", in("0", "1"))
+                            .hasSize(2);
+
+                    Map<String, Object> customer0 = (Map<String, Object>) customers.get(0);
+                    List<Map<String, Object>> ordersCustomer0 = (List<Map<String, Object>>) customer0.get("ordersByCustomer");
+                    assertThat(ordersCustomer0)
+                            .filteredOn("orderId", in("2", "4", "6", "8", "10"))
+                            .hasSize(5);
+                })
+                .thenCancel()
+                .verifyLater();
+
+        this.inputChannel.send(
+                MessageBuilder
+                        .withPayload(new RequestInput("{ customers { customerId, name, ordersByCustomer { orderId } } }", null, Collections.emptyMap(), null, UUID.randomUUID().toString()))
+                        .build()
+        );
+
+        verifier.verify(Duration.ofSeconds(10));
+
+    }
+
+    @Configuration
+    @EnableIntegration
+    static class TestConfig {
+
+        @Bean
+        OrderController orderController() {
+
+            return new OrderController();
+        }
+
+        @Bean
+        IntegrationFlow graphqlQueryMessageHandlerFlow(GraphQlMessageHandler handler) {
+
+            return IntegrationFlows.from(MessageChannels.flux("inputChannel"))
+                    .handle(handler)
+                    .channel(c -> c.flux("resultChannel"))
+                    .get();
+        }
+
+        @Bean
+        GraphQlMessageHandler handler(GraphQlService graphQlService) {
+
+            return new GraphQlMessageHandler(graphQlService);
+        }
+
+        @Bean
+        GraphQlService graphQlService(GraphQlSource graphQlSource) {
+
+            return new ExecutionGraphQlService(graphQlSource);
+        }
+
+        @Bean
+        GraphQlSource graphQlSource(AnnotatedControllerConfigurer annotatedDataFetcherConfigurer) {
+
+            return GraphQlSource.builder()
+                    .schemaResources(new ClassPathResource("graphql/schema.graphqls"))
+                    .configureRuntimeWiring(annotatedDataFetcherConfigurer)
+                    .build();
+        }
+
+        @Bean
+        AnnotatedControllerConfigurer annotatedDataFetcherConfigurer() {
+
+            return new AnnotatedControllerConfigurer();
+        }
+
+        @Bean
+        PollableChannel errorChannel() {
+
+            return new QueueChannel();
+        }
+
+    }
+
+}