RxJava führt nur ein Observable auf einmal aus

Question

Wir haben ein Fenster mit Suchfeld. Jedes Mal, wenn der Benutzer etwas eingibt, wird eine Suche durchgeführt.

1. Such-Ereignis wird in Datenstrom übersetzt.

2. Bei jeder neuen Suche müssen wir den asynchronen Netzwerkbetrieb starten und den vorherigen Vorgang schließen. Wie archiviert man diesen Effekt?

Edit1: Hier ist, was ich versucht habe. Es führt alle Observablen aus! Nicht der einzige letzte, wo ist ein Fehler?

PublishSubject<Integer> subject = PublishSubject.create(); 

subject.switchMap(integer -> Observable.fromCallable(() -> { 
    try { 
     Thread.sleep(500); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } 
    return "-" + integer + "-"; 
})).subscribe(s -> System.out.print(s)); 
for (int i = 0; i < 30; i++) subject.onNext(i); 

Answer 1

Verwenden PublishSubject mit switchMap:

PublishSubject<String> subject = PublishSubject.create(); 
subject 
    .switchMap(
    x -> networkOperationObservable(x) 
      .subscribeOn(Schedulers.io)) 
    .subscribe(subscriber); 

Um das Beste aus diesem networkOperationObservable heraus zu bekommen, würde vernünftig auf einen unsubscribe Anruf reagieren (wie in der Nähe der Socket oder was auch immer). Observable.using ist das Werkzeug der Wahl dafür in der Regel.

Answer 2

PublishSubject ist genau das, was Sie brauchen. Es ist ein Observable Sie können jederzeit Ereignisse hinzufügen.

es erstellen:

private Subject<String, String> filteringSubject = PublishSubject.create(); 

Dann Observable.throttleWithTimeout() Operator:

Subscriber<String> filteringSubscriber = new DefaultSubscriber<>(); 
filteringSubject 
    .throttleWithTimeout(250, TimeUnit.MILLISECONDS) 
    .doOnNext(new Action1<String>() { 
     @Override 
     public void call(String text) { 
      //make a network call 
     } 
    }) 
    .subscribe(filteringSubscriber); 

Verbrauch:

filteringSubject.onNext(text);