jeroenr

interface Service {	
	Flux<String> findAll();
	Mono<String> lookup(String s);
}
class Foo {
	private final Service service;
	
	Stream<String> problem() {
		return service.findAll().toStream();
	}

jeroenr

interface Service {	
	Flux<String> findAll();
}
class Foo {
	private final Service service;
	
	Iterable<String> problem() {
		return service.findAll().toIterable();
	}
}

jeroenr

interface Service {	
	Flux<String> findAll();
	Mono<Void> operation(String s);
}
class Foo {
	private final Service service;
	
	Flux<Void> problem() {
		return service.findAll()
			.concatMap(service::operation);

jeroenr

interface Service {	
	Flux<String> findAll();
	Mono<Void> operation(String s);
}
class Foo {
	private final Service service;
	
	Flux<Void> problem() {
		return service.findAll()
			.flatMap(service::operation, 4); // parallelism=4

jeroenr

interface Service {	
	Flux<String> findAll();
	Mono<Void> operation(String s);
}
class Foo {
	private final Service service;
	
	Flux<Void> problem() {
		return service.findAll()
			.flatMap(service::operation);

jeroenr

abstract class UploadService {	
	protected Mono<Void> doUpload(InputStream in);
	Mono<Void> upload(InputStream in) {
		doUpload(in).doFinally(x -> in.close());
	}
}
class Foo {
	private final UploadService service;
	
	Mono<Void> problem(byte[] data) {

jeroenr

interface Service {	
	Flux<Integer> findAll();
}
class Foo {
	private final Service service;
	
	Flux<Integer> problem() {
		AtomicInteger count = new AtomicInteger();
		return service.findAll()
			.doOnNext(count::addAndGet)

jeroenr

interface Service {	
	Mono<String> create(String s);
	Mono<Void> update(String s);
}
class Foo {
	private final Service service;
	
	Mono<Integer> problem() {
		return service.create("foo").flatMap(foo ->
			service.update(foo).thenReturn(foo.length())

jeroenr

interface Service {	
	Mono<String> create(String s);
	Mono<Void> update(String s);
}
class Foo {
	private final Service service;
	
	Mono<Integer> problem() {
		return service.create("foo").map(foo -> {
			service.update(foo).subscribe();

jeroenr

interface Service {	
	Mono<Void> update(String s);
}
class Foo {
	private final Service service;
	
	void problem() {
        	service.update("foo").onErrorResume(e -> {
                	e.printStackTrace();
                	return Mono.empty();