How to make ReactiveProperty with ReactiveCollection?

[sericaer]

for example:

    public ReactiveProperty<int> total { get; private set; }
    public ReactiveCollection<UsedElem> listUsed = new ReactiveCollection<UsedElem>();

    public class UsedElem
    {
        public ReactiveProperty<int> tvalue;
    }

ReactiveProperty total is the sum of tvalue in UsedElem listUsed, listUsed can be insert and remove, and tvalue can also changed.

How to create ReactiveProperty total ?

[guopenglun]

tvalue=listUsed.ObserveRemove().ToReactiveProperty(); tvalue=listUsed.ObserveEveryValueChanged(_ => _.Count).ToReactiveProperty();

[Zulu-Inuoe]

@guopenglun That won't detect additions, only removals

[jmprimeau]

I've done this as RxSum You need to observe add, remove, clear, replace and have those subscribeWithStwte using the output total reactive property.

The only issue I have with this is that the output property needs to be disposable and needs to dispose it's 4 subscriptions.

[jmprimeau]

I also made the more generic RxAggregate; would be nice to just override the linq methods directly, but I'm afraid of leaking Disposables.

Thoughts on that would be appreciated.

[jmprimeau]

Was at the beach when I wrote that... Here's some code. DisposableReadOnlyReactiveProperty is basically a reactive property that carries a disposable that will kill all those subscriptions; Would be happy to see a better alternative.

Also, it seems that SubscribeWithState, passing in aggregated would be better.

public static IDisposableReadOnlyReactiveProperty<TR> RxAggregate<T, TR>( this IReadOnlyReactiveCollection<T> source, TR seed, Func<TR, T, TR> doIt, Func<TR, T, TR> undoIt) { var aggregated = new ReactiveProperty<TR>(source.Aggregate(seed, doIt)); var disposable = new CompositeDisposable(); source.ObserveRemove().Subscribe(args => aggregated.Value = undoIt(aggregated.Value, args.Value)).AddTo(disposable); source.ObserveAdd().Subscribe(args => aggregated.Value = doIt(aggregated.Value, args.Value)).AddTo(disposable); source.ObserveReplace().Subscribe(args => aggregated.Value = doIt(undoIt(aggregated.Value, args.OldValue), args.NewValue)).AddTo(disposable); source.ObserveReset().Subscribe(args => aggregated.Value = seed).AddTo(disposable); return new DisposableReadOnlyReactiveProperty<TR>(aggregated, disposable); }

public static IDisposableReadOnlyReactiveProperty<long> RxSum(this IReadOnlyReactiveCollection<long> source) => source.RxAggregate(0L, (a, b) => a + b, (a, b) => a - b);
public static IDisposableReadOnlyReactiveProperty<int> RxSum(this IReadOnlyReactiveCollection<int> source) => source.RxAggregate(0, (a, b) => a + b, (a, b) => a - b);
public static IDisposableReadOnlyReactiveProperty<float> RxSum(this IReadOnlyReactiveCollection<float> source) => source.RxAggregate(0f, (a, b) => a + b, (a, b) => a - b);
public static IDisposableReadOnlyReactiveProperty<double> RxSum(this IReadOnlyReactiveCollection<double> source) => source.RxAggregate(0d, (a, b) => a + b, (a, b) => a - b);