Wie zwei sortierte Observable in einem einzigen sortierten Observable fusionieren?

Question

Gegeben:

Integer[] arr1 = {1, 5, 9, 17}; 
Integer[] arr2 = {1, 2, 3, 6, 7, 12, 15}; 
Observable<Integer> o1 = Observable.from(arr1); 
Observable<Integer> o2 = Observable.from(arr2); 

Wie ein beobachtbares zu erhalten, die 1, 1, 2, 3, 5, 6, 7, 9, 12, 15, 17 enthält?

Answer 1

Bearbeiten: Bitte beachten Sie den Kommentar von the_joric, wenn Sie dies verwenden werden. Es ist ein Grenzfall, der nicht behandelt wird, sehe ich nicht einen schnellen Weg, um es zu beheben, und so habe ich keine Zeit, es jetzt zu beheben.

Hier ist eine Lösung in C#, da Sie die system.reactive Tag haben.

static IObservable<int> MergeSorted(IObservable<int> a, IObservable<int> b) 
{ 
    var source = Observable.Merge(
     a.Select(x => Tuple.Create('a', x)), 
     b.Select(y => Tuple.Create('b', y))); 
    return source.Publish(o => 
    { 
     var published_a = o.Where(t => t.Item1 == 'a').Select(t => t.Item2); 
     var published_b = o.Where(t => t.Item1 == 'b').Select(t => t.Item2); 
     return Observable.Merge(
      published_a.Delay(x => published_b.FirstOrDefaultAsync(y => x <= y)), 
      published_b.Delay(y => published_a.FirstOrDefaultAsync(x => y <= x))); 
    }); 
} 

Die Idee ist wie folgt zusammengefasst.

• Wenn a den Wert emittiert x wir es verzögern, bis b Wert y so dass x <= y emittiert.

• Wenn b den Wert emittiert y wir es verzögern, bis a Wert x so dass y <= x emittiert.

Wenn Sie nur heiße Observable hatten, könnten Sie Folgendes tun. Aber das Folgende würde nicht funktionieren, wenn es kalte Observable in der Mischung gäbe. Ich würde empfehlen, immer die Version zu verwenden, die sowohl für heiße als auch kalte Observables funktioniert.

static IObservable<int> MergeSortedHot(IObservable<int> a, IObservable<int> b) 
{ 
    return Observable.Merge(
     a.Delay(x => b.FirstOrDefaultAsync(y => x <= y)), 
     b.Delay(y => a.FirstOrDefaultAsync(x => y <= x))); 
} 

Answer 2

Sie können, sortieren verschmelzen und die Sequenzen abflachen, aber es wird einen erheblichen Aufwand haben:

o1.mergeWith(o2).toSortedList().flatMapIterable(v -> v).subscribe(...) 

oder

o1.concatWith(o2).toSortedList().flatMapIterable(v -> v).subscribe(...) 

Andernfalls müssen Sie eine ziemlich komplizierte Betreiber schreiben .

bearbeiten 2015.04.06:

Here is ein Operator, der diese nicht effizienter sortiert-fusionieren.

Answer 3

Der vor einiger Zeit auf den RxJava mailing list diskutiert wurde, werden Sie einige Links zu möglichen Lösungen in diesem Thread finden.

Answer 4

suchte ich auch für eine Art Lösung verschmelzen, die sie nicht finden einen Gegendruck auf und konnte unterstützt. Also entschied ich mich, es basierend auf dem vorhandenen Zip-Operator auf eigene Faust zu implementieren.

Ähnlich zip, sammelt die sortierte merge Operator ein Element aus jeder Quelle beobachtbaren ersten, aber dann stellt sie in eine Prioritätswarteschlange, aus der sie nacheinander entsprechend ihrer natürlichen Reihenfolge oder den angegebenen Komparator emittiert.

Sie können es von GitHub als bereit greifen Bibliothek verwenden oder einfach kopieren/fügen Sie den Code:

https://github.com/ybayk/rxjava-recipes

Siehe Unit-Tests für den Einsatz.

Answer 5

Was ist nur zusammenführen und sortieren?

@Test 
public void testMergeChains() { 
    Observable.merge(Observable.from(Arrays.asList(1, 2, 13, 11, 5)), Observable.from(Arrays.asList(10, 4, 12, 3, 14, 15))) 
       .collect(ArrayList<Integer>::new, ArrayList::add) 
      .doOnNext(Collections::sort) 
      .subscribe(System.out::println); 

} 

können Sie weitere Beispiele hier sehen

https://github.com/politrons/reactive