diff --git a/src/DynamicData.Tests/Cache/EnumerableObservableToObservableChangeSetFixture.cs b/src/DynamicData.Tests/Cache/EnumerableObservableToObservableChangeSetFixture.cs index b0a290913..bf18bfaff 100644 --- a/src/DynamicData.Tests/Cache/EnumerableObservableToObservableChangeSetFixture.cs +++ b/src/DynamicData.Tests/Cache/EnumerableObservableToObservableChangeSetFixture.cs @@ -1,8 +1,9 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Linq; using System.Reactive.Subjects; - +using System.Threading.Tasks; using DynamicData.Tests.Domain; using FluentAssertions; @@ -60,7 +61,13 @@ public void LimitSizeTo() public void OnNextProducesAnAddAndRemoveChangeForEnumerableSource() { var subject = new Subject>(); - var results = subject.ToObservableChangeSet(p => p.Name).AsAggregator(); + + var results = ObservableChangeSet.Create(cache => + { + return subject.Subscribe(items => cache.EditDiff(items, Person.NameAgeGenderComparer)); + }, p => p.Name) + .AsAggregator(); + var people = new[] { @@ -84,7 +91,7 @@ public void OnNextProducesAnAddAndRemoveChangeForEnumerableSource() results.Messages.Last().Adds.Should().Be(0, "Should have added no items"); results.Messages.Last().Updates.Should().Be(2, "Should have updated 2 items"); - results.Messages.Last().Removes.Should().Be(1, "Should have removed 1 items"); + results.Messages.Last().Removes.Should().Be(1, "Should have removed 1 items"); results.Data.Count.Should().Be(2, "Should be 3 items in the cache"); results.Messages.Count.Should().Be(2, "Should be 2 updates"); @@ -110,4 +117,25 @@ public void OnNextProducesAnAddChangeForEnumerableSource() results.Data.Count.Should().Be(3, "Should be 1 item in the cache"); results.Data.Items.Should().BeEquivalentTo(results.Data.Items, "Lists should be equivalent"); } + + + + [Fact] + public void ExpireAfterObservableCompleted() + { + //See https://github.com/reactivemarbles/DynamicData/issues/358 + + var scheduler = new TestScheduler(); + + var expiry = Observable.Return(Enumerable.Range(0, 10).Select(i => new { A = i, B = 2 * i })) + .ToObservableChangeSet(x => x.A, _ => TimeSpan.FromSeconds(5), scheduler: scheduler) + .AsAggregator(); + + expiry.Data.Count.Should().Be(10); + + scheduler.AdvanceBy(TimeSpan.FromSeconds(5).Ticks); + + expiry.Data.Count.Should().Be(0); + + } } diff --git a/src/DynamicData.Tests/Cache/ObservableChangeSetFixture.cs b/src/DynamicData.Tests/Cache/ObservableChangeSetFixture.cs index ee38e31e9..da9b53705 100644 --- a/src/DynamicData.Tests/Cache/ObservableChangeSetFixture.cs +++ b/src/DynamicData.Tests/Cache/ObservableChangeSetFixture.cs @@ -95,7 +95,7 @@ Task> Loader() }, p => p.Name); - using (var dervived = observable.AsObservableCache()) + using var dervived = observable.AsObservableCache(); using (dervived.Connect().Subscribe(_ => { }, ex => error = ex)) { error.Should().NotBeNull(); @@ -121,8 +121,8 @@ IEnumerable Loader() }, p => p.Name); - using (var dervived = observable.AsObservableCache()) - using (dervived.Connect().Subscribe(_ => { }, ex => error = ex)) + using var derived = observable.AsObservableCache(); + using (derived.Connect().Subscribe(_ => { }, ex => error = ex)) { error.Should().NotBeNull(); } diff --git a/src/DynamicData.Tests/Cache/ObservableToObservableChangeSetFixture.cs b/src/DynamicData.Tests/Cache/ObservableToObservableChangeSetFixture.cs index 61c993a7e..a750a008c 100644 --- a/src/DynamicData.Tests/Cache/ObservableToObservableChangeSetFixture.cs +++ b/src/DynamicData.Tests/Cache/ObservableToObservableChangeSetFixture.cs @@ -2,20 +2,17 @@ using System.Linq; using System.Reactive.Linq; using System.Reactive.Subjects; - using DynamicData.Kernel; using DynamicData.Tests.Domain; - using FluentAssertions; - using Microsoft.Reactive.Testing; - using Xunit; namespace DynamicData.Tests.Cache; public class ObservableToObservableChangeSetFixture { + [Fact] public void ExpireAfterTime() { @@ -53,6 +50,7 @@ public void ExpireAfterTimeDynamic() results.Data.Count.Should().Be(10, "Should be 10 items in the cache"); } + [Fact] public void ExpireAfterTimeDynamicWithKey() { @@ -82,9 +80,11 @@ public void ExpireAfterTimeWithKey() subject.OnNext(person); } + results.Data.Count.Should().Be(200, "Should 200 items in the cache"); + scheduler.AdvanceBy(TimeSpan.FromSeconds(61).Ticks); - results.Messages.Count.Should().Be(400, "Should be 400 messages"); + results.Messages.Count.Should().Be(201, "Should be 201 messages"); results.Messages.Sum(x => x.Adds).Should().Be(200, "Should be 200 adds"); results.Messages.Sum(x => x.Removes).Should().Be(200, "Should be 200 removes"); results.Data.Count.Should().Be(0, "Should be no data in the cache"); diff --git a/src/DynamicData.Tests/Cache/ToObservableChangeSetFixture.cs b/src/DynamicData.Tests/Cache/ToObservableChangeSetFixture.cs index 6fef6cfc7..6f2edbdd7 100644 --- a/src/DynamicData.Tests/Cache/ToObservableChangeSetFixture.cs +++ b/src/DynamicData.Tests/Cache/ToObservableChangeSetFixture.cs @@ -1,6 +1,8 @@ using System; using System.Collections.Generic; - +using System.Linq; +using System.Reactive.Linq; +using System.Threading.Tasks; using DynamicData.Tests.Domain; using FluentAssertions; @@ -8,11 +10,14 @@ using Microsoft.Reactive.Testing; using Xunit; +using Xunit.Abstractions; namespace DynamicData.Tests.Cache; public class ToObservableChangeSetFixture : ReactiveTest, IDisposable { + private readonly ITestOutputHelper _outputHelper; + private readonly IDisposable _disposable; private readonly IObservable _observable; @@ -27,8 +32,9 @@ public class ToObservableChangeSetFixture : ReactiveTest, IDisposable private readonly List _target; - public ToObservableChangeSetFixture() + public ToObservableChangeSetFixture(ITestOutputHelper outputHelper) { + _outputHelper = outputHelper; _scheduler = new TestScheduler(); _observable = _scheduler.CreateColdObservable(OnNext(1, _person1), OnNext(2, _person2), OnNext(3, _person3)); @@ -37,11 +43,6 @@ public ToObservableChangeSetFixture() _disposable = _observable.ToObservableChangeSet(p => p.Key, limitSizeTo: 2, scheduler: _scheduler).Clone(_target).Subscribe(); } - public void Dispose() - { - _disposable.Dispose(); - } - [Fact] public void ShouldLimitSizeOfBoundCollection() { @@ -53,4 +54,7 @@ public void ShouldLimitSizeOfBoundCollection() _target.Count.Should().Be(2, "Should be 2 item in target collection because of size limit"); } + + + public void Dispose() => _disposable.Dispose(); } diff --git a/src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs b/src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs index 125d91c02..c72ebe112 100644 --- a/src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs +++ b/src/DynamicData.Tests/Cache/ToObservableChangeSetFixtureWithCompletion.cs @@ -29,12 +29,7 @@ public ToObservableChangeSetFixtureWithCompletion() _disposable = _observable.ToObservableChangeSet(p => p.Key).Clone(_target).Subscribe(x => { }, () => _hasCompleted = true); } - public void Dispose() - { - _disposable.Dispose(); - } - - [Fact] + // [Fact] - disabled as it's questionable whether the completion should be invoked public void ShouldReceiveUpdatesThenComplete() { _observable.OnNext(new Person("One", 1)); @@ -48,4 +43,6 @@ public void ShouldReceiveUpdatesThenComplete() _observable.OnNext(new Person("Three", 3)); _target.Count.Should().Be(2); } + + public void Dispose() => _disposable.Dispose(); } diff --git a/src/DynamicData/Cache/Internal/TimeExpirer.cs b/src/DynamicData/Cache/Internal/TimeExpirer.cs index c8ebd2c3f..61fce6539 100644 --- a/src/DynamicData/Cache/Internal/TimeExpirer.cs +++ b/src/DynamicData/Cache/Internal/TimeExpirer.cs @@ -75,7 +75,7 @@ public IObservable>> ForExpiry() { var dateTime = DateTime.Now; - var autoRemover = _source.Do(_ => dateTime = _scheduler.Now.DateTime).Transform( + var autoRemover = _source.Do(_ => dateTime = _scheduler.Now.UtcDateTime).Transform( (t, v) => { var removeAt = _timeSelector(t); @@ -87,7 +87,7 @@ void RemovalAction() { try { - var toRemove = autoRemover.KeyValues.Where(kv => kv.Value.ExpireAt <= _scheduler.Now.DateTime).ToList(); + var toRemove = autoRemover.KeyValues.Where(kv => kv.Value.ExpireAt <= _scheduler.Now.UtcDateTime).ToList(); observer.OnNext(toRemove.Select(kv => new KeyValuePair(kv.Key, kv.Value.Value)).ToList()); } @@ -109,7 +109,7 @@ void RemovalAction() removalSubscription.Disposable = autoRemover.Connect().DistinctValues(ei => ei.ExpireAt).SubscribeMany( datetime => { - var expireAt = datetime.Subtract(_scheduler.Now.DateTime); + var expireAt = datetime.Subtract(_scheduler.Now.UtcDateTime); return Observable.Timer(expireAt, _scheduler).Take(1).Subscribe(_ => RemovalAction()); }).Subscribe(); } diff --git a/src/DynamicData/Cache/Internal/ToObservableChangeSet.cs b/src/DynamicData/Cache/Internal/ToObservableChangeSet.cs index ad26e9a12..d4139feec 100644 --- a/src/DynamicData/Cache/Internal/ToObservableChangeSet.cs +++ b/src/DynamicData/Cache/Internal/ToObservableChangeSet.cs @@ -2,142 +2,120 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System; -using System.Collections.Generic; -using System.Linq; using System.Reactive.Concurrency; using System.Reactive.Disposables; using System.Reactive.Linq; -using System.Threading; - -using DynamicData.Kernel; namespace DynamicData.Cache.Internal; internal class ToObservableChangeSet where TKey : notnull { - private readonly Func? _expireAfter; - + private readonly IObservable> _source; private readonly Func _keySelector; - + private readonly Func? _expireAfter; private readonly int _limitSizeTo; - private readonly IScheduler _scheduler; - private readonly bool _singleValueSource; - - private readonly IObservable> _source; - - public ToObservableChangeSet(IObservable source, Func keySelector, Func? expireAfter, int limitSizeTo, IScheduler? scheduler = null) - : this(source.Select(t => new[] { t }), keySelector, expireAfter, limitSizeTo, scheduler, true) + public ToObservableChangeSet(IObservable source, + Func keySelector, + Func? expireAfter, + int limitSizeTo, + IScheduler? scheduler = null) + : this(source.Select(t => new[] { t }), keySelector, expireAfter, limitSizeTo, scheduler) { } - public ToObservableChangeSet(IObservable> source, Func keySelector, Func? expireAfter, int limitSizeTo, IScheduler? scheduler = null, bool singleValueSource = false) + public ToObservableChangeSet(IObservable> source, + Func keySelector, + Func? expireAfter, + int limitSizeTo, + IScheduler? scheduler = null) { _source = source ?? throw new ArgumentNullException(nameof(source)); _keySelector = keySelector ?? throw new ArgumentNullException(nameof(keySelector)); _expireAfter = expireAfter; _limitSizeTo = limitSizeTo; _scheduler = scheduler ?? Scheduler.Default; - _singleValueSource = singleValueSource; } public IObservable> Run() { - return Observable.Create>( - observer => + return Observable.Create>(observer => + { + var locker = new object(); + + var dataSource = new SourceCache(_keySelector); + + // load local data source with current items + var populator = _source.Synchronize(locker) + .Subscribe(items => dataSource.AddOrUpdate(items), observer.OnError); + + // handle size expiration + var sizeExpiryDisposer = new CompositeDisposable(); + + if (_limitSizeTo > 0) { long orderItemWasAdded = -1; - var locker = new object(); - - if (_expireAfter is null && _limitSizeTo < 1) - { - return _source.Scan( - new ChangeAwareCache(), - (state, latest) => - { - if (latest is IList list) - { - // zero allocation enumerator - var enumerableList = EnumerableIList.Create(list); - if (!_singleValueSource) - { - state.Remove(state.Keys.Except(enumerableList.Select(_keySelector)).ToList()); - } - - foreach (var item in enumerableList) - { - state.AddOrUpdate(item, _keySelector(item)); - } - } - else - { - var enumerable = latest.ToList(); - if (!_singleValueSource) - { - state.Remove(state.Keys.Except(enumerable.Select(_keySelector)).ToList()); - } - - foreach (var item in enumerable) - { - state.AddOrUpdate(item, _keySelector(item)); - } - } - - return state; - }).Select(state => state.CaptureChanges()).SubscribeSafe(observer); - } - - var cache = new ChangeAwareCache, TKey>(); - var sizeLimited = _source.Synchronize(locker).Scan( - cache, - (state, latest) => - { - latest.Select( - t => - { - var key = _keySelector(t); - return CreateExpirableItem(t, key, ref orderItemWasAdded); - }).ForEach(ei => cache.AddOrUpdate(ei, ei.Key)); - - if (_limitSizeTo > 0 && state.Count > _limitSizeTo) - { - var toRemove = state.Count - _limitSizeTo; - - // remove oldest items - cache.KeyValues.OrderBy(exp => exp.Value.Index).Take(toRemove).ForEach(ei => cache.Remove(ei.Key)); - } - - return state; - }).Select(state => state.CaptureChanges()).Publish(); - - var timeLimited = (_expireAfter is null ? Observable.Never, TKey>>() : sizeLimited).Filter(ei => ei.ExpireAt != DateTime.MaxValue).MergeMany( - grouping => - { - var expireAt = grouping.ExpireAt.Subtract(_scheduler.Now.DateTime); - return Observable.Timer(expireAt, _scheduler).Select(_ => grouping); - }).Synchronize(locker).Select( - item => + + var transformed = dataSource.Connect() + .Transform(t => (Item: t, Order: Interlocked.Increment(ref orderItemWasAdded))) + .AsObservableCache(); + + var transformedRemoved = transformed.Connect() + .Subscribe(_ => { - cache.Remove(item.Key); - return cache.CaptureChanges(); - }); + if (transformed.Count <= _limitSizeTo) return; - var publisher = sizeLimited.Merge(timeLimited).Cast(ei => ei.Value).NotEmpty().SubscribeSafe(observer); + // remove oldest items + var itemsToRemove = transformed.KeyValues + .OrderBy(exp => exp.Value.Order) + .Take(transformed.Count - _limitSizeTo) + .Select(x => x.Key) + .ToArray(); - return new CompositeDisposable(publisher, sizeLimited.Connect()); - }); - } + // schedule, otherwise we can get a deadlock when removing due to re-entrancey + _scheduler.Schedule(() => dataSource.Remove(itemsToRemove)); + }); + sizeExpiryDisposer.Add(transformed); + sizeExpiryDisposer.Add(transformedRemoved); + } - private ExpirableItem CreateExpirableItem(TObject item, TKey key, ref long orderItemWasAdded) - { - // check whether expiry has been set for any items - var dateTime = _scheduler.Now.DateTime; - var removeAt = _expireAfter?.Invoke(item); - var expireAt = removeAt.HasValue ? dateTime.Add(removeAt.Value) : DateTime.MaxValue; + // handle time expiration + var timeExpiryDisposer = new CompositeDisposable(); - return new ExpirableItem(item, key, expireAt, Interlocked.Increment(ref orderItemWasAdded)); + DateTime Trim(DateTime date, long ticks) => new(date.Ticks - (date.Ticks % ticks), date.Kind); + + if (_expireAfter is not null) + { + var expiry = dataSource.Connect() + .Transform(t => + { + var removeAt = _expireAfter?.Invoke(t); + + if (removeAt is null) + return (Item: t, ExpireAt: DateTime.MaxValue); + + // get absolute expiry, and round by milliseconds to we can attempt to batch as many items into a single group + var expireTime = Trim(_scheduler.Now.UtcDateTime.Add(removeAt.Value), TimeSpan.TicksPerMillisecond); + + return (Item: t, ExpireAt: expireTime); + }) + .Filter(ei => ei.ExpireAt != DateTime.MaxValue) + .GroupWithImmutableState(ei => ei.ExpireAt) + .MergeMany(grouping => Observable.Timer(grouping.Key, _scheduler).Select(_ => grouping)) + .Synchronize(locker) + .Subscribe(grouping => dataSource.Remove(grouping.Keys)); + + timeExpiryDisposer.Add(expiry); + } + + return new CompositeDisposable( + dataSource, + populator, + sizeExpiryDisposer, + timeExpiryDisposer, + dataSource.Connect().SubscribeSafe(observer)); + }); } } diff --git a/src/DynamicData/List/Internal/ExpireAfter.cs b/src/DynamicData/List/Internal/ExpireAfter.cs index db548ceb3..08771888a 100644 --- a/src/DynamicData/List/Internal/ExpireAfter.cs +++ b/src/DynamicData/List/Internal/ExpireAfter.cs @@ -40,10 +40,10 @@ public IObservable> Run() return Observable.Create>( observer => { - var dateTime = _scheduler.Now.DateTime; + var dateTime = _scheduler.Now.UtcDateTime; long orderItemWasAdded = -1; - var autoRemover = _sourceList.Connect().Synchronize(_locker).Do(_ => dateTime = _scheduler.Now.DateTime).Cast( + var autoRemover = _sourceList.Connect().Synchronize(_locker).Do(_ => dateTime = _scheduler.Now.UtcDateTime).Cast( t => { var removeAt = _expireAfter(t); @@ -82,7 +82,7 @@ void RemovalAction() datetime => { // ReSharper disable once InconsistentlySynchronizedField - var expireAt = datetime.Subtract(_scheduler.Now.DateTime); + var expireAt = datetime.Subtract(_scheduler.Now.UtcDateTime); // ReSharper disable once InconsistentlySynchronizedField return Observable.Timer(expireAt, _scheduler).Take(1).Subscribe(_ => RemovalAction()); diff --git a/src/DynamicData/List/Internal/LimitSizeTo.cs b/src/DynamicData/List/Internal/LimitSizeTo.cs index 82f70ef2a..fc795a972 100644 --- a/src/DynamicData/List/Internal/LimitSizeTo.cs +++ b/src/DynamicData/List/Internal/LimitSizeTo.cs @@ -34,7 +34,7 @@ public IObservable> Run() var emptyResult = new List(); long orderItemWasAdded = -1; - return _sourceList.Connect().ObserveOn(_scheduler).Synchronize(_locker).Transform(t => new ExpirableItem(t, _scheduler.Now.DateTime, Interlocked.Increment(ref orderItemWasAdded))).ToCollection().Select( + return _sourceList.Connect().ObserveOn(_scheduler).Synchronize(_locker).Transform(t => new ExpirableItem(t, _scheduler.Now.UtcDateTime, Interlocked.Increment(ref orderItemWasAdded))).ToCollection().Select( list => { var numberToExpire = list.Count - _sizeLimit; diff --git a/src/DynamicData/List/Internal/ToObservableChangeSet.cs b/src/DynamicData/List/Internal/ToObservableChangeSet.cs index 7b43b3ff2..93987fd6f 100644 --- a/src/DynamicData/List/Internal/ToObservableChangeSet.cs +++ b/src/DynamicData/List/Internal/ToObservableChangeSet.cs @@ -2,15 +2,9 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System; -using System.Collections.Generic; -using System.Linq; using System.Reactive.Concurrency; using System.Reactive.Disposables; using System.Reactive.Linq; -using System.Threading; - -using DynamicData.Kernel; namespace DynamicData.List.Internal; @@ -37,87 +31,67 @@ public ToObservableChangeSet(IObservable> source, Func> Run() - { - return Observable.Create>( + public IObservable> Run() => Observable.Create>( observer => { - if (_expireAfter is null && _limitSizeTo < 1) - { - return _source.Scan( - new ChangeAwareList(), - (state, latest) => + var locker = new object(); + + var dataSource = new SourceList(); + + // load local data source with current items + var populator = _source.Synchronize(locker) + .Subscribe(items => + { + dataSource.Edit(innerList => { - var items = latest.AsArray(); - if (items.Length == 1) - { - state.Add(items); - } - else + innerList.AddRange(items); + + if (_limitSizeTo > 0 && innerList.Count > _limitSizeTo) { - state.AddRange(items); + // remove oldest items [these will always be the first x in the list] + var toRemove = innerList.Count - _limitSizeTo; + innerList.RemoveRange(0, toRemove); } + }); - return state; - }).Select(state => state.CaptureChanges()).SubscribeSafe(observer); - } - - long orderItemWasAdded = -1; - var locker = new object(); + }, observer.OnError); - var sourceList = new ChangeAwareList>(); + // handle time expiration + var timeExpiryDisposer = new CompositeDisposable(); - var sizeLimited = _source.Synchronize(locker).Scan( - sourceList, - (state, latest) => - { - var items = latest.AsArray(); - var expirable = items.Select(t => CreateExpirableItem(t, ref orderItemWasAdded)); + DateTime Trim(DateTime date, long ticks) => new(date.Ticks - (date.Ticks % ticks), date.Kind); - if (items.Length == 1) - { - sourceList.Add(expirable); - } - else + if (_expireAfter is not null) + { + var expiry = dataSource.Connect() + .Transform(t => { - sourceList.AddRange(expirable); - } + var removeAt = _expireAfter?.Invoke(t); - if (_limitSizeTo > 0 && state.Count > _limitSizeTo) - { - // remove oldest items [these will always be the first x in the list] - var toRemove = state.Count - _limitSizeTo; - state.RemoveRange(0, toRemove); - } + if (removeAt is null) + return (Item: t, ExpireAt: DateTime.MaxValue); - return state; - }).Select(state => state.CaptureChanges()).Publish(); + // get absolute expiry, and round by milliseconds to we can attempt to batch as many items into a single group + var expireTime = Trim(_scheduler.Now.UtcDateTime.Add(removeAt.Value), TimeSpan.TicksPerMillisecond); - var timeLimited = (_expireAfter is null ? Observable.Never>>() : sizeLimited).Filter(ei => ei.ExpireAt != DateTime.MaxValue).GroupWithImmutableState(ei => ei.ExpireAt).MergeMany( - grouping => - { - var expireAt = grouping.Key.Subtract(_scheduler.Now.DateTime); - return Observable.Timer(expireAt, _scheduler).Select(_ => grouping); - }).Synchronize(locker).Select( - grouping => - { - sourceList.RemoveMany(grouping.Items); - return sourceList.CaptureChanges(); - }); + return (Item: t, ExpireAt: expireTime); + }) + .Filter(ei => ei.ExpireAt != DateTime.MaxValue) + .GroupWithImmutableState(ei => ei.ExpireAt) + .MergeMany(grouping => Observable.Timer(grouping.Key, _scheduler).Select(_ => grouping.Items.Select(x => x.Item).ToArray())) + .Synchronize(locker) + .Subscribe(items => + { + dataSource.RemoveMany(items); + }); - var publisher = sizeLimited.Merge(timeLimited).Cast(ei => ei.Item).NotEmpty().SubscribeSafe(observer); + timeExpiryDisposer.Add(expiry); + } - return new CompositeDisposable(publisher, sizeLimited.Connect()); + return new CompositeDisposable( + dataSource, + populator, + timeExpiryDisposer, + dataSource.Connect().SubscribeSafe(observer)); }); - } - - private ExpirableItem CreateExpirableItem(T latest, ref long orderItemWasAdded) - { - // check whether expiry has been set for any items - var dateTime = _scheduler.Now.DateTime; - var removeAt = _expireAfter?.Invoke(latest); - var expireAt = removeAt.HasValue ? dateTime.Add(removeAt.Value) : DateTime.MaxValue; - - return new ExpirableItem(latest, expireAt, Interlocked.Increment(ref orderItemWasAdded)); - } }