Rx：忽略订阅者引起的更新

Question

我在弄清楚如何执行此操作时遇到问题。我有两个实现INotifyPropertyChanged 的实例（源和目标），我正在跟踪两者的PropertyChanged 事件。我想要做的是在source.PropertyChanged 被提出时运行一个动作，直到target.PropertyChanged 被提出。我可以像这样很好地做到这一点：

INotifyPropertyChanged source;
INotifyPropertyChanged target;

var sourcePropertyChanged = Observable
    .FromEvent<PropertyChangedEventArgs>(source, "PropertyChanged")
    .Where(x => x.EventArgs.PropertyName == sourcePropertyName);

var targetPropertyChanged = Observable
    .FromEvent<PropertyChangedEventArgs>(target, "PropertyChanged")
    .Where(x => x.EventArgs.PropertyName == targetPropertyName);

sourcePropertyChanged
    .TakeUntil(targetPropertyChanged)
    .ObserveOnDispatcher()
    .Subscribe(_ => /*Raises target.PropertyChanged for targetPropertyName*/);

我遇到的问题是我想忽略由操作引起的PropertyChanged 通知，并且仅在外部源引发PropertyChanged 事件时才停止取值。有什么好的方法可以实现吗？

Answer 1

没有内置的方式来做你所说的事情。这是一个简单的SkipWhen 实现，每次在“其他”序列上接收到值时都会跳过下一个源值：

public static IObservable<TSource> SkipWhen(this IObservable<TSource> source, 
    IObservable<TOther> other)
{
    return Observable.Defer<TSource>(() =>
    {
        object lockObject = new object();
        Stack<TOther> skipStack = new Stack<TOther>();

        other.Subscribe(x => { lock(lockObject) { skipStack.Push(x); });

        return source.Where(_ =>
        {
            lock(lockObject);
            {
                if (skipStack.Count > 0)
                {
                    skipStack.Pop();
                    return false;
                }
                else
                {
                    return true;
                }
            }
        });
    });
}

然后您的代码将像这样更新（请参阅下面的注释）：

INotifyPropertyChanged source;
INotifyPropertyChanged target;

// See the link at the bottom of my answer
var sourcePropertyChanged = source.GetPropertyChangeValues(x => x.SourceProperty);

// Unit is Rx's "void"
var targetChangedLocally = new Subject<Unit>();

var targetPropertyChanged = target.GetPropertyChangeValues(x => x.TargetProperty)
    .SkipWhen(targetChangedLocally);

sourcePropertyChanged
    .TakeUntil(targetPropertyChanged)
    .ObserveOnDispatcher()
    .Subscribe(_ =>
    {
        targetChangedLocally.OnNext();
        /*Raises target.PropertyChanged for targetPropertyName*/
    });

注意：我最近写了一篇关于 a strongly typed IObservable wrapper around INotifyPropertyChanged events 的博客；随意窃取该代码。

Answer 2

没有内置方法，但您可以使用 Where 可观察的扩展方法过滤掉事件。过滤条件是事件的发送者。我想target.PropertyChanged 事件的发送者与另一个来源引发的PropertyChanged 事件的发送者不同。

我不完全确定这是否是您可以使用的方法。

Answer 3

以这种方式在 Rx 中使用锁很好。锁是短暂的，不会调用用户代码。