Làm thế nào để chờ cho (có thể) của Observable onComplete() trong flatMap()

Question

Tôi có một số Observable người đọc dữ liệu từ cơ sở dữ liệu. Nếu dữ liệu là null Tôi cần tải dữ liệu từ mạng. Vì vậy, tôi làm flatMap vào ngày Observable đầu tiên, kiểm tra kết quả của hoạt động cơ sở dữ liệu và nếu đó là null Tôi bắt đầu một Observable khác để tìm nạp dữ liệu từ mạng.

Lưu ý:Observable s có khác nhau Subscriber s vì tôi có xử lý sau khác nhau tùy thuộc vào nơi dữ liệu đến từ (một logic như vậy).

Observable.just(readDataFromDb()).flatMap(new Func1<SomeData, Observable<String>>() { 
     @Override public Observable<SomeData> call(SomeData s) { 
      if (s == null) { 
      getReadFromNetworkObservable().subscribe(new AnotherSubscriber()); // this one might not complete 
      return Observable.empty(); // I think I need to send this one only after readFromNetwork() completed 
      } else { 
      return Observable.just(s); 
      } 
     } 
     }).subscribe(new SomeSubscirber()); 

Cho tôi gửi Observable.empty() để loại trừ xử lý dữ liệu cho SomeSubscriber, tôi có một linh tính của tôi thứ hai Observable có thể không phải lúc nào được hoàn thành bởi vì nó có thể chỉ đơn giản là thu gom rác thải. Tôi đoán tôi đã nhìn thấy nó trong các bài kiểm tra của tôi.

Tại thời điểm này, tôi nghĩ rằng tôi chỉ cần đợi cho đến khi Observable người đọc từ mạng hoàn tất và sau đó gửi Observable.empty(). Vậy tôi có thể thực hiện đồng bộ không? Nhưng tôi vẫn có cảm giác mình làm sai.

Answer 1

Bạn có thể thực hiện bất kỳ quan sát được như chặn với .toBlocking shortcut (Xem đầy đủ thông tin https://github.com/ReactiveX/RxJava/wiki/Blocking-Observable-Operators)

Data d = getReadFromNetworkObservable() 
      .toBlocking() 
      .first() // or single() or singleOrDefault() 

// manipulate with data here 

Kết hợp bộ nhớ cache với dữ liệu mạng được mô tả ở đây: http://blog.danlew.net/2015/06/22/loading-data-from-multiple-sources-with-rxjava/

Và ở đây: RxJava and Cached Data