RolandPheasant

//Change the SubscribeMany line
var pivotLoader = _pivotTables.Connect()
.SubscribeMany(pivot => pivot.Data.Connect().Synchronize(_updateLocker).Subscribe(changes => PivotDataHasUpdated(pivot,changes)))
.Subscribe();

//Change the PivotDataHasUpdated signature
public void PivotDataHasUpdated(IDataSource pivotTable, IChangeSet<Record, Guid> changes)
{
  //...
}

RolandPheasant

    public class MyDispatcherScheduler: IScheduler
    {
        private readonly DispatcherScheduler _dispatcherScheduler;
        private readonly ImmediateScheduler _immediateScheduler = ImmediateScheduler.Instance;

        public MyDispatcherScheduler(DispatcherScheduler original)
        {
            _dispatcherScheduler = original;
        }

RolandPheasant

    public static class DynamicDataExtensions
    {
        public static IObservable<IChangeSet<TObj>> FilterOnProperty<TObj, TProp>(
            this IObservable<IChangeSet<TObj>> source,
            Expression<Func<TObj, TProp>> selectProp,
            Func<TObj, bool> predicate) where TObj : INotifyPropertyChanged
        {

            return Observable.Create<IChangeSet<TObj>>(observer =>
            {

RolandPheasant

[Test]
public async Task TestFilterOnProperty()
{
    var listA = new SourceList<X>();
    var listB = new SourceList<X>();

    using (IObservableList<X> list = listA.Connect()
					.Or(listB.Connect())
					.AsObservableList())
    {

RolandPheasant

    public static class DynamicDataExtensions
    {
        public static IObservable<IChangeSet<TObj, TKey>> FilterOnProperty<TObj, TKey, TProp>(this IObservable<IChangeSet<TObj, TKey>> source,
            Expression<Func<TObj, TProp>> selectProp,
            Func<TObj, bool> predicate) where TObj : INotifyPropertyChanged
        {

            return Observable.Create<IChangeSet<TObj, TKey>>(observer =>
            {
                //share the connection, otherwise the entire observable chain is duplicated 

RolandPheasant

namespace DynamicData.Samplz.Examples
{
    public class CombineUsingOrViewModel
    {
        public CombineUsingOrViewModel()
        {
            //maintain databaseDevices + databaseDevices anytime
            var databaseDevices = new SourceCache<IDevice,int>(device => device.Id);
            var localDevices = new SourceCache<IDevice, int>(device => device.Id);
    

RolandPheasant

    public static class MyDynamicDataExtensions
    {
        public static IObservable<IChangeSet<WidgetListItemViewModel, int>> TransformWithupdate(this IObservable<IChangeSet<Widget, int>> source )
        {
            return Observable.Create<IChangeSet<WidgetListItemViewModel, int>>(observer =>
            {
                //create local data source wich tansforms all items
                var localDataSource = source
                    .Transform(w => new WidgetListItemViewModel(w))
                    .AsObservableCache();

RolandPheasant

        public static IObservable<IChangeSet<WidgetListItemViewModel, int>> TransformWithupdate(this IObservable<IChangeSet<Widget, int>> source )
        {
            return Observable.Create<IChangeSet<WidgetListItemViewModel, int>>(observer =>
            {
                var shared = source
                    .Transform(w => new WidgetListItemViewModel(w))
                    .Publish();

                //create cache which never replaces the original problem
                var nonUpdatingObservableCache = shared

RolandPheasant

        /// <summary>
        /// Transforms the without updates. Blah Blah
        /// </summary>
        /// <typeparam name="TObject">The type of the object.</typeparam>
        /// <typeparam name="TKey">The type of the key.</typeparam>
        /// <typeparam name="TDestination">The type of the destination.</typeparam>
        /// <param name="source">The source.</param>
        /// <param name="factory">The factory.</param>
        /// <param name="updateAction">Apply changes to the original. Example (original, newitem) => original.Value = newitem.Value </param>
        /// <returns></returns>

RolandPheasant

public static clas DynamicDataJoinEx
{
	        /// <summary>
        /// Joins the left and right observable data sources, combining the content into a single
        /// </summary>
        /// <typeparam name="TLeft">The object type of the left datasource</typeparam>
        /// <typeparam name="TLeftKey">The key type of the left datasource</typeparam>
        /// <typeparam name="TRight">The object type of the right datasource</typeparam>
        /// <typeparam name="TRightKey">The key type of the right datasource</typeparam>
        /// <typeparam name="TDestination">The resulting object which </typeparam>