Rx.Net : 在 SelectMany 中调用多个 IObservable

Question

请注意：这是之前发布的的延续，但感兴趣的解决方案是不同的情况。

我正在尝试对方法进行多次调用，每个 return IObservable 但 值在 SelectMany 语句中被 returned 返回是一个任务，因此如下订阅语句不编译。

这是代码片段

 var myWorkList = new List<MyWork>
                {
                    new MyWork(),// MyWork.Execute(data) returns IObservable
                    new MyWork()
                }.ToObservable();

 var results =
   myService
    .GetData(accountId)
    .SelectMany(data => myWorkList.ForEachAsync(r => r.Execute(data))
    .Subscribe(result =>
    {
        Console.WriteLine($"Result Id: {result.Id}");
        Console.WriteLine($"Result Status: {result.Pass}");
    });

Answer 1

您只想使用 .SelectMany。试试这个：

var myWorkList = new List<MyWork>()
{
    new MyWork(),
    new MyWork()
}.ToObservable();

var query =
    from data in myService.GetData(accountId)
    from myWork in myWorkList
    from result in myWork.Execute(data)
    select result;

var results =
    query
        .Subscribe(result =>
        {
            Console.WriteLine($"Result Id: {result.Id}");
            Console.WriteLine($"Result Status: {result.Pass}");
        });

这是我的测试代码：

public static class myService
{
    public static IObservable<MyData> GetData(int x)
        => Observable.Return(new MyData());
}

public class MyWork
{
    public virtual IObservable<MyResult> Execute(MyData data)
    {
        return
            from isMatch in IsMatch(data)
            where isMatch
            select new MyResult() { Id = 1, Pass = true };
    }

    public IObservable<bool> IsMatch(MyData data)
    {
        return Observable.Return(true);
    }
}

public class MyResult
{
    public int Id;
    public bool Pass;
}

public class MyData { }

当我执行时，我得到这个：

Result Id: 1
Result Status: True
Result Id: 1
Result Status: True

在对您上一个问题的评论中，我建议将此作为代表列表。方法如下：

var myWorkList = new Func<MyData, IObservable<MyResult>>[]
{
    md => new MyWork().Execute(md),
    md => new MyWork().Execute(md),
}.ToObservable();

var query =
    from data in myService.GetData(accountId)
    from myWork in myWorkList
    from result in myWork(data)
    select result;

你得到相同的结果。

Answer 2

该列表可以像 MyWork-s 的列表一样声明 - 此处不使用 ToObservable。

var myWorkList = new List<MyWork>
            {
                new MyWork(),// MyWork.Execute(data) returns IObservable
                new MyWork()
            };

然后，我们将 myService.GetData 返回的对象映射到 myWorkList 的元素上，并将它们作为 IObservable-s。

var observables = myService
         .GetData(accountId)
         .SelectMany(data => myWorkList.Select(r => r.Execute(data)));

现在您可以观察它们了。

要么在一起-合并：

var subscription =
          observables
           .Merge()
           .Subscribe(result =>
           {
               ...
           });

或分开：

var subscriptions=
          observables
           .Select(obs => 
               obs.Subscribe(result =>
               {
                   ...
               }))
           .ToArray();

更新：后一种情况必须立即具体化以防止副作用 (.ToArray())。

Rx.Net : 在 SelectMany 中调用多个 IObservable

Rx.Net : Calling multiple IObservable in SelectMany

system.reactive

rx.net