Flux.map()

Question

निष्पादित करते समय त्रुटि को कैसे प्रबंधित करें मैं फ़्लक्स के अंदर तत्वों को मैप करते समय त्रुटियों को कैसे संभालने का प्रयास करने की कोशिश कर रहा हूं।

उदाहरण के लिए, मैं अपने व्यवसाय POJOs में से एक में एक सीएसवी स्ट्रिंग को पार्स:

myflux.map(stock -> converter.convertHistoricalCSVToStockQuotation(stock)); 

इस लाइनों की कुछ त्रुटियाँ शामिल हो सकता है, तो क्या मैं लॉग में मिलता है:

reactor.core.publisher.FluxLog: onNext([SOME_BOGUS_QUOTE]@38.09 (Fri Apr 08 00:00:00 CEST 2016) H(38.419998)/L(37.849998)/O(37.970001)) 
reactor.core.publisher.FluxLog: onNext([SOME_BOGUS_QUOTE]@38.130001 (Thu Apr 07 00:00:00 CEST 2016) H(38.189999)/L(37.610001)/O(37.799999)) 
reactor.core.publisher.FluxLog: onError(java.lang.IllegalArgumentException: Invalid CSV stock quotation: SOME_BOGUS_QUOTE,trololo) 
reactor.core.publisher.FluxLog: java.lang.IllegalArgumentException: Invalid CSV stock quotation: SOME_BOGUS_QUOTE,trololo 

मैं तरीकों से निपटने के कुछ त्रुटि एपीआई में पढ़ते हैं, लेकिन एक "त्रुटि मान" लौटने या एक fallback फ्लक्स का उपयोग कर, इस तरह के सबसे अधिक संदर्भित:

Flux.onErrorResumeWith(myflux, x -> Mono.fromCallable(() -> ... do stuff); 

हालांकि, मेरे myflux के साथ इसका उपयोग करने का अर्थ है कि पूरे प्रवाह को फिर से संसाधित किया जाता है।

तो, क्या विशेष तत्वों को संसाधित करते समय त्रुटियों को संभालने का कोई तरीका है (मैं उन्हें अनदेखा/लॉग इन कर रहा हूं) और बाकी प्रवाह को संसाधित करना जारी रखता हूं?

@akarnokd वैकल्पिक हल

public Flux<StockQuotation> getQuotes(List<String> tickers) 
{ 
    Flux<StockQuotation> processingFlux = Flux.fromIterable(tickers) 
    // Get each set of quotes in a separate thread 
    .flatMap(s -> Mono.fromCallable(() -> feeder.getCSVQuotes(s))) 
    // Convert each list of raw quotes string in a new Flux<String> 
    .flatMap(list -> Flux.fromIterable(list)) 
    // Convert the string to POJOs 
    .flatMap(x -> { 
      try { 
       return Flux.just(converter.convertHistoricalCSVToStockQuotation(x));  
      } 
      catch (IllegalArgumentException ex){ 
       System.out.println("Error decoding stock quotation: " + x); 
       return Flux.empty(); 
      } 
    }); 

    return processingFlux; 
} 

यह एक आकर्षण के रूप में काम करता है, फिर भी, जैसा कि आप देख सकते हैं कोड से पहले की तुलना में कम सुरुचिपूर्ण है के साथ अद्यतन। फ्लक्स एपीआई के पास यह कोड करने के लिए कोई तरीका नहीं है?

retry(...) 
retryWhen(...) 
onErrorResumeWith(...) 
onErrorReturn(...) 

Answer 1

आप flatMap जरूरत के बजाय जो चलो अगर संसाधन विफल आप एक खाली अनुक्रम लौट:

myflux.flatMap(v -> { 
    try { 
     return Flux.just(converter.convertHistoricalCSVToStockQuotation(stock)); 
    } catch (IllegalArgumentException ex) { 
     return Flux.empty(); 
    } 
});