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

Question

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

score 2 · Accepted Answer · answered Oct 22 '19 at 13:13

2

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.

answered Oct 22 '19 at 13:13

Aaron M. Eshbach

6,380
12
22

This also works with inferred type, like that: `let z = Source.Single(1).ConcatMany(fun i -> Source.Single(i) :> _)` – pmbanka Oct 23 '19 at 09:16
You can just use F# `upcast` as well. – Bartosz Sypytkowski Oct 24 '19 at 07:42

score 1 · Answer 2 · answered Oct 21 '19 at 10:35

1

One workaround that I found is to use Akkling.Streams wrapper library:

open Akkling.Streams

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

the question how to do this without Akkling remains open.

answered Oct 21 '19 at 10:35

pmbanka

1,902
17
24

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

2 Answers2