如何在Rx.Net中实现带有异步累加器的ScanAsync运算符

Question

Rx.Net中的Scan运算符具有签名：

public static IObservable<TAccumulate> Scan<TSource, TAccumulate>(this IObservable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> accumulator);

蓄能器为

Func<TAccumulate, TSource, TAccumulate> accumulator

在尝试实现具有异步状态转换的状态机模型时，我发现具有以下签名的ScanAsync运算符会很有帮助。

public static IObservable<TAccumulate> Scan<TSource, TAccumulate>(this IObservable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, Task<TAccumulate>> accumulator);

累加器具有签名

Func<TAccumulate, TSource, Task<TAccumulate>> accumulator

理想的应用程序代码应该是这样的(类似于普通的Scan运算符，不同之处在于使用异步累加器(。

IObservable<TEvent> events;
IObservable<State> states = events.ScanAsync(
initialState, 
async (previousState, evt) => {
var newState = await transitionAsync(previousState, evt);
return newState;
});

看起来MS正在开发AsyncRx.NET，但它还没有发布(没有时间表(。

---

相关内容：

如果通过BehaviourSubject为状态建模异步状态机，并订阅可观察的事件，如以下代码

IObservable<TEvent> events;
BehaviourSubject<State> states = new BehaviourSubject<State>(initialState);
events.Subscribe(async e => {
var newState = await transition(states.Value, e);
states.OnNext(newState);
})

我想在某些情况下可能会有比赛条件。

我试过用实现它

IObservable<TS> ScanAsync<TS, TE>(
IObservable<TE> source,
Func<TS, TE, Task<TS>> reducer,
TS initialState)
{
var states = from m in source.Take(1)
from nextState in reducer(initialState, m).ToObservable()
from s in ScanAsync(source.Skip(1), reducer, nextState)
select s;
return Observable.Return(initialState).Concat(states);
}

然而，它有时有效，有时只是被屏蔽了，我不知道是什么原因造成的。

Answer 1

您可以使用Scan运算符创建一个中间IObservable<Task<TAccumulate>>，然后使用Concat运算符对其进行展平：

public static IObservable<TAccumulate> Scan<TSource, TAccumulate>(
this IObservable<TSource> source,
TAccumulate seed,
Func<TAccumulate, TSource, Task<TAccumulate>> accumulator)
{
return source.Scan(Task.FromResult(seed), async (previousTask, item) =>
{
return await accumulator(await previousTask, item);
}).Concat();
}

上面的实现使用Concat重载，该重载接受任务的可观察项，而不是嵌套的可观察：

// Concatenates all task results, as long as the previous task terminated successfully.
public static IObservable<TSource> Concat<TSource>(
this IObservable<Task<TSource>> sources);