Rx:忽略订阅者引起的更新 [英] Rx: Ignoring updates caused by Subscribers

问题描述

我在弄清楚如何执行此操作时遇到问题.我有两个实例(源和目标)实现了 INotifyPropertyChanged 并且我正在跟踪两者的 PropertyChanged 事件.我想要做的是在 source.PropertyChanged 被引发直到 target.PropertyChanged 被引发时运行一个动作.我可以这样做:

I'm having problems figuring out how to do this. I have two instances (source & target) that implement INotifyPropertyChanged and I'm tracking the PropertyChanged event for both. What I want to do is run an action any time source.PropertyChanged is raised until target.PropertyChanged is raised. I can do that just fine like this:

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 事件时停止取值.有什么好的方法可以实现吗?

The problem I'm having is I want to ignore the PropertyChanged notifications caused by the actions and only stop taking values when the PropertyChanged event is raised by an external source. Is there a good way to get that to happen?

推荐答案

没有内在的方式来做你正在谈论的事情.这是一个简单的 SkipWhen 实现，每次在其他"序列上接收到一个值时，它都会跳过下一个源值:

There's no built in way of doing what you're talking about. Here's a simple SkipWhen implementation that skips the next source value each time a value is received on the 'other' sequence:

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;
                }
            }
        });
    });
}

然后你的代码会像这样更新(见我下面的注释):

You're code would then be updated like so (see my note below):

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*/
    });

注意:我最近写了一篇关于强类型INotifyPropertyChanged 事件的 IObservable 包装器；随意窃取该代码.

NB: I recently blogged about a strongly typed IObservable wrapper around INotifyPropertyChanged events; feel free to steal that code.

这篇关于Rx:忽略订阅者引起的更新的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！