diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 32783156d..d2ff6182d 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -107,27 +107,22 @@ public IObservable> Connect(Func? predi Observable.Create>( observer => { - var initial = InternalEx.Return(() => + lock (_locker) { - // lock getting initial changes and rely on a combination of Concat - // + _changes being synchronized to produce thread safety (I hope!) - lock (_locker) + var initial = InternalEx.Return(() => (IChangeSet)GetInitialUpdates(predicate)); + var changes = initial.Concat(_changes); + + if (predicate != null) { - return (IChangeSet)GetInitialUpdates(predicate); + changes = changes.Filter(predicate, suppressEmptyChangeSets); + } + else if (suppressEmptyChangeSets) + { + changes = changes.NotEmpty(); } - }); - var changes = Observable.Defer(() => initial).Concat(_changes); - if (predicate != null) - { - changes = changes.Filter(predicate, suppressEmptyChangeSets); - } - else if (suppressEmptyChangeSets) - { - changes = changes.NotEmpty(); + return changes.SubscribeSafe(observer); } - - return changes.SubscribeSafe(observer); }); public void Dispose() => _cleanUp.Dispose();