Как построить реактивную архитектуру, которая позволит избежать вложенных блоков Flux (Flux <Flux <T>>)? - PullRequest
0 голосов
/ 30 мая 2019

Я пытаюсь создать приложение A (например, адаптер), которое будет:

1) получать запросы POST с некоторым ключом (формат JSON)

2) оно должно изменить этот ключкаким-то образом и создайте POST-запрос к другой системе B.

3) Приложение A должно проанализировать ответ из приложения B и изменить этот ответ.

4) После этого мое приложение A должно ответить на исходныйЗапрос POST.

@RestController
@RequestMapping("/A")
public class Controller {
    @ResponseStatus(HttpStatus.OK)
    @PostMapping(value = "B", consumes = APPLICATION_JSON_VALUE)
    // to return nested Flux is a bad idea here
    private Flux<Flux<Map<String, ResultClass>>> testUpdAcc(@RequestBody Flux<Map<String, SomeClass>> keys) {
        return someMethod(keys);
    }

    // the problem comes here when I will get Flux<Flux<T>> in the return
    public Flux<Flux<Map<String, ResultClass>>> someMethod(Flux<Map<String, SomeClass>> keysFlux) {
        return keysFlux.map(keysMap -> {
                                // do something with keys and create URL
                                // also will batch keys here
                                <...>

                                // for each batch of keys:
                                WebClient.create(hostAndPort)
                                .method(HttpMethod.POST)
                                .uri(url)
                                .body(BodyInserters.fromObject(body))
                                .header(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded")
                                .accept(MediaType.APPLICATION_JSON)
                                .retrieve()
                                .bodyToMono(schema) // response will be parsed into some schema here
                                .retryWhen (// will make a retry mechanism here)

                                // ===== will join all Mono batches into single Flux
                                Flux.concat(...);
                                }
                             );
    }

}

Конечно, это можно исправить, не читая keysFlux как Flux и считывая его как Map.Но это должно сделать все менее реактивным, нет?:)

    @ResponseStatus(HttpStatus.OK)
    @PostMapping(value = "B", consumes = APPLICATION_JSON_VALUE)
    // to return nested Flux is a bad idea here
    private Flux<Map<String, ResultClass>> testUpdAcc(@RequestBody Map<String, SomeClass> keys) {
        return someMethod(keys);
    }

Также я пытался использовать block () / blockFirst () в последний момент перед возвратом запроса, но я получил ошибку:

block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor...

СпасибоВы за ваши идеи!

Ответы [ 2 ]

0 голосов
/ 30 мая 2019

Попробуйте сжать весь поток как этот

Flux.zip(flux1,flux2)

Это создаст Tuple2, так что вы можете сделать flatMap

Спасибо,
Вималеш

0 голосов
/ 30 мая 2019

Забудьте о моем вопросе - мы можем легко использовать «flatMap» вместо «map». Это решит проблему с Flux внутри Flux.

...