1

我有一个ObservableCollection<T>代表不断变化的集合的类：

public interface IObservableCollection<T> : IObservable<IEnumerable<T>>
{
    void Add(T item);
    void Remove(T item);
}

当添加或删除一个项目时，在内部 a会使用 new 调用Subject<IEnumerable<T>>其方法，该方法通过 的方法公开。OnNextIEnumerable<T>SubscribeIObservableCollection<T>

我也有一堂课Person：

public interface IPerson
{
    string Name { get; }
    IObservable<int> Position { get; }
}

我想要做的是产生一个IEnumerable<Tuple<string, int>>代表每个人的位置的流，一个人在集合中的位置。这似乎相对简单：

var peopleCollectionStream = new ObservableCollection<IPerson>();

var peoplePositions = from people in peopleCollectionStream
                      from updateList in
                          (from person in people
                           select person.Position.Select(pos => Tuple.Create(person.Name, pos)))
                           .CombineLatest()
                      select updateList;

我现在可以像这样订阅流：

peoplePositions
    .Subscribe(people =>
    {
        Console.WriteLine("Something was updated");
        foreach (var personTuple in people)
            Console.WriteLine("{0} -> {1}", personTuple.Item1, personTuple.Item2);
    });

我得到了想要的输出：

var alice = new Person() { Name = "Alice" };
peopleCollectionStream.Add(alice);        // Alice -> 0
alice.Move(2);                            // Alice -> 2
var bob = new Person() { Name = "Bob" };
peopleCollectionStream.Add(bob);          // Alice -> 2, Bob -> 0
bob.Move(3);                              // Alice -> 2, Bob -> 3

当我希望从集合中删除一个人并因此从流中排除他们的更新时，就会出现问题：

peopleCollectionStream.Remove(bob);       // Alice -> 2
bob.Move(4);                              // Alice -> 2, Bob -> 4

如果 Bob 从集合中删除，我想停止包含 Bob 的位置更新。我怎样才能做到这一点？

4

2 回答 2

1

我发现尝试使用 Add 和 Remove 事件是一个坏主意，如果你想做这些功能性的事情。将删除与添加相匹配，并确保底层代码也能做到这一点，是很多工作。

我所做的是使用易腐烂的物品/收藏品。我将每个项目与一个生命周期（取消令牌）配对，并且该项目在其生命周期结束时被视为已删除。然后我在连接其他东西时使用这些生命周期。我使用了一个名为 的集合类型，PerishableCollection<T>它采用与生命周期配对的项目，并允许您将其内容作为IObservable<Perishable<T>>.

我写了一篇关于易腐收藏品的博文，并发布了一个可供您参考的 nuget 库。

下面的代码应该展平易腐集合的易腐集合：

public static PerishableCollection<T> Flattened<T>(this PerishableCollection<PerishableCollection<T>> collectionOfCollections, Lifetime lifetimeOfResult) {
    if (collectionOfCollections == null) throw new ArgumentNullException("collectionOfCollections");

    var flattenedCollection = new PerishableCollection<T>();
    collectionOfCollections.CurrentAndFutureItems().Subscribe(
        c => c.Value.CurrentAndFutureItems().Subscribe(

            // OnItem: include in result, but prevent lifetimes from exceeding source's lifetime
            e => flattenedCollection.Add(
                item: e.Value,
                lifetime: e.Lifetime.Min(c.Lifetime)),

            // subscription to c ends when the c's lifetime ends or result is no longer needed
            c.Lifetime.Min(lifetimeOfResult)),

        // subscription ends when result is no longer needed
        lifetimeOfResult);

    return flattenedCollection;
}

上述工作通过订阅接收添加到集合集合中的集合，然后为每个订阅接收项目的人。这些项目被放入生成的集合中，其生命周期在项目死亡或其集合死亡时结束。当赋予该方法的生命周期终止时，所有订阅都会终止。

解决这个问题的另一种方法是编写一个方法来展平IObservable<Perishable<IObservable<Perishable<T>>>>. 这样做的好处是不需要调用者如此明确地管理结果的生命周期并适用于更多情况。但是，该方法很难编写，因为您必须以线程安全的方式处理失败/完成的序列。

这是使用 flatten 方法的示例（创建一个新的控制台应用程序，引用 perishable 集合，粘贴上述方法和这个方法）：

using TwistedOak.Collections;
using TwistedOak.Util;

static void Main() {
    var p = new PerishableCollection<PerishableCollection<string>>();
    var f = p.Flattened(Lifetime.Immortal);
    f.CurrentAndFutureItems().Subscribe(e => {
        Console.WriteLine("{0} added to flattened", e.Value);
        e.Lifetime.WhenDead(() => Console.WriteLine("{0} removed from flattened", e.Value));
    });

    // add some 'c' items to f via p
    var c = new PerishableCollection<string>();
    var cLife = new LifetimeSource();
    c.Add("candy", Lifetime.Immortal);
    p.Add(c, cLife.Lifetime);
    c.Add("cane", Lifetime.Immortal);

    // add some 'd' items to f via p
    var d = new PerishableCollection<string>();
    p.Add(d, Lifetime.Immortal);
    d.Add("door", Lifetime.Immortal);
    d.Add("dock", Lifetime.Immortal);


    // should remove c's items from f via removing c from p
    cLife.EndLifetime();
}

代码应输出：

candy added to flattened
cane added to flattened
door added to flattened
dock added to flattened
candy removed from flattened
cane removed from flattened

希望这足以让您走上一条更轻松的道路。

于 2013-07-11T11:17:44.140 回答
0

答案是.Switch运营商。通过仅选择要订阅的最新的 observables 列表，流会排除最新版本的集合中不存在的任何内容：

var peoplePositions = (from people in peopleCollectionStream
                       select
                           (from person in people
                            select person.Position
                                .Select(pos => Tuple.Create(person.Name, pos))
                            ).CombineLatest()
                       ).Switch();

（顺便说一句，如果有人在使用方括号/嵌套 linq 查询语法时对格式化有任何好的建议，请告诉我，因为上面看起来很糟糕！）

于 2013-07-11T13:11:24.993 回答