IEnumerable<Task<T>> 转换为 IObservable<T>。我在这里找到了解决方案:here:
IObservable<T> ToObservable<T>(IEnumerable<Task<T>> source)
{
return source.Select(t => t.ToObservable()).Merge();
}
通常情况下这没问题,但我需要处理在这些任务中可能出现的异常... 因此,IObservable<T>
不应该在第一次异常后就结束。
根据我的阅读,对于这种情况的建议是使用一些包装器,它将携带实际值或错误。因此,我的尝试是
IObservable<Either<T, Exception>> ToObservable<T>(IEnumerable<Task<T>> source)
{
var subject = new Subject<Either<T, Exception>>();
foreach (var observable in GetIntsIEnumerable().Select(t => t.ToObservable()))
{
observable.Subscribe(i => subject.OnNext(i), e => subject.OnNext(e));
}
return subject;
}
使用从这篇文章中借鉴的Either<T, Exception>
类型。
但是这还不够好,因为OnCompleted()
没有被调用。我应该如何解决呢?我对Rx概念还比较陌生。
下面是完整的测试代码...
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Threading.Tasks;
using System.Threading;
using System.Threading.Tasks;
namespace Test
{
class Program
{
static Task Main()
{
SemaphoreSlim signal = new SemaphoreSlim(0, 1);
//GetInts1().Subscribe(
// i => Console.WriteLine($"OK: {i}"),
// e => Console.WriteLine($"ERROR: {e.Message}"),
// () => signal.Release());
GetInts2().Subscribe(r => Console.WriteLine(r.Match(
i => $"OK: {i}",
e => $"ERROR: {e.Message}")),
() => signal.Release());
return signal.WaitAsync();
}
static IObservable<int> GetInts1()
{
return GetIntsIEnumerable().Select(t => t.ToObservable()).Merge();
}
static IObservable<Either<int, Exception>> GetInts2()
{
var subject = new Subject<Either<int, Exception>>();
foreach (var observable in GetIntsIEnumerable().Select(t => t.ToObservable()))
{
observable.Subscribe(i => subject.OnNext(i), e => subject.OnNext(e));
}
return subject;
}
static IEnumerable<Task<int>> GetIntsIEnumerable()
{
Random rnd = new Random();
foreach (int i in Enumerable.Range(1, 10))
{
yield return Task.Run(async () =>
{
await Task.Delay(rnd.Next(0, 5000));
if (i == 6)
throw new ArgumentException();
return i;
});
}
}
}
/// <summary>
/// Functional data data to represent a discriminated
/// union of two possible types.
/// </summary>
/// <typeparam name="TL">Type of "Left" item.</typeparam>
/// <typeparam name="TR">Type of "Right" item.</typeparam>
public class Either<TL, TR>
{
private readonly TL left;
private readonly TR right;
private readonly bool isLeft;
public Either(TL left)
{
this.left = left;
this.isLeft = true;
}
public Either(TR right)
{
this.right = right;
this.isLeft = false;
}
public T Match<T>(Func<TL, T> leftFunc, Func<TR, T> rightFunc)
{
if (leftFunc == null)
{
throw new ArgumentNullException(nameof(leftFunc));
}
if (rightFunc == null)
{
throw new ArgumentNullException(nameof(rightFunc));
}
return this.isLeft ? leftFunc(this.left) : rightFunc(this.right);
}
/// <summary>
/// If right value is assigned, execute an action on it.
/// </summary>
/// <param name="rightAction">Action to execute.</param>
public void DoRight(Action<TR> rightAction)
{
if (rightAction == null)
{
throw new ArgumentNullException(nameof(rightAction));
}
if (!this.isLeft)
{
rightAction(this.right);
}
}
public TL LeftOrDefault() => this.Match(l => l, r => default);
public TR RightOrDefault() => this.Match(l => default, r => r);
public static implicit operator Either<TL, TR>(TL left) => new Either<TL, TR>(left);
public static implicit operator Either<TL, TR>(TR right) => new Either<TL, TR>(right);
}
}
Notification<T>
包装器。这是更好、更干净的解决方案。SemaphoreSlim
仅用于测试。我想在调用OnCompleted()
之前保持控制台应用程序运行。 - Matěj Pokorný.Dematerialize()
将流返回到标准的IObservable<T>
。 - Enigmativity