c# - 转换 IEnumerable> 到 IObservable带异常处理
问题描述
我想转换IEnumerable<Task<T>>
为IObservable<T>
. 我在这里找到了解决方案:
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);
}
}
解决方案
有一个内置的机制来处理这样的错误。只需使用将.Materialize()
an 更改IObservable<T>
为 anIObservable<Notification<T>>
并允许将错误和完成视为正常值的运算符。
所以,作为一个例子,Observable.Return<int>(42)
产生一个值42
和一个完成,但Observable.Return<int>(42).Materialize()
产生一个值Notification.CreateOnNext<int>(42)
,然后是一个值Notification.CreateOnCompleted<int>()
,然后是一个正常的完成。
如果您有一个产生错误的序列,那么您实际上会得到一个值Notification.CreateOnError<T>(exception)
,然后是正常完成。
这一切都意味着您可以像这样更改代码:
IObservable<Notification<T>> ToObservable<T>(IEnumerable<Task<T>> source)
{
return source.Select(t => t.ToObservable().Materialize()).Merge();
}
根据我的喜好,您的测试代码有点复杂。您永远不需要以使用它们的方式使用 aSemaphoreSlim
或 a Subject
。
我已经编写了自己的测试代码。
void Main()
{
var r = new Random();
IEnumerable<Task<int>> source =
Enumerable
.Range(0, 10).Select(x => Task.Factory.StartNew(() =>
{
Thread.Sleep(r.Next(10000));
if (x % 3 == 0) throw new NotSupportedException($"Failed on {x}");
return x;
}));
IObservable<Notification<int>> query = source.ToObservable();
query
.Do(x =>
{
if (x.Kind == NotificationKind.OnError)
{
Console.WriteLine(x.Exception.Message);
}
})
.Where(x => x.Kind == NotificationKind.OnNext) // Only care about vales
.Select(x => x.Value)
.Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("Done."));
}
public static class Ex
{
public static IObservable<Notification<T>> ToObservable<T>(this IEnumerable<Task<T>> source)
{
return source.Select(t => t.ToObservable().Materialize()).Merge();
}
}
该代码的典型运行会产生:
失败 3 2 5 4 0 失败 9日失败 6日失败 7 1 8 完毕。
推荐阅读
- angular - Angular 工作区 - 我可以在多个 package.json 中分离依赖项吗?
- graphql - 在 shopify 应用开发中自定义 ResourcePicker 数据
- python - 中间件内的烧瓶会话
- flutter - 如何在 Flutter 中向标题 ExpansionTile 添加渐变
- html - html表中列之间的空格太多
- javascript - VueJS:在 Vuex mapAction then 方法中未定义
- jquery - mdbAutocomplete - 新旧自动完成列表的叠加
- c++ - 哪种方式可以更有效地重载运算符,为什么?
- ios - 当媒体段数较少时,AVPlayer 是否播放视频块
- reactjs - 正如我们在 PHP 中使用的那样,如何在 html 元素的属性双引号中使用 JXS?