以下可觀察序列將每個元素添加到 ReplaySubject 中，以便我以后可以訪問任何元素，甚至等待 ReplaySubject 的完成。它在達到一個時間跨度后完成收割主體。ReplaySubject<string> rfidPlayer = new ReplaySubject<string>();characteristic.WhenNotificationReceived() .TakeUntil(Observable.Timer(TimeSpan.FromSeconds(1))) .Subscribe( onNext: result => { string nextTag = BitConverter.ToString(result.Data); nextTag = nextTag.Replace("-", ""); rfidPlayer.OnNext(nextTag); }, onCompleted: () => { rfidPlayer.OnCompleted(); });我希望序列一直運行到自上次“OnNext”調(diào)用以來的給定時間，然后完成。這在各種藍牙通信場景中非常有用，其中藍牙設(shè)備將向我提供一系列數(shù)據(jù)，然后停止而沒有任何類型的完成消息或事件。在這些情況下，我需要啟發(fā)式確定序列何時完成，然后自己完成。因此，如果自上次藍牙通知以來“太長”，我想完成ReplaySubject。我可以通過創(chuàng)建一個計時器來做到這一點，在收到每個元素時重置它，然后在計時器達到“太長”時完成ReplaySubject，但是我聽說創(chuàng)建一個對象并從可觀察的訂閱中操作它不是線程安全的。關(guān)于如何在“太長”的時間間隔后完成序列的任何建議？這是一個與我所聽到的不是線程安全的版本，但應(yīng)該按預(yù)期工作：bool reading = true;System.Timers.Timer timer = new System.Timers.Timer(1000);timer.Elapsed += (sender, e) =>{ reading = false;};ReplaySubject<string> rfidPlayer = new ReplaySubject<string>();characteristic.WhenNotificationReceived() .TakeWhile(x => reading) .Subscribe( onNext: result => { string nextTag = BitConverter.ToString(result.Data); nextTag = nextTag.Replace("-", ""); timer.Stop(); timer.Start(); rfidPlayer.OnNext(nextTag); }, onCompleted: () => { rfidPlayer.OnCompleted(); });根據(jù)西蒙娜爾的第一個答案，這似乎是令人滿意的： characteristic.WhenNotificationReceived() .Timeout(TimeSpan.FromSeconds(1)) .Subscribe( onNext: result => { string nextTag = BitConverter.ToString(result.Data); nextTag = nextTag.Replace("-", ""); rfidPlayer.OnNext(nextTag); }, onError: error => { rfidPlayer.OnCompleted(); });
3 回答
慕村9548890
TA貢獻1884條經(jīng)驗 獲得超4個贊
您可以考慮使用超時運算符 。唯一的缺點是它以錯誤信號終止。您可能需要處理錯誤
超時運算符允許您中止具有 onError 終止的可觀察對象，如果該可觀察對象在指定的時間跨度內(nèi)未能發(fā)出任何項。
如果您使用以下方法，則可以超越錯誤
.Timeout(200, Promise.resolve(42));
另一個變體允許您指示超時切換到您指定的備份 Observable，而不是在觸發(fā)超時條件時以錯誤終止。
characteristic.WhenNotificationReceived()
.Timeout(TimeSpan.FromSeconds(1))
.Subscribe(
onNext: result =>
{
....
rfidPlayer.OnNext(....);
},
onError: error =>
{
rfidPlayer.OnCompleted();
});
侃侃爾雅
TA貢獻1801條經(jīng)驗 獲得超16個贊
由于例外情況，我發(fā)現(xiàn)使用起來很討厭。Timeout
我更喜歡將值注入到可用于終止序列的序列中。例如，如果我的序列產(chǎn)生非負數(shù)，那么如果我注入一個我知道結(jié)束序列。-1
下面是一個示例：
從這個可觀察量開始，它生成從 1 開始的冪 2，并且它還延遲生成每個值的掛起值的毫秒數(shù)。
Observable
.Generate(1, x => true, x => 2 * x, x => x, x => TimeSpan.FromMilliseconds(x))
所以1，2，4，8等，越來越慢。
現(xiàn)在我想停止這個序列，如果秒沒有值，那么我可以這樣做：3.0
.Select(x => Observable.Timer(TimeSpan.FromSeconds(3.0)).Select(y => -1).StartWith(x))
.Switch()
.TakeWhile(x => x >= 0)
如果我運行此序列，我將得到以下輸出：
1
2
4
8
16
32
64
128
256
512
1024
2048
該序列即將生成，但它首先等待毫秒以產(chǎn)生該值 - 與此同時，觸發(fā)并輸出一個從而停止序列。40964096Observable.Timer(TimeSpan.FromSeconds(3.0))-1
此查詢的關(guān)鍵部分是使用 。它通過僅訂閱最新的外部可觀察和取消訂閱前一個外部可觀察來獲取并生成 a。SwitchIObservable<IObservable<T>>IObservable<T>
因此，在我的查詢中，序列生成的每個新值都會停止并重新啟動 .Timer
在你的例子中，你的可觀察量看起來像這樣：
characteristic
.WhenNotificationReceived()
.Select(result => BitConverter.ToString(result.Data).Replace("-", ""))
.Select(x => Observable.Timer(TimeSpan.FromSeconds(1.0)).Select(y => (string)null).StartWith(x))
.Switch()
.TakeWhile(x => x != null)
.Subscribe(rfidPlayer);
UYOU
TA貢獻1878條經(jīng)驗 獲得超4個贊
下面是一個可以使用的自定義運算符，它是內(nèi)置超時運算符頂部的薄層。TakeUntilTimeout
/// <summary>
/// Applies a timeout policy for each element in the observable sequence.
/// If the next element isn't received within the specified timeout duration
/// starting from its predecessor, the sequence terminates.
/// </summary>
public static IObservable<T> TakeUntilTimeout<T>(
this IObservable<T> source,
TimeSpan timeout)
{
return source.Timeout(timeout, Observable.Empty<T>());
}
- 3 回答
- 0 關(guān)注
- 96 瀏覽
添加回答
舉報
0/150
提交
取消