RxJava Zwróć wartość subskrybowaną w funkcji

Question

Z jakiegoś powodu, czasami chcę używać RxOperators zamiast normalnej java-metody przekształcania struktur danych, ponieważ jest ona czystsza i czystsza. Na przykład:

Observable.from(listOfStrings) 
.filter(string -> string.getNumber() >5) 
.toList() 

Czy jest jakiś sposób oczekiwania na skutek obserwowalne i powrócić w w funkcji: To byłoby zrobić (ale nie działa):

private String getFilteredString(String string){ 
    Observable.from(listOfStrings) 
     .filter(string -> string.getNumber() >5) 
     .toList() 
     .subscribe(finalStirng -> { 
     return finalString; 
     }) 
} 

Answer 1

Można przekształcić dowolny Observable do synchronicznego jednym z użyciem .toBlocking():

private List<String> getFilteredString(List<String> strings) { 
    return Observable.from(strings) 
     .filter(string -> string.length() > 5) 
     .toList() 
     .toBlocking() 
     .single(); 
} 

UPDATE: Chociaż powyższy kod jest całkowicie poprawny i będzie działać zgodnie z oczekiwaniami, to nie jest tak, jak powinno być używane RxJava. Jeśli jedynym celem jest przekształcenie kolekcji, są lepsze sposoby, aby to zrobić:

• Java 8 ma Stream API
• Guava posiada własny collections API
• Apache ma commons-collections bibliotekę

Answer 2

Istnieje metoda instancji pod numerem rx.Observable o nazwie x() (link), która może służyć do bezpośredniego i płynnego ukrywania obserwowalnej innej wartości. Intencją tego operatora jest zbudowanie repozytorium funkcji konwersji, które mogą przełączać Obserwowalne na inne rodzaje współbieżnych/asynchronicznych struktur danych (takich jak inne rodzaje obserwowalnych obiektów) lub po prostu odwijają zawartość obserwowalnego do zawartych wartości. Napisz funkcję konwersji i przechowuj ją do ponownego użycia.

Metoda podpis:

public <R> R x(Func1<? super OnSubscribe<T>, ? extends R> conversion); 

Wykorzystanie do konwersji potencjalnie asynchroniczny obserwowalne do jednoczesnego listy:

List<Integer> list = Observable.range(1, 10).x(new Func1<OnSubscribe<Integer>, List<Integer>>() { 
    @Override 
    public List<Integer> call(OnSubscribe<Integer> onSubscribe) { 
     List<Integer> allElements = new ArrayList<Integer>(); 
     CountDownLatch latch = new CountDownLatch(1); 
     AtomicReference<Throwable> failure = new AtomicReference<Throwable>(); 
     onSubscribe.call(new Subscriber<Integer>(){ 
      @Override 
      public void onCompleted() { 
       latch.countDown(); 
      } 

      @Override 
      public void onError(Throwable e) { 
       failure.set(e); 
      } 

      @Override 
      public void onNext(Integer t) { 
       allElements.add(t); 
      }}); 
     while (true) { 
      try { 
       latch.await(); 
       break; 
      } catch (InterruptedException e1) { 
       // continue waiting 
      } 
     } 
     Throwable e = failure.get(); 
     if (e != null) 
      throw new RuntimeException(e); 
     return allElements; 

    }}); 
System.out.println(list); 

wyjściowego

[1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 

Należy pamiętać, że operator jest obecnie @Experimental i jest bardzo prawdopodobne, że zostanie zmieniona (aby "przedłużyć"), jednak e Funkcja xperimental w RxJava może być changed in any release.