Q:

Reactor: function creating Monos to Flux

Basically, I'm making a queue processor in Spring Boot and want to use Reactor for async.  I've made a function needs to loop forever as it's the one that pulls from the queue then marks the item as processed.
here's the blocking version that works Subscribe returns a Mono
while(true) {
    manager.Subscribe().block()
}

I'm not sure how to turn this into a Flux I've looked a interval, generate, create, etc. and I can't get anything to work without calling block()
Here's an example of what I've tried
Flux.generate(() -> manager,
    (state, sink) -> {
        state.Subscribe().block();
        sink.next("done");
        return state;
    }));

Being a newbie to Reactor, I haven't been able to find anything about just loop and processing the Monos synchronously without blocking.
Here's what the Subscribe method does using the 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())));
}

Basically, I'm just polling an SQS queue, deleting the messages then I want to do it again.  This is all just exploratory for me.
Thanks!

A:

You need two things: a way to subscribe in a loop and a way to ensure that the Subscribe() method is effectively called on each iteration (because the Future needs to be recreated).
repeat() is a baked in operator that will resubscribe to its source once the source completes. If the source errors, the repeat cycle stops. The simplest variant continues to do so Long.MAX_VALUE times.
The only problem is that in your case the Mono from Subscribe() must be recreated on each iteration.
To do so, you can wrap the Subscribe() call in a defer: it will re-invoke the method each time a new subscription happens, which includes each repeat attempt:
Flux<Stuff> repeated = Mono
    .defer(manager::Subscribe)
    .repeat();

