Throttle an IObservable based on value
c#, system.reactive
Solution
Here is a fairly compact solution. Your post a little ambigious about whether the duration would reset as soon as a distinct value arrives or not - so I have provided two solutions to both interpretations.
Variation 1 - Distinct "in between" values don't reset timer
This is when you care strictly about the duration for suppression and don't care if there are any "in between" values (as per McGarnagle's solution) - i.e. if you get `"a", "b" ,"a"` quickly, you still want to suppress the second `"a"`. Fortunately, this is very easy with a `GroupByUntil` which groups for the duration and emits the first element of each group:
public static IObservable<T> DistinctUntilChanged<T>(
this IObservable<T> source, TimeSpan duration, IScheduler scheduler)
{
if (scheduler == null) scheduler = Scheduler.Default;
return source.GroupByUntil(k => k,
_ => Observable.Timer(duration, scheduler))
.SelectMany(y => y.FirstAsync());
}
If you are wondering about the method name - I came up with Variation 2b first; I left the name above as it is so the unit tests still pass. It probably needs a better name like `SuppressDuplicatesWithinWindow` or similar...
Variation 2a - "In between" distinct values DO reset timer
This is slightly more complex - now any event in a different group will end a given group. I use the Publish().RefCount() combo to prevent more than one subscription to the source and have to be very careful with nulls:
public static IObservable<T> DistinctUntilChanged<T>(
this IObservable<T> source, TimeSpan duration, IScheduler scheduler)
{
if (scheduler == null) scheduler = Scheduler.Default;
var sourcePub = source.Publish().RefCount();
return sourcePub.GroupByUntil(
k => k,
x => Observable.Timer(duration, scheduler)
.TakeUntil(
sourcePub.Where(i => ReferenceEquals(null, i)
? !ReferenceEquals(null, x.Key)
: !i.Equals(x.Key))))
.SelectMany(y => y.FirstAsync());
}
Variation 2b
This is the original approach I tried, I've added it in as it's not all that bad now since my refinements to 2a have it more complex:
It is a variation of `Observable.DistinctUntilChanged` that accepts a duration. Given an event, successive duplicate events within that duration are suppressed. If a different event arrives, or an event arrives outside that duration, it is emitted and the suppression timer is reset.
It works by using the overload of `DistinctUntilChanged` that accepts an IEqualityComparer. The comparer considers events with a TimeStamp applied to be equal if the values match and the timestamps are within the specified duration.
public static partial class ObservableExtensions
{
public static IObservable<T> DistinctUntilChanged<T>(
this IObservable<T> source, TimeSpan duration, IScheduler scheduler)
{
if (scheduler == null) scheduler = Scheduler.Default;
return source.Timestamp(scheduler)
.DistinctUntilChanged(new Comparer<T>(duration))
.Select(ts => ts.Value);
}
private class Comparer<T> : IEqualityComparer<Timestamped<T>>
{
private readonly TimeSpan _duration;
public Comparer(TimeSpan duration)
{
_duration = duration;
}
public bool Equals(Timestamped<T> x, Timestamped<T> y)
{
if (y.Timestamp - x.Timestamp > _duration) return false;
return ReferenceEquals(x.Value, y.Value)
&& !ReferenceEquals(null,x.Value)
&& x.Value.Equals(y.Value);
}
public int GetHashCode(Timestamped<T> obj)
{
if (ReferenceEquals(null,obj.Value)) return obj.Timestamp.GetHashCode();
return obj.Value.GetHashCode() ^ obj.Timestamp.GetHashCode();
}
}
}
Here are the unit tests I used (include nuget packages rx-testing and nunit):
public class TestDistinct : ReactiveTest
{
[Test]
public void DuplicateWithinDurationIsSupressed()
{
var scheduler = new TestScheduler();
var source =scheduler.CreateColdObservable(
OnNext(100, "a"),
OnNext(200, "a"));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100, "a"));
}
[Test]
public void NonDuplicationWithinDurationIsNotSupressed()
{
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
OnNext(100, "a"),
OnNext(200, "b"));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100,"a"),
OnNext(200,"b"));
}
[Test]
public void DuplicateAfterDurationIsNotSupressed()
{
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
OnNext(100, "a"),
OnNext(400, "a"));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100, "a"),
OnNext(400, "a"));
}
[Test]
public void NonDuplicateAfterDurationIsNotSupressed()
{
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
OnNext(100, "a"),
OnNext(400, "b"));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100, "a"),
OnNext(400, "b"));
}
[Test]
public void TestWithSeveralValues()
{
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
OnNext(100, "a"),
OnNext(200, "a"),
OnNext(300, "b"),
OnNext(350, "c"),
OnNext(450, "b"),
OnNext(900, "a"));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100, "a"),
OnNext(300, "b"),
OnNext(350, "c"),
OnNext(450, "b"),
OnNext(900, "a"));
}
[Test]
public void CanHandleNulls()
{
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
OnNext(100, "a"),
OnNext(400, (string)null),
OnNext(500, "b"),
OnNext(600, (string)null),
OnNext(700, (string)null));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100, "a"),
OnNext(400, (string)null),
OnNext(500, "b"),
OnNext(600, (string)null));
}
[Test]
public void TwoDuplicatesWithinDurationAreSupressed()
{
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
OnNext(100, "a"),
OnNext(150, "a"),
OnNext(200, "a"));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100, "a"));
}
[Test]
public void TwoNullDuplicatesWithinDurationAreSupressed()
{
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
OnNext(100, (string)null),
OnNext(150, (string)null),
OnNext(200, (string)null));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100, (string)null));
}
}
Finally for completeness - Variation 1 would pass the following variant of the TestWithSeveralValues test:
[Test]
public void TestWithSeveralValuesVariation1()
{
var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable(
OnNext(100, "a"),
OnNext(200, "a"),
OnNext(300, "b"),
OnNext(350, "c"),
OnNext(450, "b"),
OnNext(900, "a"));
var duration = TimeSpan.FromTicks(250);
var results = scheduler.CreateObserver<string>();
source.DistinctUntilChanged(duration, scheduler).Subscribe(results);
scheduler.AdvanceBy(1000);
results.Messages.AssertEqual(
OnNext(100, "a"),
OnNext(300, "b"),
OnNext(350, "c"),
OnNext(900, "a"));
}
And the null test would change to have at the end:
results.Messages.AssertEqual(
OnNext(100, "a"),
OnNext(400, (string)null),
OnNext(500, "b"),
OnNext(700, (string)null)); /* This line changes */
Problem
I have an `IObservable<String>`. I am trying to detect (and handle) the case where the same string is notified in short succession. I want a filter/stream/observable such that if the same string is notified within 250ms of each other, it only notifies once. Not really sure where to start.