将 IDictionary<T, Task<bool>> 转换为 IObservable<KeyValuePair<T, bool>>

Question

我有一个功能可以用另一个实体替换存储库（例如数据库）中的单个实体。此 return 是一个任务，其结果指示替换是否成功。

// Single Replace
Task<bool> Replace(Entity e, Entity Other);

现在假设我想创建相同的函数，但允许替换多个实体。我对函数签名的想法是：

// Multiple Replace
IObservable<KeyValuePair<Entity, bool>> Replace(IDictionary<Entity, Entity> Map);

Map 包含一个字典，其键是要替换的实体，其值是要替换旧实体的新实体。

这个Map字典的每个KeyValuePair应该return一个KeyValuePair< Entity, bool >，其中key对应Map.Key（=要被替换的实体），Value是一个bool表示替换成功与否

我想 return 这作为一个流，因此我选择了 IObservable< KeyValuePair< Entity,bool >>，为每个可用的结果推出一个 KeyValuePair< Entity,bool >。

目前，我只想让"Multiple Replace"函数使用"Single Replace"函数进行转换；但是我怎样才能进行这些调用并 return 所需格式的结果？

Task<bool> Replace(Entity e, Entity Other)
{
  // ... do the work ...
}

IObservable<KeyValuePair<Entity, bool>> Replace(IDictionary<Entity, Entity> Map)
{
  // How this should work:
  // for each KeyValuePair in Map, call the Single Replace function
  // and - once available - return its result (bool) mapped to the Entity in the
  // resulting Observable stream

  IDictionary<Entity, Task<bool>> dict = Map.ToDictionary(x => x.Key, x => Replace(x.Key, x.Value));

  // .. now what? How can I now convert the dict into an IObservable<KeyValuePair<Entity,bool>> ?

}

不胜感激！

Answer 1

如果没有看到一个很好的 Minimal, Complete, and Verifiable example 清楚地表明您在做什么，尤其是您打算如何实现从您的方法返回的 IObservable<T> 对象，则不可能提供具体的建议。不过，我可以提供一些建议，至少应该为您指明一个有用的方向。

首先，我建议使用 Task<Tuple<Entity, bool>>（或 Task<KeyValuePair<Entity, bool>> 更有用，尽管我不太喜欢在非字典相关代码中使用字典相关类型…也就是说，你实际上只是在配对东西，而不是真正的 key/value 配对）。这样，当任务完成时，您可以从任务本身了解您需要的所有信息。

这可以通过一个简单的辅助方法来完成，例如：

async Task<Tuple<Entity, bool>> ReplaceAsync(Entity e, Entity other)
{
    return Tuple.Create(e, await Replace(e, other));
}

我在这里使用了 Tuple，当然您可以为此目的创建一个命名类型，这样做实际上可能有助于提高代码的可读性。

完成后，您可以枚举输入字典（或其他更合适的集合 :) ）中的所有项目，然后使用 Task.WhenAny() 迭代删除并发布每个已完成的项目发生：

IObservable<Tuple<Entity, bool>> Replace(IDictionary<Entity, Entity> Map)
{
    Task<Tuple<Entity, bool>>[] tasks = new Task<Tuple<Entity, bool>>(Map.Count);
    int i = 0;

    foreach (var kvp in Map)
    {
        tasks[i++] = ReplaceAsync(kvp.Key, kvp.Value);
    }

    // Create the IObservable object somehow. Make sure it has the
    // array of tasks in it. E.g. (see below for TaskObservable example)

    TaskObservable<Tuple<Entity, bool>> observable =
        new TaskObservable<Tuple<Entity, bool>>(tasks);

    // Now, start running the observable object. Note: not really all that great
    // to ignore the returned Task object but without more context in your question
    // I can't offer anything more specific. You will probably want to store the
    // Task object *somewhere*, await it, wrap all that in try/catch, etc. to
    // make sure you are correctly monitoring the progress of the task.
    var _ = observable.Run();

    return observable;
}

IObservable<T> 可能类似于：

class TaskObservable<T> : IObservable<T>
{
    private class ObserverItem : IDisposable
    {
        public readonly IObserver<T> Observer;
        public readonly TaskObservable<T> Owner;

        public ObserverItem(IObserver<T> observer, TaskObservable<T> owner)
        {
            Observer = observer;
            Owner = owner;
        }

        public void Dispose()
        {
            if (!Owner._observerItems.Contains(this))
            {
                throw new InvalidOperationException(
                    "This observer is no longer subscribed");
            }

            Owner._observerItems.Remove(this);
        }
    }

    private readonly Task<T>[] _tasks;
    private readonly List<ObserverItem> _observerItems = new List<ObserverItem>();

    public TaskObservable(Task<T>[] tasks)
    {
        _tasks = tasks;
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        ObserverItem item = new ObserverItem(observer, this);

        _observerItems.Add(item);
        return item;
    }

    private void Publish(T t)
    {
        foreach (ObserverItem item in _observerItems)
        {
            item.Observer.OnNext(t);
        }
    }

    private void Complete()
    {
        foreach (ObserverItem item in _observerItems)
        {
            item.Observer.OnCompleted();
        }
    }

    async Task Run()
    {
        for (int i = 0; i < _tasks.Length; i++)
        {
            Task<T> completedTask = await Task.WhenAny(tasks);

            Publish(completedTask.Result);
        }

        Complete();
    }
}

Answer 2

基于 Peter 的解决方案，利用 Task.ToObservable 的一点修改：

    IObservable<Tuple<Entity, bool>> Replace(IDictionary<Entity, Entity> Map)
    {
        Task<Tuple<Entity, bool>>[] tasks = new Task<Tuple<Entity, bool>>(Map.Count);
        int i = 0;

        foreach (var kvp in Map)
        {
            tasks[i++] = ReplaceAsync(kvp.Key, kvp.Value);
        }

        return tasks
            .Select(x => x.ToObservable())
            .Merge();   // or use .Concat for sequential execution
    }

似乎工作正常，有人发现任何潜在问题吗？

使用 .Merge 提供一个并行启动任务的 IObservable，在任务完成时提供结果，并在所有任务完成时完成。

或者我可以使用 .Concat() 顺序执行它们，即先启动第一个任务，等待它完成，然后启动第二个任务，等等。

我相信错误处理是免费的，所以每当任何任务抛出异常时，都会自动调用 IObservable 的 OnError。

Answer 3

我不确定我是否遗漏了什么，但我认为这可能对你有用：

public IObservable<KeyValuePair<T, bool>> Convert<T>(IDictionary<T, Task<bool>> source)
{
    return
        Observable
            .Create<KeyValuePair<T, bool>>(o =>
                (
                    from s in source.ToObservable()
                    from v in s.Value.ToObservable()
                    select new KeyValuePair<T, bool>(s.Key, v)
                ).Subscribe(o));
}

将 IDictionary<T, Task<bool>> 转换为 IObservable<KeyValuePair<T, bool>>

Convert IDictionary<T, Task<bool>> to IObservable<KeyValuePair<T, bool>>

c#

asynchronous

system.reactive