如何在 F# 中使用 Akka.Streams.*.ConcatMany？答案

【问题标题】：How to use Akka.Streams.*.ConcatMany in F#?如何在 F# 中使用 Akka.Streams.*.ConcatMany？
【发布时间】：2019-10-21 10:19:30
【问题描述】：

我想创建一个流，从传入的元素中创建一个新的源（它将是一个持久性查询），然后将结果展平。像这个简化的例子：

var z = Source.Single(1).ConcatMany(i => Source.Single(i));

此代码按预期编译和工作。我的问题是，当我将它翻译成 F# 时：

let z = Source.Single(1).ConcatMany(fun i -> Source.Single(i))

我收到一个错误提示

This expression was expected to have type
    'IGraph<SourceShape<'a>,Akka.NotUsed>'    
but here has type
    'Source<int,Akka.NotUsed>'

我认为其原因是 F# 处理协/逆变的方式与 C# 不同，并且不能简单地转换这些泛型特化 (https://github.com/fsharp/fslang-suggestions/issues/162)，但我无法找到在 int 和SourceShape<int>。是否可以将此示例转换为 F#？

【问题讨论】：

标签： f# akka.net akka.net-streams

【解决方案1】：

查看the code on GitHub，看来Source<TOut, TMat> 是IGraph 的直接实现，所以您应该可以转换它：

public sealed class Source<TOut, TMat> : IFlow<TOut, TMat>, IGraph<SourceShape<TOut>, TMat>

let z = Source.Single(1).ConcatMany(fun i -> Source.Single(i) :> IGraph<SourceShape<int>,Akka.NotUsed>)

我认为 C# 和 F# 用法之间的最大区别在于 C# 会自动为您进行向上转换。

【讨论】：

这也适用于推断类型，例如：let z = Source.Single(1).ConcatMany(fun i -> Source.Single(i) :> _)
你也可以使用 F# upcast。

【解决方案2】：

我发现的一种解决方法是使用 Akkling.Streams 包装库：

open Akkling.Streams

let x =
    Source.singleton 1
    |> Source.collectMap(fun x -> Source.singleton x)

如何在没有 Akkling 的情况下做到这一点的问题仍然悬而未决。

【讨论】：