Java mit Project Reactor: Warum sieht der Mono-Block() den Fehler nicht?

Question

Gegeben der folgende Code, result.block() ist gleich "xx", isError() == false, aber der Boom-Error-Handler läuft, der Stacktrace wird angezeigt, das Versprechen ist nicht abgeschlossen. Ich würde erwarten, dass das Ergebnis "ko" ist.

Was mache ich falsch? Block() Javadoc sagt

null zurück, wenn onComplete, T wenn OnNext

public class RApp { 

static final Logger LOG = LoggerFactory.getLogger(RApp.class); 

public static void main(String[] args) { 

    MonoProcessor<String> texecute = MonoProcessor.create(); 
    Mono<String> result = texecute.delaySubscription(Duration.ofSeconds(2)) 
      .onErrorReturn("ko") 
      .doOnNext(s -> parse(s) 
      .doOnSuccess(p -> LOG.info("promise completed {}", p)) 
      .doOnTerminate((z, e) -> LOG.info("term value: {} , {}", z, e)) 
      .doOnError(t -> { 
       LOG.error("boom", t); 
      }) 
      .subscribe()); 

    texecute.onNext("xx"); 

    LOG.info("...............;"); 
    String block = result.block(); 
    LOG.info("r={}", block); 
    boolean error = texecute.isError(); 
    LOG.info(error ? "error" : "no error"); 
    texecute.dispose(); 

} 

public static Mono<String> parse(String s) { 
    System.out.println("parse s = " + s); 
    if (s.equals("xx")) { 
     return Mono.error(new Exception("no xx")); 
    } 
    return Mono.just(s); 
} 
} 

Answer 1

mich Answering auf diesen einen: die do * sind Nebenwirkungen Methoden die Sequenz nicht pro https://projectreactor.io/docs/core/release/reference/#error.handling ändern, und die Bestellung von onErrorReturn Angelegenheiten.

Proper Arbeitslösung unten, mit Boni von reactor.core.Exceptions.propagate geprüft Ausnahmen zu wickeln und ein Java-8 Fehlerzähler:

LongAdder failureStat = new LongAdder(); 

    MonoProcessor<String> texecute = MonoProcessor.create(); 
    Mono<String> result = texecute 
      .delaySubscription(Duration.ofSeconds(2)) 
      .map(e -> parse2(e)).doOnError(e -> { 
       failureStat.increment(); 
      }).doOnSuccess(s -> { 
       LOG.info("success {}", s); 
      }) 
      .onErrorReturn("ko") 
      .subscribe(); 

    texecute.onNext("xx"); 

    LOG.info("...............;"); 
    String block = result.block(); 
    LOG.info("r={}", block); 
    System.out.println("failureStat = " + failureStat); 
    texecute.dispose(); 

public static String parse2(String s) { 
    System.out.println("parse s = " + s); 
    if (s.equals("xx")) { 
     try { 
      throw new Exception("no xx"); 
     } catch (Exception e) { 
      throw Exceptions.propagate(e); 
     } 
    } 
    return s; 
}