Reactor: функция создания Monos to Flux

По сути, я делаю процессор очереди в Spring Boot и хочу использовать Reactor для async. Я сделал функцию, которая должна повторяться вечно, поскольку она извлекает из очереди, а затем отмечает элемент как обработанный.

вот версия блокировки, которая работает. Подписка возвращает Mono

while(true) {
    manager.Subscribe().block()
}

Я не уверен, как превратить это в поток. Я просмотрел интервал, сгенерировал, создал и т.д., и я не могу заставить что-либо работать без вызова block ()

Вот пример того, что я пробовал

Flux.generate(() -> manager,
    (state, sink) -> {
        state.Subscribe().block();
        sink.next("done");
        return state;
    }));

Поскольку я новичок в Reactor, я не смог найти ничего о простом цикле и синхронной обработке Monos без блокировки.

Вот что делает метод Subscribe с помощью AWS Java SDK v2:

public Mono Subscribe() {
    return Mono.fromFuture(_client.receiveMessage(ReceiveMessageRequest.builder()
            .waitTimeSeconds(10)
            .queueUrl(_queueUrl)
            .build()))
            .filter(x -> x.messages() != null)
            .flatMap(x -> Mono.when(x.messages()
                    .stream()
                    .map(y -> {
                        _log.warn(y.body());
                        return Mono.fromFuture(_client.deleteMessage(DeleteMessageRequest.builder()
                                .queueUrl(_queueUrl)
                                .receiptHandle(y.receiptHandle())
                                .build()));
                    }).collect(Collectors.toList())));
}

По сути, я просто опрашиваю очередь SQS, удаляю сообщения и хочу сделать это снова. Для меня это всего лишь проба.

Спасибо!