c# - 来自 *Async 方法的 IObservable 序列

Question

我有一个源元素列表 Foo , 和一个异步方法 GetBoosAsync()将它们转换为 Boo 类型的元素:

public class Foo
{
    public string Value { get; set; }
}

public class Boo
{
    public string Value { get; set; }
}

static void Main(string[] args)
{
    IReadOnlyCollection<Foo> foos = GetFoos();

    Task<IReadOnlyCollection<Boo>> boosTask = GetBoosAsync(foos);

    boosTask.Wait();

    foreach (Boo boo in boosTask.Result)
    {
        Console.WriteLine(boo.Value);
    }

    Console.ReadKey();
}

public static IReadOnlyCollection<Foo> GetFoos()
{
    return Enumerable.Range(1, 100).Select(i => new Foo
    {
        Value = i.ToString(CultureInfo.CurrentCulture)

    }).ToList();
}

public static async Task<IReadOnlyCollection<Boo>> GetBoosAsync(IReadOnlyCollection<Foo> foos)
{
    List<Task<Boo>> booTasks = foos.Select(ConvertFooToBooAsync).ToList();

    // Waiting for ALL conversions of Foos to Boos
    await Task.WhenAll(booTasks.ToArray<Task>());

    return booTasks.Select(booTask => booTask.Result).ToList();
}

public static Task<Boo> ConvertFooToBooAsync(Foo foo)
{
    return Task.Factory.StartNew(() =>
    {
        Thread.Sleep(100);
        return new Boo { Value = foo.Value };
    });
}

方法GetBoosAsync()电话 ConvertFoosToBoosAsync对于每个元素，等待所有转换完成，然后返回结果列表。

问题:

如何实现方法GetBoosAsync返回 IObservable<Boo>相反，并在处理单个元素时异步返回它们？

即

public static IObservable<Boo> OGetBoos(IReadOnlyCollection<Foo> foos)
{
    ...
}

并像这样使用它:

static void Main(string[] args)
{
    IReadOnlyCollection<Foo> foos = GetFoos();

    IObservable<Boo> boos = OGetBoos(foos);

    boos = boos.Do(boo =>
    {
        Console.WriteLine(boo.Value);
    });

    boos.Wait();

    Console.ReadKey();
}

Answer 1

这对你有用吗:

public static IObservable<Boo> OGetBoos(IReadOnlyCollection<Foo> foos)
{
    var query =
        from f in foos.ToObservable()
        from b in ConvertFooToBooAsync(f).ToObservable()
        select b;

    return query;
}

顺便说一句，感谢您轻松编译问题中的工作代码!