Wie warte ich auf einen Wert oder bis eine feste Zeit abgelaufen ist mit Reactive Extensions (Rx.Net)

Question

Ich möchte einen Thread warten (blockieren) bis entweder eine Zeit verstrichen ist oder ein anderer Stream einen Wert pumpt, dachte ich folgendes könnte dies zu erreichen, aber es löst eine Ausnahme, weil der erste Strom leer ist,

// class level subject manipulated by another thread... 
_updates = new Subject<Unit>(); 
... 
// wait for up to 5 seconds before carrying on...  
var result = Observable.Timer(DateTime.Now.AddSeconds(5)) 
    .TakeUntil(_updates) 
    .Wait(); 

Wie kann ich die Fähigkeit erreichen bis zu 5 Sekunden oder bis der andere Strom pumpt Wert zu blockieren?

Answer 1

können Sie Observable.Timeout wie folgt verwenden:

var result = _updates.Take(1).Timeout(DateTime.Now.AddSeconds(5)).Wait(); 

I Take(1) verwenden, da Timeout-Sequenz, nicht nur nächsten Wert produziert abzuschließen erwartet. Bei Zeitüberschreitung wird System.TimeoutException ausgelöst.

Wenn Sie keine Ausnahme wollen - Sie können Catch verwenden, anstatt einen Wert zu bieten:

var result = _updates.Take(1).Timeout(DateTime.Now.AddSeconds(5)) 
    .Catch(Observable.Return(default(Unit))).Wait(); 
// should catch specific exception, not all 

Wenn Ihr Unit in der Tat, dass rx Einheit von @Shlomo erwähnt ist - Sie können es wie folgt ändern:

var result = _updates.Select(c => (Unit?) c).Take(1) 
    .Timeout(DateTime.Now.AddSeconds(5)).Catch(Observable.Return((Unit?) null)).Wait(); 

oder fangen Sie einfach diese Ausnahme wie üblich.

Answer 2

Können Sie zwei Aufgaben erstellen und warten dann? Das wäre wahrscheinlich der "einfachste" Weg, um das zu erreichen.

Bearbeiten: Titel geändert, um zu spezifizieren Reagieren, Antwort würde wahrscheinlich nicht mehr anwendbar sein.

Answer 3

Hier ist eine andere Alternative, wenn Sie mit Ausnahmen beschäftigen möchten nicht:

var _updates = new Subject<Unit>(); 

var result = Observable.Merge(
    _updates.Materialize(), 
    Observable.Empty<Unit>() 
     .Delay(TimeSpan.FromSeconds(5)) 
     .Materialize() 
    ) 
    .Take(1) 
    .Wait(); 

switch (result.Kind) 
{ 
    case NotificationKind.OnCompleted: 
     //time's up, or someone called _updates.OnCompleted(). 
     break; 
    case NotificationKind.OnNext: 
     var message = result.Value; 
     //Message received. Handle message 
     break; 
    case NotificationKind.OnError: 
     var exception = result.Exception; 
     //Exception thrown. Handle exception 
     break; 
}