Search code examples
f#akka.netakka.net-streams

How to use Akka.Streams.*.ConcatMany in F#?


I want to create a flow that creates a new source (it will be a persistence query) out of incoming elements, and then flattens the results. Something like this simplified example:

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

this code compiles and works as expected. My problem is that when I translate it to F#:

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

I get an error saying

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

I think that the cause of that is that F# handles co/contravariance differently than C# and cannot simply convert these generic specializations (https://github.com/fsharp/fslang-suggestions/issues/162), but I cannot figure out a way to make a convertion between an int and a SourceShape<int>. Is it possible to convert this example to F#?


Solution

  • Looking at the code on GitHub, it appears that Source<TOut, TMat> is a direct implementation of IGraph, so you should just be able to cast it:

    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>)

    I think the biggest difference between the C# and F# usage is that C# will automatically do the upcast for you.