Fs2: Provide a conversion from `cats.Id` to `fs2.Pure`

[gvolpe]

I wonder whether it makes sense to add an implicit or explicit conversion from cats.Id to fs2.Pure since they both represent pure values.

Motivation

We recently introduced fs2 in one of our teams at work. We were already using cats.effect.IO as our main effect and all the tests were written in terms of Id.

Now a few methods that have the signature Stream[F, ?] need to interact with other methods defined as F[_] so in our tests we ended up having Stream[Id, ?] and in some cases we want to be able to compile this to a List[String] where a Sync[Id] instance is needed (and of course does not exist). So one way to workaround this is to cast it to Stream[IO, ?]:

val streamId: Stream[Id, String] = ???
val rs1: List[String] = streamId.asInstanceOf[Stream[IO, String]].toList.unsafeRunSync()

fs2 already defines covary[F] which can go from Pure to IO for example but it does not work for cats.Id so the following conversion fails to compile:

// Error: type arguments [cats.effect.IO] do not conform to method covary's type parameter bounds [F2[x] >: cats.Id[x]]
val rs2: List[String] = streamId.covary[IO].toList.unsafeRunSync()

Proposal

I thought about a few options but I'd like to hear your thoughts:

1. Adding a covary that supports casting from Id to Pure and from Id to IO in the same way Pure can now be converted to IO.
2. Have an internal implicit conversion from Id to Pure so every Stream[Id, ?] is simply threatened as Stream[Pure, ?].

[mpilquist]

Note that Pure[A] is uninhabited whereas Id[A] is inhabited by all the values of A. Hence you can call Stream.eval, which then breaks if you covary:

@ Stream.eval[Id, Int](1).asInstanceOf[Stream[Pure, Int]].toList
java.lang.ClassCastException: java.lang.Integer cannot be cast to cats.effect.IO
  cats.effect.IOInstances$$anon$1.attempt(IO.scala:447)
  fs2.internal.CompileScope.interruptibleEval(CompileScope.scala:449)
  fs2.internal.Algebra$.$anonfun$compileLoop$2(Algebra.scala:319)
  cats.effect.internals.IORunLoop$.liftedTree3$1(IORunLoop.scala:207)
  cats.effect.internals.IORunLoop$.step(IORunLoop.scala:207)
  cats.effect.IO.unsafeRunTimed(IO.scala:304)
  cats.effect.IO.unsafeRunSync(IO.scala:239)
  fs2.Stream$PureOps$.toList$extension(Stream.scala:2715)
  ammonite.$sess.cmd5$.<init>(cmd5.sc:1)
  ammonite.$sess.cmd5$.<clinit>(cmd5.sc)

[mpilquist]

BTW, a safe way to do this is translating to IO using IO.pure constructor. The call to unsafeRunSync is totally safe here since we know there's no Sync[Id] and hence there's no suspended side-effects in the stream.

If you go this route, I recommend keeping such a conversion in an implicit class or other utility rather than using compile.toList.unsafeRunSync directly in application code, coupled with a comment explaining why it's safe in this case.

@ val idToIo = new (Id ~> IO) { def apply[A](a: Id[A]) = IO.pure(a) }
idToIo: AnyRef with Id ~> IO = ammonite.$sess.cmd12$$anon$1@585b73dc

@ implicit class StreamIdOps[A](private val stream: Stream[Id, A]) {
    def toList: List[A] = stream.translate(idToIo).compile.toList.unsafeRunSync
  }
defined class StreamIdOps

@ Stream.eval[Id, Int](42).toList
res14: List[Int] = List(42)

We could add this to FS2 but the use case is kind of specialized I think. No strong feelings though.

Note we'd have to use more syntax tricks to make a Stream[Id, A] seamlessly interoperable. For example, this arguably should work:

@ Stream.eval[Id, Int](42) ++ Stream.eval(IO(43))
cmd16.sc:1: type mismatch;
 found   : fs2.Stream[cats.effect.IO,Int]
 required: fs2.Stream[cats.Id,?]
val res16 = Stream.eval[Id, Int](42) ++ Stream.eval(IO(43))
                                                   ^
Compilation Failed

We can extend the example above though:

@ def idToApplicative[F[_]: Applicative]: Id ~> F = 
      new (Id ~> F) { def apply[A](a: Id[A]) = a.pure[F] }
defined function idToApplicative

@ implicit class StreamIdOps[A](private val stream: Stream[Id, A]) {
    def toList: List[A] = stream.translate(idToIo).compile.toList.unsafeRunSync
    def ++[F[_]: Applicative, A2 >: A](s2: => Stream[F, A2]): Stream[F, A2] =
      stream.translate(idToApplicative[F]).append(s2)
  }
defined class StreamIdOps

@ Stream.eval[Id, Int](42) ++ Stream.eval(IO(43))
res17: Stream[IO, Int] = Stream(..)

We could extend this idea and duplicate all operations defined in PureOps.

[gvolpe]

Thanks for the detailed thoughts @mpilquist !

Given that we are still on the road to 1.0.0 maybe it's worth adding these operations if you're guys okay with it, I can work on a PR.

[mpilquist]

I’m on board - only downside is keeping IdOps in sync with PureOps and InvariantOps but that isn’t that big of a deal.

[gvolpe]

I found a few problems adding IdOps:

• It also defines a toList: List[A] operation as syntax of Stream and it conflicts with the one defined in PureOps. For example this does not compile:

Pull.pure(1).stream.toList // ERROR 

<console>:17: error: type mismatch;
 found   : fs2.Stream[Nothing,Nothing]
 required: ?{def toList: ?}
Note that implicit conversions are not applicable because they are ambiguous:
 both method PureOps in object Stream of type [O](s: fs2.Stream[fs2.Pure,O])fs2.Stream.PureOps[O]
 and method IdOps in object Stream of type [O](s: fs2.Stream[cats.Id,O])fs2.Stream.IdOps[O]
 are possible conversion functions from fs2.Stream[Nothing,Nothing] to ?{def toList: ?}
       Pull.pure(1).stream.toList

One needs to specify the type of that Stream to get it compiling:

Pull.pure(1).stream.asInstanceOf[Stream[Pure, Nothing]].toList
Pull.pure(1).stream.asInstanceOf[Stream[Id, Nothing]].toList

• append or ++ does not work when trying to append a Stream[Id, O] to a Stream[IO, O]. Example:

val sid: Stream[Id, Int] = Stream(1,2,3)

sid ++ Stream.eval(IO.pure(4)) // OK
Stream.eval(IO.pure(4)) ++ sid.covary[IO] // OK
Stream.eval(IO.pure(4)) ++ sid // ERROR found   : fs2.Stream[cats.Id,Int] required: fs2.Stream[cats.effect.IO,?]

What do you guys suggest? Maybe we're better off without it in fs2 core I guess :smile:

[SystemFw]

Start refactoring your tests :P

Seriously though, I wouldn't have minded to add it for convenience, but it looks like it's going to cause problems in its current form :(

So I'd say to keep it out unless there are fresh ideas to make it work nicely (haven't thought it through though, there might be a way)

[gvolpe]

I wish it was only about refactoring tests XD

I think it will affect the users in this way so I guess it's better to keep it out of the core and use the natural transformation trick whenever this is needed, unless @mpilquist had something else in mind.

[mpilquist]

It also defines a toList: List[A] operation as syntax of Stream and it conflicts with the one defined in PureOps. For example this does not compile:

Argh, I didn't see the conflict with inferring unconstrained type params coming. Not sure there's a good solution to this one. Maybe something with prioritizing the implicit conversion to IdOps but that's a delicate balance honestly -- can't deprioritize it to the point that InvariantOps ends up getting preferred.

append or ++ does not work when trying to append a Stream[Id, O] to a Stream[IO, O]

I think this one is easier to fix -- you'd just need an implicit conversion def liftIdStream[F[_]: Applicative, O](s: Stream[Id, O]): Stream[F, O] = s.translate(idToF) in the companion of Stream -- I think.

[SystemFw]

How about a less convenient but safer alternative, just an explicit covaryId[F]?
It will definitely make it more annoying than with Pure streams, but maybe not too much?
Do you think it's worth trying?

[gvolpe]

In my case covaryId[F] will do and wrt the fs2 core I think it helps not polluting the code with an IdOps that has to be kept in sync with PureOps in time. Will go for this one, thanks! :+1: