Spring webflux: Race of parallel calls and cancel/return when one of the responses has data

I have three parallel calls to three different endpoints, and only one of them is going to return the data that I like to process (the response will have data only for enpoint-1, enpoint-2 or endpoint-3).
If one of this calls return data, I'd like to return this data immediately and forget about the other calls...¿How can I achieve this with Spring Webflux?

I have the three calls with:

Mono<MyResponse> result = client.post()

  .uri('uri')

  .body(Mono.just(request), MyRequest.class)

  .retrieve()

  .bodyToMono(MyResponse.class);

Added to a Mono list

 List<Mono<MyResponse>> calls

And I have a merge of all the responses and look for the first item:

Flux.merge(calls).toStream().forEach(response -> myResponseList.addAll(response));

But I would like to return the data as soon as any of the calls return something different than null and not wait for the 'merge' to be completed.

Thanks!

asked Nov 14 '18 at 10:53

arturo.galan

314

The code above should work. Flux.merge(calls) should return the data as soon as response from any of the monos is available and pass it to the forEach method. Maybe i don't understand what is that you are trying to accomplish. I run the simple example: Mono<Long> delay1 = Mono.delay(Duration.ofMillis(1000)); Mono<Long> delay2 = Mono.delay(Duration.ofMillis(2000)); List<Mono<Long>> listOfMonos = new ArrayList<>(); listOfMonos.add(delay1); listOfMonos.add(delay2); Flux.merge(listOfMonos).toStream().forEach(System.out::println);`

– piotr szybicki
Nov 14 '18 at 11:12

Thanks piotr, I want to check the reponses as they come in and return if the response is not null. So what I have to do is to check in the foreach if the response is not null and ¿then break the foreach? or ¿maybe something like a filter.findFirst() to return when some data is found?

– arturo.galan
Nov 14 '18 at 11:30

I think in that case this is your best option: Long aLong = Flux.merge(listOfMonos).filter((t) -> t != null).toStream().findFirst().get();

– piotr szybicki
Nov 14 '18 at 11:59

add a comment |

I have the three calls with:

Mono<MyResponse> result = client.post()

  .uri('uri')

  .body(Mono.just(request), MyRequest.class)

  .retrieve()

  .bodyToMono(MyResponse.class);

Added to a Mono list

 List<Mono<MyResponse>> calls

And I have a merge of all the responses and look for the first item:

Flux.merge(calls).toStream().forEach(response -> myResponseList.addAll(response));

But I would like to return the data as soon as any of the calls return something different than null and not wait for the 'merge' to be completed.

Thanks!

asked Nov 14 '18 at 10:53

arturo.galan

314

The code above should work. Flux.merge(calls) should return the data as soon as response from any of the monos is available and pass it to the forEach method. Maybe i don't understand what is that you are trying to accomplish. I run the simple example: Mono<Long> delay1 = Mono.delay(Duration.ofMillis(1000)); Mono<Long> delay2 = Mono.delay(Duration.ofMillis(2000)); List<Mono<Long>> listOfMonos = new ArrayList<>(); listOfMonos.add(delay1); listOfMonos.add(delay2); Flux.merge(listOfMonos).toStream().forEach(System.out::println);`

– piotr szybicki
Nov 14 '18 at 11:12

Thanks piotr, I want to check the reponses as they come in and return if the response is not null. So what I have to do is to check in the foreach if the response is not null and ¿then break the foreach? or ¿maybe something like a filter.findFirst() to return when some data is found?

– arturo.galan
Nov 14 '18 at 11:30

I think in that case this is your best option: Long aLong = Flux.merge(listOfMonos).filter((t) -> t != null).toStream().findFirst().get();

– piotr szybicki
Nov 14 '18 at 11:59

add a comment |

I have the three calls with:

Mono<MyResponse> result = client.post()

  .uri('uri')

  .body(Mono.just(request), MyRequest.class)

  .retrieve()

  .bodyToMono(MyResponse.class);

Added to a Mono list

 List<Mono<MyResponse>> calls

And I have a merge of all the responses and look for the first item:

Flux.merge(calls).toStream().forEach(response -> myResponseList.addAll(response));

But I would like to return the data as soon as any of the calls return something different than null and not wait for the 'merge' to be completed.

Thanks!

asked Nov 14 '18 at 10:53

arturo.galan

314

I have the three calls with:

Mono<MyResponse> result = client.post()

  .uri('uri')

  .body(Mono.just(request), MyRequest.class)

  .retrieve()

  .bodyToMono(MyResponse.class);

Added to a Mono list

 List<Mono<MyResponse>> calls

And I have a merge of all the responses and look for the first item:

Flux.merge(calls).toStream().forEach(response -> myResponseList.addAll(response));

But I would like to return the data as soon as any of the calls return something different than null and not wait for the 'merge' to be completed.

Thanks!

spring spring-webflux

asked Nov 14 '18 at 10:53

arturo.galan

314

asked Nov 14 '18 at 10:53

arturo.galan

314

asked Nov 14 '18 at 10:53

arturo.galan

314

asked Nov 14 '18 at 10:53

arturo.galan

314

asked Nov 14 '18 at 10:53

arturo.galan

314

The code above should work. Flux.merge(calls) should return the data as soon as response from any of the monos is available and pass it to the forEach method. Maybe i don't understand what is that you are trying to accomplish. I run the simple example: Mono<Long> delay1 = Mono.delay(Duration.ofMillis(1000)); Mono<Long> delay2 = Mono.delay(Duration.ofMillis(2000)); List<Mono<Long>> listOfMonos = new ArrayList<>(); listOfMonos.add(delay1); listOfMonos.add(delay2); Flux.merge(listOfMonos).toStream().forEach(System.out::println);`

– piotr szybicki
Nov 14 '18 at 11:12

Thanks piotr, I want to check the reponses as they come in and return if the response is not null. So what I have to do is to check in the foreach if the response is not null and ¿then break the foreach? or ¿maybe something like a filter.findFirst() to return when some data is found?

– arturo.galan
Nov 14 '18 at 11:30

I think in that case this is your best option: Long aLong = Flux.merge(listOfMonos).filter((t) -> t != null).toStream().findFirst().get();

– piotr szybicki
Nov 14 '18 at 11:59

add a comment |

The code above should work. Flux.merge(calls) should return the data as soon as response from any of the monos is available and pass it to the forEach method. Maybe i don't understand what is that you are trying to accomplish. I run the simple example: Mono<Long> delay1 = Mono.delay(Duration.ofMillis(1000)); Mono<Long> delay2 = Mono.delay(Duration.ofMillis(2000)); List<Mono<Long>> listOfMonos = new ArrayList<>(); listOfMonos.add(delay1); listOfMonos.add(delay2); Flux.merge(listOfMonos).toStream().forEach(System.out::println);`

– piotr szybicki
Nov 14 '18 at 11:12

Thanks piotr, I want to check the reponses as they come in and return if the response is not null. So what I have to do is to check in the foreach if the response is not null and ¿then break the foreach? or ¿maybe something like a filter.findFirst() to return when some data is found?

– arturo.galan
Nov 14 '18 at 11:30

I think in that case this is your best option: Long aLong = Flux.merge(listOfMonos).filter((t) -> t != null).toStream().findFirst().get();

– piotr szybicki
Nov 14 '18 at 11:59

The code above should work. Flux.merge(calls) should return the data as soon as response from any of the monos is available and pass it to the forEach method. Maybe i don't understand what is that you are trying to accomplish. I run the simple example: Mono<Long> delay1 = Mono.delay(Duration.ofMillis(1000)); Mono<Long> delay2 = Mono.delay(Duration.ofMillis(2000)); List<Mono<Long>> listOfMonos = new ArrayList<>(); listOfMonos.add(delay1); listOfMonos.add(delay2); Flux.merge(listOfMonos).toStream().forEach(System.out::println);`

– piotr szybicki
Nov 14 '18 at 11:12

Thanks piotr, I want to check the reponses as they come in and return if the response is not null. So what I have to do is to check in the foreach if the response is not null and ¿then break the foreach? or ¿maybe something like a filter.findFirst() to return when some data is found?

– arturo.galan
Nov 14 '18 at 11:30

I think in that case this is your best option: Long aLong = Flux.merge(listOfMonos).filter((t) -> t != null).toStream().findFirst().get();

– piotr szybicki
Nov 14 '18 at 11:59

add a comment |

1 Answer
1

active

oldest

votes

You could take only the first element of your flux

 Flux.merge(calls).take(1)

You get new Flux with only first element and it completes after producing it.

answered Nov 14 '18 at 11:39

Alexander Pankin

64126

I think this not meet the original criteria. As the null check on the response is missing. The code you proposed simply return whatever comes back first.

– piotr szybicki
Nov 14 '18 at 12:05

Why don't filter webclient retrieve result to return emty if response is null?

– Alexander Pankin
Nov 14 '18 at 13:07

Yes, you are right. That would work.

– piotr szybicki
Nov 14 '18 at 13:16

add a comment |

Your Answer

StackExchange.ifUsing("editor", function () {
StackExchange.using("externalEditor", function () {
StackExchange.using("snippets", function () {
StackExchange.snippets.init();
});
});
}, "code-snippets");

StackExchange.ready(function() {
var channelOptions = {
tags: "".split(" "),
id: "1"
};
initTagRenderer("".split(" "), "".split(" "), channelOptions);

StackExchange.using("externalEditor", function() {
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled) {
StackExchange.using("snippets", function() {
createEditor();
});
}
else {
createEditor();
}
});

function createEditor() {
StackExchange.prepareEditor({
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader: {
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
},
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
});

}
});

draft saved

draft discarded

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53298508%2fspring-webflux-race-of-parallel-calls-and-cancel-return-when-one-of-the-respons%23new-answer', 'question_page');
}
);

Post as a guest

Name

Required, but never shown

1 Answer
1

active

oldest

votes

1 Answer
1

active

oldest

votes

You could take only the first element of your flux

 Flux.merge(calls).take(1)

You get new Flux with only first element and it completes after producing it.

answered Nov 14 '18 at 11:39

Alexander Pankin

64126

I think this not meet the original criteria. As the null check on the response is missing. The code you proposed simply return whatever comes back first.

– piotr szybicki
Nov 14 '18 at 12:05

Why don't filter webclient retrieve result to return emty if response is null?

– Alexander Pankin
Nov 14 '18 at 13:07

Yes, you are right. That would work.

– piotr szybicki
Nov 14 '18 at 13:16

add a comment |

You could take only the first element of your flux

 Flux.merge(calls).take(1)

You get new Flux with only first element and it completes after producing it.

answered Nov 14 '18 at 11:39

Alexander Pankin

64126

I think this not meet the original criteria. As the null check on the response is missing. The code you proposed simply return whatever comes back first.

– piotr szybicki
Nov 14 '18 at 12:05

Why don't filter webclient retrieve result to return emty if response is null?

– Alexander Pankin
Nov 14 '18 at 13:07

Yes, you are right. That would work.

– piotr szybicki
Nov 14 '18 at 13:16

add a comment |

You could take only the first element of your flux

 Flux.merge(calls).take(1)

You get new Flux with only first element and it completes after producing it.

answered Nov 14 '18 at 11:39

Alexander Pankin

64126

You could take only the first element of your flux

 Flux.merge(calls).take(1)

You get new Flux with only first element and it completes after producing it.

answered Nov 14 '18 at 11:39

Alexander Pankin

64126

answered Nov 14 '18 at 11:39

Alexander Pankin

64126

answered Nov 14 '18 at 11:39

Alexander Pankin

64126

answered Nov 14 '18 at 11:39

Alexander Pankin

64126

I think this not meet the original criteria. As the null check on the response is missing. The code you proposed simply return whatever comes back first.

– piotr szybicki
Nov 14 '18 at 12:05

Why don't filter webclient retrieve result to return emty if response is null?

– Alexander Pankin
Nov 14 '18 at 13:07

Yes, you are right. That would work.

– piotr szybicki
Nov 14 '18 at 13:16

add a comment |

I think this not meet the original criteria. As the null check on the response is missing. The code you proposed simply return whatever comes back first.

– piotr szybicki
Nov 14 '18 at 12:05

Why don't filter webclient retrieve result to return emty if response is null?

– Alexander Pankin
Nov 14 '18 at 13:07

Yes, you are right. That would work.

– piotr szybicki
Nov 14 '18 at 13:16

I think this not meet the original criteria. As the null check on the response is missing. The code you proposed simply return whatever comes back first.

– piotr szybicki
Nov 14 '18 at 12:05

Why don't filter webclient retrieve result to return emty if response is null?

– Alexander Pankin
Nov 14 '18 at 13:07

Yes, you are right. That would work.

– piotr szybicki
Nov 14 '18 at 13:16

add a comment |

draft saved

draft discarded

Thanks for contributing an answer to Stack Overflow!

Please be sure to answer the question. Provide details and share your research!

But avoid …

Asking for help, clarification, or responding to other answers.

Making statements based on opinion; back them up with references or personal experience.

To learn more, see our tips on writing great answers.

draft saved

draft discarded

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

Post as a guest

Name

Required, but never shown

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

Sign up or log in

StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});

Post as a guest

Name

Required, but never shown

Name

Required, but never shown

Name

Required, but never shown

This page is only for reference, If you need detailed information, please check here

搜尋此網誌

Vfrdtyky