RX PropertyChanged GroupBy死锁

Question

我正在尝试使用Reactive Extensions来限制PropertyChanged通知。有一些使用GroupBy进行此操作的示例，但为每个PropertyName创建一个Subscription。

我想处理所有属性的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();