收聽列表中項目內的可觀察事件

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實際運行一些委託。這樣，我不必在命令被調用時實際訂閱。儘管如此，我更願意將我的清單中的物品完全不知道他們的周圍環境。