Как использовать Akka.Streams. *. ConcatMany в F #?

Question

Я хочу создать поток, который создает новый источник (это будет запрос постоянства) из входящих элементов, а затем выравнивает результаты. Что-то вроде этого упрощенного примера:

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 #?

Answer 1

Глядя на код на 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 # автоматически сделает upcast для вас.

Answer 2

Один из найденных мной обходных путей - использовать Akkling.Streams Библиотека обёрток:

open Akkling.Streams

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

Вопрос о том, как это сделать без Akkling, остается открытым.