Rx技术问询：无需Subject创建Observable及WPF供应商主选实现

嘿，刚接触Rx就能想到用响应式对象处理WPF的数据绑定与状态管理，这点真的很棒！针对你提出的「无需Subject创建Observable」来实现仅一个供应商可设为主选的需求，我给你几个实用的落地方案：

方案1：利用INotifyPropertyChanged事件转Observable #

如果你的代理响应式对象已经实现了INotifyPropertyChanged（WPF绑定的标准接口），完全可以直接把属性变更事件转换成Observable，不需要手动创建Subject。

实现步骤： #

1. 为单个供应商的IsPrimary属性创建变更流：

// 假设SupplierProxy是你的代理类，已实现INotifyPropertyChanged
IObservable<bool> GetPrimaryChangeStream(SupplierProxy supplier)
{
    return Observable.FromEventPattern<PropertyChangedEventHandler, PropertyChangedEventArgs>(
        handler => supplier.PropertyChanged += handler,
        handler => supplier.PropertyChanged -= handler)
    .Where(e => e.EventArgs.PropertyName == nameof(SupplierProxy.IsPrimary))
    .Select(e => ((SupplierProxy)e.Sender).IsPrimary)
    .DistinctUntilChanged(); // 避免重复触发相同状态的变更
}

2. 合并所有供应商的变更流，处理单选逻辑：

// 假设suppliers是你的供应商集合
var allPrimaryChanges = suppliers.Select(supplier => 
    GetPrimaryChangeStream(supplier).Select(isPrimary => (Supplier: supplier, IsPrimary: isPrimary)));

Observable.Merge(allPrimaryChanges)
.Where(change => change.IsPrimary) // 只处理设为true的变更
.Subscribe(change => 
{
    // 将其他所有供应商的IsPrimary设为false
    foreach(var supplier in suppliers.Where(s => s != change.Supplier))
    {
        // 这里注意：如果设置属性会触发PropertyChanged，DistinctUntilChanged会帮我们过滤重复触发
        supplier.IsPrimary = false;
    }
});

方案2：使用ReactiveProperty（推荐WPF场景） #

如果你的项目可以引入ReactiveUI的ReactiveProperty（专门为.NET响应式开发设计的库），那它本身就是Observable，完全不需要Subject。ReactiveProperty天生支持WPF绑定，还能简化状态管理。

实现示例： #

1. 让代理类的IsPrimary属性改用ReactiveProperty<bool>：

public class SupplierProxy
{
    public ReactiveProperty<bool> IsPrimary { get; } = new ReactiveProperty<bool>(false);
    // 其他属性...
}

2. 合并流并处理单选逻辑：

// 合并所有供应商的IsPrimary变更流
var allPrimaryStreams = suppliers.Select(s => 
    s.IsPrimary.Select(isP => (Supplier: s, IsPrimary: isP)));

Observable.Merge(allPrimaryStreams)
.Where(x => x.IsPrimary)
.Subscribe(x => 
{
    // 批量设置其他供应商为非主选
    foreach(var supplier in suppliers.Where(s => s != x.Supplier))
    {
        // ReactiveProperty的Value属性变更会自动通知UI
        supplier.IsPrimary.Value = false;
    }
});

方案3：监听ObservableCollection的集合变更 #

如果你的供应商集合是ObservableCollection<SupplierProxy>，可以监听集合的CollectionChanged事件，自动为新增的供应商添加变更流订阅，不需要手动维护：

var suppliersCollection = new ObservableCollection<SupplierProxy>();

// 监听集合变更，动态合并/移除供应商的变更流
var dynamicPrimaryStreams = Observable.FromEventPattern<NotifyCollectionChangedEventHandler, NotifyCollectionChangedEventArgs>(
    h => suppliersCollection.CollectionChanged += h,
    h => suppliersCollection.CollectionChanged -= h)
.SelectMany(_ => 
    suppliersCollection.Select(s => 
        GetPrimaryChangeStream(s).Select(isP => (Supplier: s, IsPrimary: isP))))
.Merge();

// 同样的单选逻辑订阅
dynamicPrimaryStreams
.Where(x => x.IsPrimary)
.Subscribe(x => 
{
    foreach(var s in suppliersCollection.Where(supp => supp != x.Supplier))
    {
        s.IsPrimary = false;
    }
});

关键注意事项 #

• 一定要加上DistinctUntilChanged()：避免因为重复设置相同状态而触发无限循环的变更事件。
• 线程安全：如果你的操作涉及后台线程，记得用ObserveOnDispatcher()切换回WPF的UI线程，避免跨线程绑定异常。
• 订阅管理：如果集合会频繁变更，记得用TakeUntil或Dispose来清理订阅，防止内存泄漏。

内容的提问来源于stack exchange，提问作者Dan