RxJava Return Giá trị đã đăng ký trong một hàm

Question

Vì một lý do nào đó, đôi khi tôi muốn sử dụng RxOperators thay vì java-way bình thường để chuyển đổi cấu trúc dữ liệu, vì nó sạch hơn và sạch hơn. Ví dụ:

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

Có cách nào chờ đợi kết quả của các quan sát và trở lại trong trong một chức năng: Điều này sẽ làm (nhưng nó không hoạt động):

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

Answer 1

Bạn có thể chuyển đổi bất kỳ Observable sang một đồng bộ sử dụng .toBlocking():

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

UPDATE: Mặc dù mã ở trên là hoàn toàn hợp lệ và sẽ hoạt động như mong đợi, đây không phải là cách RxJava được cho là được sử dụng. Nếu mục tiêu duy nhất của bạn là để biến đổi một bộ sưu tập, có những cách tốt hơn để làm điều đó:

• Java 8 có Stream API
• Guava có riêng của mình collections API
• Apache có một thư viện commons-collections

Answer 2

Có một phương pháp thể hiện trên rx.Observable được gọi là x() (link) có thể được sử dụng để bí mật một giá trị quan sát trực tiếp và trôi chảy. Mục đích của toán tử này là xây dựng một kho lưu trữ các hàm chuyển đổi có thể chuyển một Observable thành các kiểu dữ liệu đồng bộ/không đồng bộ khác (chẳng hạn như các loại quan sát khác) hoặc chỉ đơn giản là unwrap nội dung của quan sát vào các giá trị chứa. Viết chức năng chuyển đổi một lần và lưu trữ nó đi để sử dụng lặp đi lặp lại.

Phương pháp chữ ký:

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

Cách sử dụng để chuyển đổi một tiềm năng không đồng bộ quan sát được vào một danh sách đồng thời:

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); 

đầu ra

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

Xin lưu ý rằng toán tử này hiện đang @Experimental và rất có khả năng được đổi tên (để "mở rộng"), tuy nhiên e chức năng xperimental trong RxJava có thể là changed in any release.