Wie kann ich einen ExtentMap-Handler in Rx.Net implementieren?

Question

Ich suche etwas Ähnliches wie der exhaustMapOperator von rxjs, RX.NETscheint aber keinen solchen Operator zu haben.

Was ich erreichen muss, ist, dass ich für jedes Element des Quelldatenstroms einen asyncHandler starten muss , und bis es fertig ist, möchte ich alle Elemente aus der Quelle löschen. Sobald der Handler fertig ist, nehmen Sie die Aufnahme von Elementen wieder auf.

Was ich nicht möchte, ist, für jedes Element einen asynchronen Handler zu starten - während der Handler ausgeführt wird, möchte ich Quellelemente löschen .

Ich vermute auch, dass ich hier den Defer-Operator geschickt einsetzen muss?

Vielen Dank!

Answer 1

Hier ist eine Implementierung des ExhaustMapOperators. Die beobachtbare Quelle wird auf eine projiziert IObservable<Task<TResult>>, wobei jede nachfolgende Aufgabe entweder die vorherige ist, wenn sie noch ausgeführt wird, oder auf andere Weise eine neue Aufgabe, die dem aktuellen Element zugeordnet ist. Wiederholte Vorkommen derselben Aufgabe werden dann mit dem DistinctUntilChangedBediener entfernt, und schließlich wird das Beobachtbare mit dem ConcatBediener abgeflacht .

public static IObservable<TResult> ExhaustMap<TSource, TResult>(
    this IObservable<TSource> source,
    Func<TSource, Task<TResult>> function)
{
    return source
        .Scan(Task.FromResult<TResult>(default), (previousTask, item) =>
        {
            return !previousTask.IsCompleted ? previousTask : HideIdentity(function(item));
        })
        .DistinctUntilChanged()
        .Concat();

    async Task<TResult> HideIdentity(Task<TResult> task) => await task;
}

Es functionwird nicht garantiert, HideIdentitydass die von der zurückgegebenen Aufgaben eindeutig sind, weshalb die lokale Funktion erforderlich ist , die unterschiedliche Wrapper der Aufgaben zurückgibt.

Anwendungsbeispiel:

Observable
    .Interval(TimeSpan.FromMilliseconds(200))
    .Select(x => (int)x + 1)
    .Take(10)
    .Do(x => Console.WriteLine($"Input: {x}")) .ExhaustMap(async x => { await Task.Delay(x % 3 == 0 ? 500 : 100); return x; }) .Do(x => Console.WriteLine($"Result: {x}"))
    .Wait();

Ausgabe:

Input: 1
Result: 1
Input: 2
Result: 2
Input: 3
Input: 4
Input: 5
Result: 3
Input: 6
Input: 7
Input: 8
Result: 6
Input: 9
Input: 10
Result: 9