Skip to content

Fixed initialization logic for Join operators #945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JakenVeina merged 2 commits into from
Dec 24, 2024
Merged

Fixed initialization logic for Join operators #945

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
21921f9
Fixed initialization logic for Join operators, to ensure that only on…
JakenVeina Sep 11, 2024
10a79cd
Merge branch 'main' into issues/943
JakenVeina Dec 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions src/DynamicData.Tests/Cache/InnerJoinFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
using System;

using DynamicData.Tests.Utilities;

using FluentAssertions;

using Xunit;
Expand Down Expand Up @@ -206,6 +208,30 @@ public void MoreRight()

}

[Fact]
public void InitializationWaitsForBothSources()
{
var left = new[] { 1, 2, 3 };
var right = new[] { 4, 6, 2 };

ObservableCacheEx
.InnerJoin(
left: left.AsObservableChangeSet(static left => 2 * left),
right: right.AsObservableChangeSet(static right => right),
rightKeySelector: static right => right,
resultSelector: static (left, right) => (left, right))
.ValidateSynchronization()
.ValidateChangeSets(static pair => (2 * pair.left, pair.right))
.RecordCacheItems(out var results);

results.Error.Should().BeNull();

results.RecordedChangeSets.Count.Should().Be(1, "Initialization should only emit one changeset.");
results.RecordedChangeSets[0].Should().OnlyContain(change => change.Reason == ChangeReason.Add, "Initialization should only emit Add changes.");

results.RecordedItemsByKey.Values.Should().OnlyContain(pair => (2 * pair.left) == pair.right, "Source items should have been joined correctly");
}

public class Device(string name) : IEquatable<Device>
{
public string Name { get; } = name;
Expand Down
27 changes: 27 additions & 0 deletions src/DynamicData.Tests/Cache/LeftJoinFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
using System.Linq;

using DynamicData.Kernel;
using DynamicData.Tests.Utilities;

using FluentAssertions;

Expand Down Expand Up @@ -165,6 +166,32 @@ public void UpdateRight()
_result.Data.Items.All(dwm => dwm.MetaData != Optional<DeviceMetaData>.None).Should().BeTrue();
}

[Fact]
public void InitializationWaitsForBothSources()
{
// https://github.com/reactivemarbles/DynamicData/issues/943

var left = new[] { 1, 2, 3 };
var right = new[] { 4, 6, 2 };

ObservableCacheEx
.LeftJoin(
left: left.AsObservableChangeSet(static left => 2 * left),
right: right.AsObservableChangeSet(static right => right),
rightKeySelector: static right => right,
resultSelector: static (left, right) => (left, right: right.HasValue ? right.Value : default(int?)))
.ValidateSynchronization()
.ValidateChangeSets(static pair => 2 * pair.left)
.RecordCacheItems(out var results);

results.Error.Should().BeNull();

results.RecordedChangeSets.Count.Should().Be(1, "Initialization should only emit one changeset.");
results.RecordedChangeSets[0].Should().OnlyContain(change => change.Reason == ChangeReason.Add, "Initialization should only emit Add changes.");

results.RecordedItemsByKey.Values.Should().OnlyContain(pair => (2 * pair.left) == pair.right, "Source items should have been joined correctly");
}

public class Device(string name) : IEquatable<Device>
{
public string Name { get; } = name;
Expand Down
23 changes: 23 additions & 0 deletions src/DynamicData.Tests/Cache/RightJoinFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
using System.Linq;

using DynamicData.Kernel;
using DynamicData.Tests.Utilities;

using FluentAssertions;

Expand Down Expand Up @@ -216,7 +217,29 @@ public void MoreRight()
_result.Data.Items.Count(dwm => dwm.Device == Optional<Device>.None).Should().Be(1);
}

[Fact]
public void InitializationWaitsForBothSources()
{
var left = new[] { 1, 2, 3 };
var right = new[] { 4, 6, 2 };

ObservableCacheEx
.RightJoin(
left: left.AsObservableChangeSet(static left => 2 * left),
right: right.AsObservableChangeSet(static right => right),
rightKeySelector: static right => right,
resultSelector: static (left, right) => (left: left.HasValue ? left.Value : default(int?), right))
.ValidateSynchronization()
.ValidateChangeSets(static pair => pair.right)
.RecordCacheItems(out var results);

results.Error.Should().BeNull();

results.RecordedChangeSets.Count.Should().Be(1, "Initialization should only emit one changeset.");
results.RecordedChangeSets[0].Should().OnlyContain(change => change.Reason == ChangeReason.Add, "Initialization should only emit Add changes.");

results.RecordedItemsByKey.Values.Should().OnlyContain(pair => (2 * pair.left) == pair.right, "Source items should have been joined correctly");
}

public class Device(string name) : IEquatable<Device>
{
Expand Down
81 changes: 45 additions & 36 deletions src/DynamicData/Cache/Internal/InnerJoin.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,47 +37,52 @@ internal sealed class InnerJoin<TLeft, TLeftKey, TRight, TRightKey, TDestination
// joined is the final cache
var joinedCache = new ChangeAwareCache<TDestination, (TLeftKey, TRightKey)>();

var leftLoader = leftCache.Connect().Select(changes =>
{
foreach (var change in changes.ToConcreteType())
{
var leftCurrent = change.Current;
var rightLookup = rightGrouped.Lookup(change.Key);
var hasInitialized = false;

if (rightLookup.HasValue)
var leftLoader = leftCache.Connect()
.Select(changes =>
{
foreach (var change in changes.ToConcreteType())
{
switch (change.Reason)
var leftCurrent = change.Current;
var rightLookup = rightGrouped.Lookup(change.Key);

if (rightLookup.HasValue)
{
case ChangeReason.Add:
case ChangeReason.Update:
foreach (var keyvalue in rightLookup.Value.KeyValues)
{
joinedCache.AddOrUpdate(_resultSelector((change.Key, keyvalue.Key), leftCurrent, keyvalue.Value), (change.Key, keyvalue.Key));
}

break;

case ChangeReason.Remove:
foreach (var keyvalue in rightLookup.Value.KeyValues)
{
joinedCache.Remove((change.Key, keyvalue.Key));
}

break;

case ChangeReason.Refresh:
foreach (var key in rightLookup.Value.Keys)
{
joinedCache.Refresh((change.Key, key));
}

break;
switch (change.Reason)
{
case ChangeReason.Add:
case ChangeReason.Update:
foreach (var keyvalue in rightLookup.Value.KeyValues)
{
joinedCache.AddOrUpdate(_resultSelector((change.Key, keyvalue.Key), leftCurrent, keyvalue.Value), (change.Key, keyvalue.Key));
}

break;

case ChangeReason.Remove:
foreach (var keyvalue in rightLookup.Value.KeyValues)
{
joinedCache.Remove((change.Key, keyvalue.Key));
}

break;

case ChangeReason.Refresh:
foreach (var key in rightLookup.Value.Keys)
{
joinedCache.Refresh((change.Key, key));
}

break;
}
}
}
}

return joinedCache.CaptureChanges();
});
return joinedCache.CaptureChanges();
})
// Don't forward initial changesets from the left side, only the right
.Where(_ => hasInitialized);

var rightLoader = rightCache.Connect().Select(changes =>
{
Expand Down Expand Up @@ -119,7 +124,11 @@ internal sealed class InnerJoin<TLeft, TLeftKey, TRight, TRightKey, TDestination

lock (locker)
{
return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache, rightShare.Connect());
var observerSubscription = leftLoader.Merge(rightLoader).SubscribeSafe(observer);

hasInitialized = true;

return new CompositeDisposable(observerSubscription, leftCache, rightCache, rightShare.Connect());
}
});
}
19 changes: 14 additions & 5 deletions src/DynamicData/Cache/Internal/LeftJoin.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,12 +30,15 @@ public IObservable<IChangeSet<TDestination, TLeftKey>> Run() => Observable.Creat
var locker = new object();

// create local backing stores
var leftCache = _left.Synchronize(locker).AsObservableCache(false);
var leftShare = _left.Synchronize(locker).Publish();
var leftCache = leftShare.AsObservableCache(false);
var rightCache = _right.Synchronize(locker).ChangeKey(_rightKeySelector).AsObservableCache(false);

// joined is the final cache
var joined = new ChangeAwareCache<TDestination, TLeftKey>();

var hasInitialized = false;

var leftLoader = leftCache.Connect().Select(
changes =>
{
Expand Down Expand Up @@ -66,8 +69,8 @@ public IObservable<IChangeSet<TDestination, TLeftKey>> Run() => Observable.Creat
return joined.CaptureChanges();
});

var rightLoader = rightCache.Connect().Select(
changes =>
var rightLoader = rightCache.Connect()
.Select(changes =>
{
foreach (var change in changes.ToConcreteType())
{
Expand Down Expand Up @@ -117,11 +120,17 @@ public IObservable<IChangeSet<TDestination, TLeftKey>> Run() => Observable.Creat
}

return joined.CaptureChanges();
});
})
// Don't forward initial changesets from the right side, only the left
.Where(_ => hasInitialized);

lock (locker)
{
return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache);
var observerSubscription = leftLoader.Merge(rightLoader).SubscribeSafe(observer);

hasInitialized = true;

return new CompositeDisposable(observerSubscription, leftCache, rightCache, leftShare.Connect());
}
});
}
84 changes: 48 additions & 36 deletions src/DynamicData/Cache/Internal/RightJoin.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ public IObservable<IChangeSet<TDestination, TRightKey>> Run() => Observable.Crea
// joined is the final cache
var joinedCache = new ChangeAwareCache<TDestination, TRightKey>();

var hasInitialized = false;

var rightLoader = rightCache.Connect().Select(changes =>
{
foreach (var change in changes.ToConcreteType())
Expand Down Expand Up @@ -70,48 +72,58 @@ public IObservable<IChangeSet<TDestination, TRightKey>> Run() => Observable.Crea
return joinedCache.CaptureChanges();
});

var leftLoader = leftCache.Connect().Select(changes =>
{
foreach (var change in changes.ToConcreteType())
var leftLoader = leftCache.Connect()
.Select(changes =>
{
var left = change.Current;
var right = rightGrouped.Lookup(change.Key);

if (right.HasValue)
foreach (var change in changes.ToConcreteType())
{
switch (change.Reason)
var left = change.Current;
var right = rightGrouped.Lookup(change.Key);

if (right.HasValue)
{
case ChangeReason.Add:
case ChangeReason.Update:
foreach (var keyvalue in right.Value.KeyValues)
{
joinedCache.AddOrUpdate(_resultSelector(keyvalue.Key, left, keyvalue.Value), keyvalue.Key);
}

break;

case ChangeReason.Remove:
foreach (var keyvalue in right.Value.KeyValues)
{
joinedCache.AddOrUpdate(_resultSelector(keyvalue.Key, Optional<TLeft>.None, keyvalue.Value), keyvalue.Key);
}

break;

case ChangeReason.Refresh:
foreach (var key in right.Value.Keys)
{
joinedCache.Refresh(key);
}

break;
switch (change.Reason)
{
case ChangeReason.Add:
case ChangeReason.Update:
foreach (var keyvalue in right.Value.KeyValues)
{
joinedCache.AddOrUpdate(_resultSelector(keyvalue.Key, left, keyvalue.Value), keyvalue.Key);
}

break;

case ChangeReason.Remove:
foreach (var keyvalue in right.Value.KeyValues)
{
joinedCache.AddOrUpdate(_resultSelector(keyvalue.Key, Optional<TLeft>.None, keyvalue.Value), keyvalue.Key);
}

break;

case ChangeReason.Refresh:
foreach (var key in right.Value.Keys)
{
joinedCache.Refresh(key);
}

break;
}
}
}
}

return joinedCache.CaptureChanges();
});
return joinedCache.CaptureChanges();
})
// Don't forward initial changesets from the left side, only the right
.Where(_ => hasInitialized);

lock (locker)
{
var observerSubscription = leftLoader.Merge(rightLoader).SubscribeSafe(observer);

hasInitialized = true;

return new CompositeDisposable(leftLoader.Merge(rightLoader).SubscribeSafe(observer), leftCache, rightCache, rightShare.Connect());
return new CompositeDisposable(observerSubscription, leftCache, rightCache, rightShare.Connect());
}
});
}