RX PropertyChanged GroupBy 死锁

Question

我正在尝试使用响应式扩展来限制 PropertyChanged 通知。有使用 GroupBy 执行此操作的示例，但为每个 PropertyName 创建一个订阅。

我想为所有属性处理 PropertyChanged 事件，并且我需要为每个 PropertyName 限制这些事件。

这是我目前所拥有的，但它会导致死锁。

ValuesPropertyChanged = Observable.FromEventPattern<PropertyChangedEventArgs>(value, "PropertyChanged")
    .GroupBy(o => o.EventArgs.PropertyName)
    .First()
    .Throttle(TimeSpan.FromSeconds(2))
    .Subscribe(args => HandlePropertyChanged(args.EventArgs.PropertyName));

死锁发生在对 .First() 的调用中。

如果我将该行更改为：

.Select(o => o.First())

我也试过了

.Select(o => o.FirstAsync())

GroupBy here 的示例看起来非常简洁，但我无法将这些示例转换为我的解决方案。

为什么这会导致死锁，我应该怎么做才能使它工作？

Answer 1

我认为这可能就是你所追求的：

// assume MyObj : INotifyPropertyChanged, naturally
var value = new MyObj();

Action<string> HandlePropertyChanged = 
    name => Console.WriteLine("Got a change for name:" + name);

// The query
var valuesPropertyChanged = 
    // create from event stream
    from propChange in Observable.FromEventPattern<PropertyChangedEventArgs>(
             value, 
             "PropertyChanged")            
    // group events by property name
    group propChange by propChange.EventArgs.PropertyName into batchByName
    // Throttle the resulting batch
    from throttledByName in batchByName.Throttle(TimeSpan.FromSeconds(1))
    // then select each item of the "throttled output"
    select throttledByName;

valuesPropertyChanged.Subscribe(args => 
    HandlePropertyChanged(args.EventArgs.PropertyName));

for(int i=0;i<10;i++) 
{ 
    value.Value1 = i.ToString(); 
    value.Value2 = (i-1).ToString();
}

输出：

Got a change for name:Value2
Got a change for name:Value1

---

这里是一样的，但有扩展方法：

var valuesPropertyChanged =
    Observable.FromEventPattern<PropertyChangedEventArgs>(
                _vm,
                "PropertyChanged")
                .GroupBy(propchange => propchange.EventArgs.PropertyName)
                .Select(o => o.Throttle(TimeSpan.FromSeconds(1)))
                .Merge();