Реактивному API необходимо вызывать параллельные нижестоящие API и ожидать ответов. ⇐ JAVA

[Anonymous]

У меня есть несколько устройств (на данный момент восемь, но в будущем их может быть больше) с простой блокировкой HTTP REST API. Один делает вызов GET, чтобы передать значение, API отвечает успешно или неудачно, но всегда 200. Есть еще один GET, чтобы найти текущий параметр. Итак, типичная последовательность будет такой:

Код: Выделить всё

http://\/devicename/set/10

Код: Выделить всё

http://\/devicename/get

У меня есть реактивный API Spring Boot для параллельного вызова этих нижестоящих API, который в настоящее время использует Mono.zip(). У меня есть только код set(). Когда я вызываю реактивный API, он выполняет все вызовы set() и отправляет ответ вызывающему реактивному API, показывая результаты.
Проблема, с которой я столкнулся: мне нужно программно дождаться результатов всех вызовов set(), выполнить вызовы get(), ожидая этих ответов, а затем, при необходимости, выполнить больше раундов set(). -> get(), пока все значения не будут соответствовать требованиям или пока не будут исчерпаны повторные попытки.
Я пробовал несколько вариантов:

• Код: Выделить всё

  block()

  — ошибки во время выполнения с помощью функции block()/blockFirst()/blockLast() блокируются, что не поддерживается в Thread Reactor-http-nio-3

• Код: Выделить всё

  CountDownLatch

  внутри subscribe() — никогда не завершается и не уменьшает защелку

• Код: Выделить всё

  subscribe()

  (несколько вариантов) – по-прежнему ожидает не всех ответов.

Мне кажется, что я упускаю что-то тривиальное. Очевидно, что где-то в реактивном API происходит ожидание, поскольку оно всегда возвращает вызывающему объекту все завершенные ответы set(). Часто для завершения всего этого требуется от 30 до 40 секунд, что вполне ожидаемо и нормально.
Мне нужно иметь возможность сделать это в пользовательском коде и инициировать другие вызовы как упомянул. Я обращаюсь к реактивному API с помощью браузера, и, честно говоря, даже если время вызова браузера истекает, это нормально, пока работа выполняется. Мне необходимо во всех случаях сохранять параллельные вызовы нисходящего потока, поскольку каждое устройство может выполнять работу одновременно.

Подробнее здесь: https://stackoverflow.com/questions/791 ... -responses