DetectStale within a stream

The DetectStale<T> extension method within the Reactive Trader RX example application detects when a stream has stopped producing values and injects an IsStale indicator to allow the UI to react.

The idea is that while the stream is steadily providing values all is good.  As soon as a value hasn’t been received within a set period of time, a new value is sent down the stream to say “this steam has gone stale”.

First here is an example of it being used within a test:

The DetectStale internally uses a timer to know whether it should be sending in effect an null/empty stale value. Here are another couple of tests for the DetectStale extension method:

Here is the DetectStale code itself: You can the source subscription on line 42 calling the normal/expected OnNext method with the value.  After that they call the scheduleStale method containing the timer with Stale / null value.