How to include if-else statements into reactive flow

Well, I don't think there is a good understanding in general of the reactive library. What I mean is that generally people approach like Java 8 streams in that they are trying to do functional programming. Of course the reactive library is based in functional programming, but I think the purpose is to be asynchronous around blocking I/O. Consider the (current) front page of the WebFlux project.

What is reactive processing? Reactive processing is a paradigm that enables developers build non-blocking, asynchronous applications that can handle back-pressure (flow control).

So, this is a longwinded way of saying I think it is better to focus on where the I/O is happening rather than creating functional code. If you need if statements, then you need if statements. Instead of trying to figure out how to do if statements with functional programming try to figure out where the I/O is taking place and handle it in an asynchronous fashion. One "trick" I like to use is Mono::zip or Flux::zip. These functions combine many I/O calls into one publisher to be returned to the client. So, consider this example code.

Let's make some reactive r2dbc functions:

Mono<Client> save(Client client) {
    client.id = 1L;
    System.out.println("Save client: " + client.id);
    return Mono.just(client);
}
Mono<Phone> save(Phone phone) {
    System.out.println("Save phone: " + phone.clientId);
    return Mono.just(phone);
}
Mono<Referral> save(Referral referral) {
    System.out.println("Save referral: " + referral.clientId);
    return Mono.just(referral);
}
Mono<Contact> save(Contact contact) {
    System.out.println("Save contact: " + contact.clientId);
    return Mono.just(contact);
}

We need some example classes to use:

class DTO {
    Client client;
    List<Phone> phones;
    Optional<Contact> contact;
    Optional<Referral> referral;
}

class Client {
    Long id;
}

class Contact {
    Long clientId;
}

class Referral {
    Long clientId;
}

class Phone {
    Long clientId;
}

Our input is probably a Mono<DTO> since that is what the Request should supply, so our Service layer needs to start with that and return a Mono<Long> of the client id.

Mono<Long> doWork(Mono<DTO> monoDto) {
    return monoDto.flatMap(dto->{
        return save(dto.client).flatMap(client->{
            List<Mono<?>> publishers = new ArrayList<>();
            dto.phones.forEach(phone->{
                phone.clientId = client.id;
                publishers.add(save(phone));    
            });
            if ( dto.contact.isPresent()) {
                Contact c = dto.contact.get();
                c.clientId = client.id;
                publishers.add(save(c));
            }
            if ( dto.referral.isPresent()) {
                Referral r = dto.referral.get();
                r.clientId = client.id;
                publishers.add(save(r));
            }
            if ( publishers.size() > 0 )
                return Mono.zip(publishers, obs->client.id);
            else
                return Mono.just(client.id);
        });
    });
}

I ran this with the following example code:

@Override
public void run(ApplicationArguments args) throws Exception {
    saveClient(new Client(), null, null, null).subscribe(System.out::println);
    saveClient(new Client(), new Phone(), null, null).subscribe(System.out::println);
    saveClient(new Client(), new Phone(), new Contact(), null).subscribe(System.out::println);
    saveClient(new Client(), new Phone(), new Contact(), new Referral()).subscribe(System.out::println);
}


private Mono<Long> saveClient(Client client, Phone phone, Contact contact,
        Referral referral) {
    // TODO Auto-generated method stub
    DTO dto = new DTO();
    dto.client = client;
    dto.phones = new ArrayList<>();
    if ( phone != null ) dto.phones.add(phone);     
    dto.contact = Optional.ofNullable(contact);
    dto.referral = Optional.ofNullable(referral);
    return doWork(Mono.just(dto));
}

So, this uses the Mono.zip trick. The saved client is flatmapped so that is done first. Then a list of monos is created for all subsequent saves that need to be done. These monos are all executed asynchronously by the Mono.zip function. The "combiner" function does nothing with the results, it just returns the clientId which is what is wanted for the client. The Mono.zip combines all the Monos into a single Mono to return to the client. In a sense this is just taking procedural code and wrapping it in the reactive library rather than getting overly focused on functional programming. This is easy to read and modify if the business "process" changes going forward.

This is a starting point if you like it. I didn't use Repository::saveAll so that could be an improvement.

It's important to be sure all your Flux and Mono publishers are chained together. In your final example you seemed to be dropping them. Simply creating them is not enough, they all have to be returned to the client somehow. Also, your code has a subscribe call which is a no-no. Only the client should have subscribe. I think you should have used a map there.

EDIT: Fixed a bug. Check your code carefully.

EDIT II: I noticed that I used an "Optional" as a parameter in the class. This is an anti-pattern. The correct way is to use the type as the parameter and wrap it in an "Optional.of" in a special getter.

a plain if-statement can be done for instance in a flatMap and then acted upon.

public Mono<String> foobar() {
    return Mono.just("foo").flatMap(value -> {
        if(value != null)
            return Mono.just("Has value");
        else
            return Mono.empty();
    }
}

foobar()
    .switchIfEmpty(Mono.just("Is empty"))
    .subscribe(output -> System.out.println(output);