| | | 1 | | // ----------------------------------------------------------------------- |
| | | 2 | | // <copyright file="MergeManyEx.cs" company="Stéphane ANDRE"> |
| | | 3 | | // Copyright (c) Stéphane ANDRE. All rights reserved. |
| | | 4 | | // </copyright> |
| | | 5 | | // ----------------------------------------------------------------------- |
| | | 6 | | |
| | | 7 | | using System; |
| | | 8 | | using System.Linq; |
| | | 9 | | using System.Reactive.Linq; |
| | | 10 | | using DynamicData; |
| | | 11 | | |
| | | 12 | | #pragma warning disable IDE0130 // Namespace does not match folder structure |
| | | 13 | | namespace MyNet.Observable; |
| | | 14 | | #pragma warning restore IDE0130 // Namespace does not match folder structure |
| | | 15 | | |
| | | 16 | | /// <summary> |
| | | 17 | | /// Forwards inner change sets when outer items are added and synthesizes remove change sets when outer items are remove |
| | | 18 | | /// </summary> |
| | | 19 | | /// <remarks> |
| | | 20 | | /// <paramref name="observableSelector"/> is invoked again on outer remove; it must return the same inner snapshot |
| | | 21 | | /// for the same outer item (no side effects, stable child collection). |
| | | 22 | | /// </remarks> |
| | | 23 | | internal sealed class MergeManyEx<T, TDestination>(IObservable<IChangeSet<T>> source, |
| | | 24 | | Func<T, IObservable<IChangeSet<TDestination>>> observableSelector) |
| | | 25 | | where T : notnull |
| | | 26 | | where TDestination : notnull |
| | | 27 | | { |
| | 6 | 28 | | private readonly IObservable<IChangeSet<T>> _source = source ?? throw new ArgumentNullException(nameof(source)); |
| | 6 | 29 | | private readonly Func<T, IObservable<IChangeSet<TDestination>>> _observableSelector = observableSelector ?? throw ne |
| | | 30 | | |
| | 6 | 31 | | public IObservable<IChangeSet<TDestination>> Run() => System.Reactive.Linq.Observable.Create<IChangeSet<TDestination |
| | 6 | 32 | | observer => |
| | 6 | 33 | | { |
| | 6 | 34 | | var locker = new object(); |
| | 6 | 35 | | return _source |
| | 6 | 36 | | |
| | 6 | 37 | | // SubscribeMany will forward initial cascaded inner items when it observers new item, |
| | 6 | 38 | | // but if one item was removed, any item belonged to it won't be forwarded to observer. |
| | 6 | 39 | | .SubscribeMany(t => _observableSelector(t).Synchronize(locker).Subscribe(observer.OnNext)) |
| | 6 | 40 | | |
| | 6 | 41 | | // So I add an observer here to subscribe all items belonged to the removed item, buildup a |
| | 6 | 42 | | // ChangeSet and forward it to the original observer. |
| | 6 | 43 | | .Subscribe(t => |
| | 6 | 44 | | { |
| | 6 | 45 | | foreach (var item in t.GetRemovedItems()) ForwardWhenRemove(observer, item); |
| | 6 | 46 | | }, |
| | 6 | 47 | | observer.OnError); |
| | 6 | 48 | | }); |
| | | 49 | | |
| | | 50 | | private void ForwardWhenRemove(IObserver<IChangeSet<TDestination>> observer, T sourceItem) |
| | | 51 | | { |
| | 6 | 52 | | var observableList = _observableSelector(sourceItem).AsObservableList(); |
| | 6 | 53 | | var changeset = new ChangeSet<TDestination>( |
| | 6 | 54 | | [ |
| | 6 | 55 | | new(ListChangeReason.RemoveRange, observableList.Items) |
| | 6 | 56 | | ]); |
| | 6 | 57 | | observableList.Dispose(); |
| | 6 | 58 | | observer.OnNext(changeset); |
| | 6 | 59 | | } |
| | | 60 | | } |
| | | 61 | | |
| | | 62 | | /// <inheritdoc cref="MergeManyEx{T, TDestination}"/> |
| | | 63 | | internal sealed class MergeManyEx<T, TKey, TDestination, TDestinationKey>(IObservable<IChangeSet<T, TKey>> source, |
| | | 64 | | Func<T, IObservable<IChangeSet<TDestination, TDestinationKey>>> observableSelector, |
| | | 65 | | Func<TDestination, TDestinationKey> observableKeySelector) |
| | | 66 | | where T : notnull |
| | | 67 | | where TDestination : notnull |
| | | 68 | | where TKey : notnull |
| | | 69 | | where TDestinationKey : notnull |
| | | 70 | | { |
| | | 71 | | private readonly IObservable<IChangeSet<T, TKey>> _source = source ?? throw new ArgumentNullException(nameof(source) |
| | | 72 | | private readonly Func<T, IObservable<IChangeSet<TDestination, TDestinationKey>>> _observableSelector = observableSel |
| | | 73 | | private readonly Func<TDestination, TDestinationKey> _observableKeySelector = observableKeySelector ?? throw new Arg |
| | | 74 | | |
| | | 75 | | public IObservable<IChangeSet<TDestination, TDestinationKey>> Run() => System.Reactive.Linq.Observable.Create<IChang |
| | | 76 | | observer => |
| | | 77 | | { |
| | | 78 | | var locker = new object(); |
| | | 79 | | return _source |
| | | 80 | | |
| | | 81 | | // SubscribeMany will forward initial cascaded inner items when it observers new item, |
| | | 82 | | // but if one item was removed, any item belonged to it won't be forwarded to observer. |
| | | 83 | | .SubscribeMany(t => _observableSelector(t).Synchronize(locker).Subscribe(observer.OnNext)) |
| | | 84 | | |
| | | 85 | | // So I add an observer here to subscribe all items belonged to the removed item, buildup a |
| | | 86 | | // ChangeSet and forward it to the original observer. |
| | | 87 | | .Subscribe(t => |
| | | 88 | | { |
| | | 89 | | foreach (var item in t.GetRemovedItems()) ForwardWhenRemove(observer, item); |
| | | 90 | | }, |
| | | 91 | | observer.OnError); |
| | | 92 | | }); |
| | | 93 | | |
| | | 94 | | private void ForwardWhenRemove(IObserver<IChangeSet<TDestination, TDestinationKey>> observer, T sourceItem) |
| | | 95 | | { |
| | | 96 | | var observableList = _observableSelector(sourceItem).AsObservableCache(); |
| | | 97 | | var changeset = new ChangeSet<TDestination, TDestinationKey>(); |
| | | 98 | | changeset.AddRange(observableList.Items.Select(x => new Change<TDestination, TDestinationKey>(ChangeReason.Remov |
| | | 99 | | observableList.Dispose(); |
| | | 100 | | observer.OnNext(changeset); |
| | | 101 | | } |
| | | 102 | | } |
| | | 103 | | |