RxJava scalania debounced i nie debounced obserwable

Question

Mam dwa obserwable:

Obserwowalne O (otwarte): Plik z niektórych treści otwarty w TextView

Obserwowalne E (edit): zawartość pliku edytowane w TextView

Chcę wydobyć E obserwowalne i połączyć go z O obserwowalnym.

obs = Observable.merge(E.debounce(2000, TimeUnit.MILLISECONDS) , O) 
       .subscribe(content->System.out.println("new content: " + content)) 

Problem polega na tym, że jeśli E emituje zdarzenie E1 i zaraz po tym O emituje zdarzenie O1, mamy wyjścia:

new content: O1 
new content: E1 // this output is rebundant (cuz we already have newer content O1) 

To jest schemat tego, co się dzieje:

Jak pozbyć się tego przesadnego starego wydarzenia z wyraźnego obserwowalnego?

Answer 1

Można spróbować

Observable.merge(O, O.switchMap(o -> E.debounce())) 
      .subscribe() 

switchMap zachowuje się podobnie jak flatMap, z wyjątkiem, że gdy nowy element jest emitowanego przez źródło obserwowalne, będzie zrezygnować i zatrzymać mirroring obserwowalnym który został wygenerowany z wcześniej emitowanego elementu i rozpoczyna się tylko lustrzane odbicie bieżącego.

Answer 2

Widzę dwie główne opcje. Jedną z nich jest użycie znaczników czasowych, które są dość łatwe, ale istnieją teoretyczne warunki wyścigu (ale prawdopodobnie jest to mało prawdopodobne), a drugą opcją jest użycie unikalnego identyfikatora powiązanego z każdym otwarciem pliku, a zdarzenia emitowane z edycji do tekstu tego otwartego pliku są wraz z identyfikatorem otwierania pliku.

Korzystanie timestamps:

obs = Observable.defer(() -> { 
    AtomicBoolean first = new AtomicBoolean(true); 
    e.timestamp() 
    .debounce(2000, TimeUnit.MILLISECONDS)) 
    .mergeWith(o.timestamp()) 
    .buffer(2,1) 
    .flatMap(list -> { 
    Observable<Object> start; 
    if (first.compareAndSet(true, false)) 
     start = Observable.just(list.get(0).getValue()); 
    else 
     start = Observable.empty(); 
    if (list.size() == 1) 
     return start; 
    else { 
     Timestamped<Object> a = list.get(0); 
     Timestamped<Object> b = list.get(1); 
     if (a.getTimestampMillis() <= b.getTimestampMillis()) 
     return start.concatWith(Observable.just(b.getValue())); 
     else 
     return start; 
    } 
    }) 
}); 

Podejrzewam, że wersja o czasie, będą wystarczające.