收听列表中项目内的可观察事件

Question

我有IObservable<ImmutableArray<T>>，其中每个T都有相应的Delete命令，IObservable<Unit>。我试图在用户单击列表中的项目上的删除时作出响应。当一个项目被添加到列表中时，我想要开始监听（订阅）Delete命令。当一个项目从列表中删除时，我想停止收听（取消订阅）。如果一个项目X被添加到列表中，并且列表多次更改，我想确保我只在X上订阅了删除命令 - 当它被添加时 - 而不是每次列表更改时退订和重新订阅。

本来我试图使用Switch命令来做到这一点。但后来我意识到，每次列表更改时，它可能会取消订阅并重新订阅每个项目。这使得使用Pairwise或Scan很困难，因为每次列表更改时，我的订阅将被清除，并且我将从新的订阅开始。我想我遇到了其他问题，但它通常不是所有订阅者和退订的正确答案。

所以我认为答案涉及到使用TakeUntil。我会监视列表，在它上面做一个Pairwise，并且能够始终知道什么是新建和删除。然后，当我订阅每件商品时，我会执行一个TakeUntil该商品位于已移除集合中。那是这个想法，但我在代码中遇到了麻烦。

这就是我的工作：

interface IListItem { 
    IObservable<Unit> Delete { get; } 
    string ListItemName { get; } 
} 

IObservable<ImmutableList<IListItem>> _list; 

_list...something here...Subscribe(i=>{ 
    Console.WriteLine($"You requested to delete {i}!"); 
}); 

Answer 1

嗯，我想我有一个像样的答案在这里。稍微长一点，但我认为它的工作原理。如果有人想提出更短或更简单的建议，我很乐意听到它。部分.Select(add => add.Delete.Select(_ => add).TakeUntil...很奇怪。基本上，当为特定列表项调用delete命令时，我想返回删除命令被调用的列表项。

_items 
.Pairwise((before, after) => new 
{ 
    AddedToList = after.Except(before), 
    RemovedFromList = before.Except(after) 
}) 
.Publish(p => 
{ 
    var additions = p.SelectMany(i => i.AddedToList); 
    var removals = p.SelectMany(i => i.RemovedFromList); 
    return 
     additions 
     .Select(add => 
      add 
      .Delete 
      .Select(_ => add) 
      .TakeUntil(removals.Where(rem => rem == add))) 
     .Merge(); 
}).Subscribe(i => 
{ 
    // process the delete request on i 
    // at the end, submit the modified array to _items 
}); 

Answer 2

这是我最终建立的。它对我来说工作得很好。第一部分是计算和报告两组之间差异的通用类。

public class SetComparison<T> 
{ 
    private Lazy<IImmutableSet<T>> _added; 
    private Lazy<IImmutableSet<T>> _removed; 
    private Lazy<IImmutableSet<T>> _intersection; 

    public SetComparison(IEnumerable<T> previous, IEnumerable<T> current) 
    { 
     if (previous == null) throw new ArgumentNullException(nameof(previous)); 
     if (current == null) throw new ArgumentNullException(nameof(current)); 
     Previous = previous.ToImmutableHashSet(); 
     Current = current.ToImmutableHashSet(); 
     _added = new Lazy<IImmutableSet<T>>(() => Current.Except(Previous)); 
     _removed = new Lazy<IImmutableSet<T>>(() => Previous.Except(Current)); 
     _intersection = new Lazy<IImmutableSet<T>>(() => Current.Intersect(Previous)); 
    } 

    public IImmutableSet<T> Previous { get; } 
    public IImmutableSet<T> Current { get; } 
    public IImmutableSet<T> Added => _added.Value; 
    public IImmutableSet<T> Removed => _removed.Value; 
    public IImmutableSet<T> Intersection => _intersection.Value; 
} 

这是一个通用的运营商，如F＃成对采取序列，并把它变成一系列重叠的项目配对的。

public static IObservable<TResult> Pairwise<TSource, TResult>(
    this IObservable<TSource> source, 
    Func<TSource, TSource, TResult> resultSelector) => 
    source.Scan(
     (default(TSource), default(TSource)), 
     (pair, current) => (pair.Item2, current)) 
     .Skip(1) 
     .Select(p => resultSelector(p.Item1, p.Item2)); 

而这现在成对的两组。

public static IObservable<SetComparison<T>> PairwiseSetComparison<T, TCollection>(this IObservable<TCollection> source) where TCollection : IEnumerable<T> => 
    source 
    .Pairwise((a, b) => new SetComparison<T>(a, b)); 


public static IObservable<SetComparison<T>> PairwiseSetComparison<T>(this IObservable<ImmutableArray<T>> source) => 
    source.Pairwise((a, b) => new SetComparison<T>(a, b)); 

最后这一点，

public static IObservable<TResult> Merge<TResult, T>(this IObservable<SetComparison<T>> source, Func<T, IObservable<TResult>> result) => 
    source 
    .Publish(s => 
    { 
     var additions = s.SelectMany(i => i.Added); 
     var removals = s.SelectMany(i => i.Removed); 
     return 
      additions 
      .Select(add => result(add).TakeUntil(removals.Where(rem => rem.Equals(add)))) 
      .Merge(); 
    }); 

所以现在当我想订阅了仅在当前设置，我做这样的事情观测...

_items 
.PairwiseSetComparison() 
.Merge(i => i.Delete.Select(_ => i)) 
.Subscribe(i=> /* user clicked Delete on item i */) 

完成这一切的另一种方法是在用户调用它时，让列表中的每个项目上的Delete ICommand实际运行一些委托。这样，我不必在命令被调用时实际订阅。尽管如此，我更愿意将我的清单中的物品完全不知道他们的周围环境。