FilterOnPropertyEx

    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 
                var shared = source.Publish();

                //do not filter on initial value otherwise every object loaded will invoke a requery
                var changed = shared.WhenPropertyChanged(selectProp, false)
                    .Select(v => v.Sender);

                //start with predicate to ensure filter on loaded
                var changedMatchFunc = changed.Select(_ => predicate).StartWith(predicate);

                // filter all in source, based on match funcs that update on prop change
                var changedAndMatching = shared.Filter(changedMatchFunc);

                var publisher = changedAndMatching.SubscribeSafe(observer);

                return new CompositeDisposable(publisher, shared.Connect());
            });
        }

        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 =>
            {
                //share the connection, otherwise the entire observable chain is duplicated 
                var shared = source.Publish();

                //do not filter on initial value otherwise every object loaded will invoke a requery
                var changed = shared.WhenPropertyChanged(selectProp, false)
                    .Select(v => v.Sender);

                //start with predicate to ensure filter on loaded
                var changedMatchFunc = changed.Select(_ => predicate).StartWith(predicate);

                // filter all in source, based on match funcs that update on prop change
                var changedAndMatching = shared.Filter(changedMatchFunc, FilterPolicy.CalculateDiffSet);

                var publisher = changedAndMatching.SubscribeSafe(observer);

                return new CompositeDisposable(publisher, shared.Connect());
            });
        }
    }