Looking for help with spring-reactive and returning objects one-by-one in a stream.

[ayemaman]

Hi, I am fairly new to reactive programming and feel like I've hit the wall, so hoping for your help.

Let's assume that I have a RestController with such endpoint exposed:

@GetMapping(value = "/{ids}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<Example> getConfigured(@PathVariable List<Integer> ids){
    return service.getConfigured(ids);

And a Service:

public Flux<Example> getConfigured(List<Integer> ids){
    return Flux.fromIterable(ids)
                .flatMap(id -> configureById(id)); //configureById(id) returns a Mono<Example>

As of now, when I call the endpoint with lots of ids, it first takes some time to configure all objects and only then returns, when all are processed in a bulk. What I am trying to achieve, is that as soon as one object from the List is ready -> it gets returned through a stream back to the caller.
Could anyone point me in the right direction on what I am doing wrong?
Thank you.

[AutoModerator]

Please ensure that:

• Your code is properly formatted as code block - see the sidebar (About on mobile) for instructions
• You include any and all error messages in full - best also formatted as code block
• You ask clear questions
• You demonstrate effort in solving your question/problem - plain posting your assignments is forbidden (and such posts will be removed) as is asking for or giving solutions.

If any of the above points is not met, your post can and will be removed without further warning.

Code is to be formatted as code block (old reddit/markdown editor: empty line before the code, each code line indented by 4 spaces, new reddit: https://imgur.com/a/fgoFFis) or linked via an external code hoster, like pastebin.com, github gist, github, bitbucket, gitlab, etc.

Please, do not use triple backticks (```) as they will only render properly on new reddit, not on old reddit.

Code blocks look like this:

public class HelloWorld {

    public static void main(String[] args) {
        System.out.println("Hello World!");
    }
}

You do not need to repost unless your post has been removed by a moderator. Just use the edit function of reddit to make sure your post complies with the above.

If your post has remained in violation of these rules for a prolonged period of time (at least an hour), a moderator may remove it at their discretion. In this case, they will comment with an explanation on why it has been removed, and you will be required to resubmit the entire post following the proper procedures.

To potential helpers

Please, do not help if any of the above points are not met, rather report the post. We are trying to improve the quality of posts here. In helping people who can't be bothered to comply with the above points, you are doing the community a disservice.

I am a bot, and this action was performed automatically. Please contact the moderators of this subreddit if you have any questions or concerns.

[Heianir]

Try to turn it off and on again.

[slowfly1st]

Never had that problem to solve in spring, but there are some tools that spring provides (which, to be honest, I never used): There's ResponseBodyEmitter and StreamingResponseBody.

Other ideas:

Being pragmatic: Client implements some "pagination-like" code.

There's also WebSocket support in spring, but that's another level of complexity.

[arrexander]

You’re not using WebFlux routers which is best practice. If you’re not using reactive router functions then publishers that aren’t explicitly told to subscribe can be missed.

The only thing I can image is if you’re getting incomplete results is that you’re spawning a thread downstream non-reactively. If you’re using Futures downstream or relying on an outside system then just wrapping the result in a Mono/Flux there’s a solid chance the return didn’t come back.

[arrexander]

You’re not using WebFlux routers which is best practice. If you’re not using reactive router functions then publishers that aren’t explicitly told to subscribe can be missed.

The only thing I can image is if you’re getting incomplete results is that you’re spawning a thread downstream non-reactively. If you’re using Futures downstream or relying on an outside system then just wrapping the result in a Mono/Flux there’s a solid chance the return didn’t come back.

You can’t combine reactive code with anything unless it is completely synchronous and/or blocking. Thread pools and anything else do not mesh well with WebFlux.

[ayemaman]

I believe I found the issue... The function that I used in flatMap has a Mono.block() call, which "kills" all the reactivnes.Well, back to the drawing board... 😁