Enable use of Kotlin Flow as return value

[rstoyanchev]

Using ReactiveAdapterRegistry when handling return values from annotated controller methods such as @SchemaMapping, @BatchMapping, and @GraphQlExceptionHandler would enable use of Kotlin's Flow.

[hantsy]

Also need to enable Flow in Subscription, and batch data loader(instead of CompletableFuture).

[koenpunt]

@hantsy batch data loader already works with Monos, so you could use the mono {} helper from kotlinx.coroutines.reactor, e.g.;

registry.forName<UUID, Merchant>("merchantDataLoader").registerMappedBatchLoader { merchantIds, _ ->
    mono {
        merchantService.findAllById(merchantIds).associateBy { it.id }
    }
}

Or for a method returning a flow, it can be returned as Flux;

registry.forName<UUID, Merchant>("merchantDataLoader").registerBatchLoader { merchantIds, _ ->
    merchantService.findAllById(merchantIds).asFlux()
}

[rstoyanchev]

@SubscriptionMapping is just a shortcut annotation for @SchemaMapping(typeName="subscription"), but I confirm it will support subscriptions.

[hantsy]

I tried to update my example project to use Flow.

@QueryMapping
fun allPosts(): Flow<Post> = postService.allPosts()// remove .toList() to return Flow type directly

But run the following tests in the QueryTests.

@Test
  fun `get all posts`() = runTest {
      coEvery { postService.allPosts() } returns
              flowOf(
                  Post(
                      id = UUID.randomUUID(),
                      title = "Post 1",
                      content = "Post 1 content",
                      status = PostStatus.DRAFT,
                      createdAt = LocalDateTime.now()
                  ),
                  Post(
                      id = UUID.randomUUID(),
                      title = "Post 2",
                      content = "Post 2 content",
                      status = PostStatus.DRAFT,
                      createdAt = LocalDateTime.now()
                  )
              )
      val query = "{ allPosts { title content }}"
//         graphQlTester.document(query).execute()
//             .errors().satisfy { it.forEach { error -> log.debug("error message: ${error.message}") } }
      graphQlTester.document(query)
         .execute()
         .path("data.allPosts[*].title")
         .entityList(String::class.java).hasSize(2).contains("POST 1", "POST 2")

      coVerify(exactly = 1) { postService.allPosts() }
  }

And I got the error [TypeMismatchError{path=[allPosts], expectedType=[Post!]}].

The example project here: https://github.com/hantsy/spring-graphql-sample/tree/master/spring-graphql-rsocket-kotlin-co

[koenpunt]

@hantsy is your postService.getPosts() defined as suspend fun? Then that might be the problem, because returning a flow doesn't require the method to be a suspend fun, only the consumer of the flow needs to be in a coroutine context.

[hantsy]

@koenpunt Check here, https://github.com/hantsy/spring-graphql-sample/blob/master/spring-graphql-rsocket-kotlin-co/src/main/kotlin/com/example/demo/Services.kt#L52

[hantsy]

Created an new issue for this.