RxJava scalaj obserwowalne, ale przyjmuj wynik tylko z pierwszego - rx-java

Chcę osiągnąć coś takiego: Mam żądanie, które produkuje obserwowalne, a także mam rodzaj pamięci podręcznej, która może być przedmiotem zachowania. W moim kodzie chcę scalić te dwa obserwowalne w taki sposób, że jeśli podmiot ma jakieś dane, to chcę je wziąć i nie zaczynać zauważać, czy można osiągnąć coś takiego?

W tej chwili robię coś takiego:

val resp = run { run(Task.GetMessages()) }

return InboxModel(news = Observable.merge(IoImplementation.cachedUserMessages(), resp).distinctUntilChanged())

ale nie sądzę, że to jest dobre

Odpowiedzi:

Wygląda na to, że chcesz spróbować znaleźć pracę w pamięci podręcznej i tylko wtedy, gdy jest ona pusta, spróbuj wykonać prawdziwą pracę. observable.switchIfEmpty(Observable) operator powinien zrobić to, czego potrzebujesz. Jeśli obserwowalna pamięć podręczna jest pusta, to zasubskrybuje rzeczywistą pracę możliwą do zaobserwowania.

public void handleRequests(Observable<Request> requests) {
requests.flatMap(Request r -> {
String key = r.getSomeCacheKey();
return getFromCache(key)
.switchIfEmpty(doRealWork(r)
.doOnNext(saveToCache(key))
);
})
.subscribe((String response) -> System.out::println);
}

// hashmap based cache may not be what you want but i"m
// including it for completeness of example
Map<String, Observable<String>> cache = //...

Observable<String> getFromCache(String key) {
return cache.get(key);
}

Action1<String> saveToCache(String key) {
return (String response) -> cache.put(key, response);
}

Observable<String> doRealWork(Request someRequest) {
// imagine real work here
return Observable.just("some response");
}