[BUG] Connect with predicate not showing initial state as previous when next updates occur in future.

[kepam]

Describe the bug
When connected with predicate initial state of the observable collection is not seen as previous when the next updates occur.
I think its bug, as it working differently before commit 63960b0, which change behaviour of connect().

Steps To Reproduce
Example of unit tests that reproduce the issue.
(in my opinion, both should pass, and they are doing that with code before 63960b0)

[Fact]
public void UpdatesExistedBeforeConnectWithPredicateShouldBeVisibleAsPreviousWhenNewUpdatesTriggered()
{
    // having
    var source = new SourceCache<int, int>(it => it);
    source.AddOrUpdate(1);
    var results = source.Connect(it => true).AsAggregator();
    
    // when
    source.AddOrUpdate(1);

    // then
    results.Messages.Count.Should().Be(2, "Should be 2 updates");
    results.Messages[1].First().Previous.HasValue.Should().Be(true, "Should have previous value");
}

[Fact]
public void UpdatesExistedBeforeConnectWithoutPredicateShouldBeVisibleAsPreviousWhenNewUpdatesTriggered()
{
    // having
    var source = new SourceCache<int, int>(it => it);
    source.AddOrUpdate(1);
    var results = source.Connect().AsAggregator();

    // when
    source.AddOrUpdate(1);

    // then
    results.Messages.Count.Should().Be(2, "Should be 2 updates");
    results.Messages[1].First().Previous.HasValue.Should().Be(true, "Should have previous value");
}

Expected behavior
The initial state should be visible as previous if we connect with non-null predicate.

Environment

• OS: Windows
• Version: 10
• Working Version: 6.16.6

Additional context
I think that problem is with the code in ObservableCache.cs in Connect()

var initial = GetInitialUpdates(predicate);
if (initial.Count != 0)
{
    observer.OnNext(initial);
}
var updateSource = (predicate == null ? _changes : _changes.Filter(predicate)).NotEmpty();

We do not take into account that we OnNext is not enough for proper handling of initial state.

Code below solve the problem

var initial = GetInitialUpdates(predicate);
var changes = Observable.Return(initial).Concat(_changes);
var updateSource = (predicate == null ? changes : changes.Filter(predicate)).NotEmpty();

Unfortunately, I looked into the history of bugs and I found this that stating that Observable.Return could trigger some memory leaks, so I'm not sure if this is the correct solution.

[RolandPheasant]

This is serious. I will look at it and hopefully find a solution which does not use Observable.Return

[ghost]

@RolandPheasant - Any updates for this bug? Can confirm I'm seeing it too in our application.

[j-fritsch]

@RolandPheasant - not trying to be a pest, but I'm also wondering if you have any updates on this? We happen to be clearing our SourceCache and seeing this issue, or at least something related. I can pin it down to the 6.15.4 release as well.

Here are some other unit tests.

[Fact]
public void ClearingSourceCacheWithPredicateShouldClearTheData()
{
    // having
    var source = new SourceCache<int, int>(it => it);
    source.AddOrUpdate(1);
    var results = source.Connect(it => true).AsAggregator();

    // when
    source.Clear();

    // then
    results.Data.Count.Should().Be(0, "Should be 0");
}

A potential workaround for some is to use .Filter instead.

[Fact]
public void ClearingSourceCacheWithFilterShouldClearTheData()
{
    // having
    var source = new SourceCache<int, int>(it => it);
    source.AddOrUpdate(1);
    var results = source.Connect().Filter(it => true).AsAggregator();
    
    // when
    source.Clear();
    
    // then
    results.Data.Count.Should().Be(0, "Should be 0");
}

[davecader]

It looks like I have encountered the same issue. This is a s big problem for us.

[RolandPheasant]

Sorry for the delay (very long).

The following is a hack to replace Observable.Return. All unit tests pass and I think in theory it should be thread safe.

        public IObservable<IChangeSet<TObject, TKey>> Connect(Func<TObject, bool>? predicate = null) =>
            Observable.Create<IChangeSet<TObject, TKey>>(
                observer =>
                {
                    // dumb hack of observable.return which does not create memory leak
                    IObservable<IChangeSet<TObject, TKey>> Initial()
                    {
                        return Observable.Create<IChangeSet<TObject, TKey>>(o =>
                        {
                            // lock getting initial changes and rely on a combination of Concat
                            // + _changes being synchronized to produce thread safety  (I hope!)
                            lock (_locker)
                            {
                                var initial = GetInitialUpdates(predicate);
                                o.OnNext(initial);
                                o.OnCompleted();
                                return () => { };
                            }
                        });
                    }

                    var changes = Observable.Defer(Initial).Concat(_changes);
                    if (predicate != null)
                    {
                        changes = changes.Filter(predicate);
                    }
                    else
                    {
                        changes = changes.NotEmpty();
                    }

                    return changes.SubscribeSafe(observer);
                });

All unit tests pass. I just need to sit on it overnight to think about whether there are any ramifications + I will also pull the observable return hack into an extension method as it will also be need elsewhere.

[github-actions]

This issue has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.