Merge pull request #1262 from Roman-Statsura/streams-scala3

Add support Scala 3 for modules streams and fs2CE3Interop

build.sbt

@@ -395,13 +395,19 @@ lazy val fs2CE2Interop = projectMatrix
 lazy val fs2CE3Interop = projectMatrix
   .in(interop / "fs2" / "ce3")
   .settings(
-    name                             := "tofu-fs2-ce3-interop",
-    libraryDependencies += fs2CE3,
-    libraryDependencies += glassMacro % Test,
-    defaultSettings
+    name := "tofu-fs2-ce3-interop",
+    libraryDependencies ++= {
+      CrossVersion.partialVersion(scalaVersion.value) match {
+        case Some((2, _)) => Seq(fs2CE3, glassMacro % Test)
+        case Some((3, _)) => Seq(fs2CE3)
+        case _            => Seq.empty
+      }
+    },
+    defaultSettings,
+    scala3MigratedModuleOptions
   )
-  .jvmPlatform(scala2Versions)
-  .dependsOn(coreCE3, streams, derivation % "compile->test")
+  .jvmPlatform(scalaVersions = scala2And3Versions)
+  .dependsOn(coreCE3, streams)
 
 lazy val doobie = projectMatrix
   .in(modules / "doobie" / "core-ce2")
@@ -446,9 +452,10 @@ lazy val streams = projectMatrix
   .settings(
     libraryDependencies ++= List(fs2 % Test),
     defaultSettings,
+    scala3MigratedModuleOptions,
     name := "tofu-streams",
   )
-  .jvmPlatform(scala2Versions)
+  .jvmPlatform(scalaVersions = scala2And3Versions)
   .dependsOn(kernel)
 
 val examples = file("examples")

modules/interop/fs2/ce3/src/main/scala-2/tofu/fs2/package.scala → modules/interop/fs2/ce3/src/main/scala/tofu/fs2/package.scala

@@ -5,7 +5,7 @@ import tofu.lift.Lift
 
 package object fs2 {
 
-  type LiftStream[S[_], F[_]] = Lift[Stream[F, *], S]
+  type LiftStream[S[_], F[_]] = Lift[Stream[F, _], S]
 
   object LiftStream {
     def apply[S[_], F[_]](implicit ev: LiftStream[S, F]): LiftStream[S, F] = ev

modules/interop/fs2/ce3/src/main/scala-2/tofu/fs2Instances/fs2Instances.scala → modules/interop/fs2/ce3/src/main/scala/tofu/fs2Instances/fs2Instances.scala

@@ -20,18 +20,18 @@ private[fs2Instances] trait Fs2Instances1 extends Fs2Instances2 {
 
   final implicit def fs2StreamRunContext[F[_], G[_], R](implicit
       ctx: WithRun[F, G, R]
-  ): WithRun[Stream[F, *], Stream[G, *], R] =
+  ): WithRun[Stream[F, _], Stream[G, _], R] =
     new FS2RunContext[F, G, R] {
       override val F: WithRun[F, G, R] = ctx.self
 
       override implicit def WP: WithProvide[F, G, R] = ctx.self
     }
 
-  implicit def fs2LiftInstance[F[_]]: Lift[F, Stream[F, *]] =
-    Lift.byFunK(funK(Stream.eval(_)))
+  implicit def fs2LiftInstance[F[_]]: Lift[F, Stream[F, _]] =
+    Lift.byFunK(funK[F, Stream[F, _]](Stream.eval(_)))
 
-  implicit def fs2ChunksInstance[F[_]]: Chunks[Stream[F, *], Chunk] =
-    new Chunks[Stream[F, *], Chunk] {
+  implicit def fs2ChunksInstance[F[_]]: Chunks[Stream[F, _], Chunk] =
+    new Chunks[Stream[F, _], Chunk] {
       override def chunkN[A](fa: Stream[F, A])(n: Int): Stream[F, Chunk[A]] = fa.chunkN(n)
 
       override def chunks[A](fa: Stream[F, A]): Stream[F, Chunk[A]] = fa.chunks
@@ -41,13 +41,13 @@ private[fs2Instances] trait Fs2Instances1 extends Fs2Instances2 {
       override def cons[A](fa: Stream[F, A])(c: Chunk[A]): Stream[F, A] = fa.cons(c)
     }
 
-  implicit def fs2MergeInstance[F[_]: Concurrent]: Merge[Stream[F, *]] =
-    new Merge[Stream[F, *]] {
+  implicit def fs2MergeInstance[F[_]: Concurrent]: Merge[Stream[F, _]] =
+    new Merge[Stream[F, _]] {
       override def merge[A](fa: Stream[F, A])(that: Stream[F, A]): Stream[F, A] = fa merge that
     }
 
-  implicit def fs2CompileInstance[F[_]: Sync]: Compile[Stream[F, *], F] =
-    new Compile[Stream[F, *], F] {
+  implicit def fs2CompileInstance[F[_]: Sync]: Compile[Stream[F, _], F] =
+    new Compile[Stream[F, _], F] {
 
       override def drain[A](fa: Stream[F, A]): F[Unit] = fa.compile.drain
 
@@ -57,34 +57,34 @@ private[fs2Instances] trait Fs2Instances1 extends Fs2Instances2 {
         fa.compile.to(ev)
     }
 
-  implicit def fs2ParFlattenInstance[F[_]: Concurrent]: ParFlatten[Stream[F, *]] =
-    new ParFlatten[Stream[F, *]] {
+  implicit def fs2ParFlattenInstance[F[_]: Concurrent]: ParFlatten[Stream[F, _]] =
+    new ParFlatten[Stream[F, _]] {
       override def parFlatten[A](ffa: Stream[F, Stream[F, A]])(maxConcurrent: Int): Stream[F, A] =
         ffa.parJoin(maxConcurrent)
     }
 
-  implicit def fs2PaceInstance[F[_]: CETemporal]: Pace[Stream[F, *]] =
-    new Pace[Stream[F, *]] {
+  implicit def fs2PaceInstance[F[_]: CETemporal]: Pace[Stream[F, _]] =
+    new Pace[Stream[F, _]] {
 
       override def throttled[A](fa: Stream[F, A])(rate: FiniteDuration): Stream[F, A] = fa.metered(rate)
 
       override def delay[A](fa: Stream[F, A])(d: FiniteDuration): Stream[F, A] = fa.delayBy(d)
     }
 
-  implicit def fs2TemporalInstance[F[_]: CETemporal]: Temporal[Stream[F, *], Chunk] =
-    new Temporal[Stream[F, *], Chunk] {
+  implicit def fs2TemporalInstance[F[_]: CETemporal]: Temporal[Stream[F, _], Chunk] =
+    new Temporal[Stream[F, _], Chunk] {
       override def groupWithin[A](fa: Stream[F, A])(n: Int, d: FiniteDuration): Stream[F, Chunk[A]] =
         fa.groupWithin(n, d)
     }
 
-  implicit def fs2RegionThrowInstance[F[_]]: Region[Stream[F, *], F, Resource.ExitCase] =
-    new Region[Stream[F, *], F, Resource.ExitCase] {
+  implicit def fs2RegionThrowInstance[F[_]]: Region[Stream[F, _], F, Resource.ExitCase] =
+    new Region[Stream[F, _], F, Resource.ExitCase] {
       override def regionCase[R](open: F[R])(close: (R, Resource.ExitCase) => F[Unit]): Stream[F, R] =
         Stream.bracketCase(open)(close)
     }
 
-  implicit def fs2BroadcastInstances[F[_]: Concurrent]: Broadcast[Stream[F, *]] =
-    new Broadcast[Stream[F, *]] {
+  implicit def fs2BroadcastInstances[F[_]: Concurrent]: Broadcast[Stream[F, _]] =
+    new Broadcast[Stream[F, _]] {
 
       override def broadcast[A](fa: Stream[F, A])(processors: Stream[F, A] => Stream[F, Unit]*): Stream[F, Unit] =
         fa.broadcastThrough(processors: _*)
@@ -95,46 +95,47 @@ private[fs2Instances] trait Fs2Instances1 extends Fs2Instances2 {
 }
 
 private[fs2Instances] trait Fs2Instances2 extends Fs2Instances3 {
-  final implicit def fs2StreamLocal[F[_], R](implicit ctx: F WithLocal R): WithLocal[Stream[F, *], R] =
+  final implicit def fs2StreamLocal[F[_], R](implicit ctx: F WithLocal R): WithLocal[Stream[F, _], R] =
     new FS2Local[F, R] { override val F: F WithLocal R = ctx.asWithLocal }
 
   final implicit def fs2StreamProvide[F[_], G[_], R](implicit
       ctx: WithProvide[F, G, R]
-  ): WithProvide[Stream[F, *], Stream[G, *], R] =
+  ): WithProvide[Stream[F, _], Stream[G, _], R] =
     new FS2Provide[F, G, R] { override val WP: WithProvide[F, G, R] = ctx }
 }
 
 private[fs2Instances] trait Fs2Instances3 {
-  final implicit def fs2StreamContext[F[_], R](implicit ctx: F WithContext R): WithContext[Stream[F, *], R] =
+  final implicit def fs2StreamContext[F[_], R](implicit ctx: F WithContext R): WithContext[Stream[F, _], R] =
     new FS2Context[F, R] { override val F: F WithContext R = ctx.asWithContext }
 }
 
-class FS2StreamHKInstance[A] extends Embed[Stream[*[_], A]] with FunctorK[Stream[*[_], A]] {
+class FS2StreamHKInstance[A]
+    extends Embed[({ type L[x[_]] = Stream[x, A] })#L] with FunctorK[({ type L[x[_]] = Stream[x, A] })#L] {
   def embed[F[_]: FlatMap](ft: F[Stream[F, A]]): Stream[F, A]      = Stream.force(ft)
   def mapK[F[_], G[_]](af: Stream[F, A])(fk: F ~> G): Stream[G, A] = af.translate(fk)
 }
 
-trait FS2Context[F[_], R] extends WithContext[Stream[F, *], R] {
+trait FS2Context[F[_], R] extends WithContext[Stream[F, _], R] {
   implicit def F: F WithContext R
 
-  val functor: Functor[Stream[F, *]] = implicitly
+  val functor: Functor[Stream[F, _]] = implicitly
   def context: Stream[F, R]          = Stream.eval(F.context)
 }
 
-trait FS2Local[F[_], R] extends FS2Context[F, R] with WithLocal[Stream[F, *], R] {
+trait FS2Local[F[_], R] extends FS2Context[F, R] with WithLocal[Stream[F, _], R] {
   implicit def F: F WithLocal R
 
   def local[A](fa: Stream[F, A])(project: R => R): Stream[F, A] = fa.translate(funKFrom[F](F.local(_)(project)))
 }
 
-trait FS2Provide[F[_], G[_], R] extends WithProvide[Stream[F, *], Stream[G, *], R] {
+trait FS2Provide[F[_], G[_], R] extends WithProvide[Stream[F, _], Stream[G, _], R] {
   implicit def WP: WithProvide[F, G, R]
 
   def runContext[A](fa: Stream[F, A])(ctx: R): Stream[G, A] = fa.translate(funKFrom[F](WP.runContext(_)(ctx)))
   def lift[A](fa: Stream[G, A]): Stream[F, A]               = fa.translate(funKFrom[G](WP.lift(_)))
 }
 
 trait FS2RunContext[F[_], G[_], R]
-    extends FS2Local[F, R] with FS2Provide[F, G, R] with WithRun[Stream[F, *], Stream[G, *], R] {
+    extends FS2Local[F, R] with FS2Provide[F, G, R] with WithRun[Stream[F, _], Stream[G, _], R] {
   implicit def F: WithRun[F, G, R]
 }

modules/interop/fs2/ce3/src/main/scala/tofu/fs2Syntax/pre.scala

@@ -0,0 +1,16 @@
+package tofu.fs2Syntax
+
+import cats.tagless.ApplyK
+import cats.data.Tuple2K
+import fs2.Stream
+import tofu.higherKind.Pre.T
+import tofu.syntax.funk.funK
+
+object pre {
+  implicit final class TofuPreStreamSyntax[F[_], U[f[_]]](private val self: U[T[F, _]]) extends AnyVal {
+    def attachStream(alg: U[Stream[F, _]])(implicit U: ApplyK[U]): U[Stream[F, _]] =
+      U.map2K(self, alg)(
+        funK[Tuple2K[T[F, _], Stream[F, _], _], Stream[F, _]](t2k => Stream.exec(t2k.first.value) ++ t2k.second)
+      )
+  }
+}

modules/interop/fs2/ce3/src/test/scala/tofu/fs2Instances/Fs2InstancesSuite.scala

@@ -0,0 +1,38 @@
+package tofu.fs2Instances
+
+import cats.tagless.FunctorK
+import fs2._
+import tofu._
+import tofu.higherKind.Embed
+
+import scala.annotation.nowarn
+
+object Fs2InstancesSuite {
+
+  def summonFs2HKInstances[A](): Unit = {
+    @nowarn val x = implicitly[FunctorK[({ type L[x[_]] = Stream[x, A] })#L]]
+    @nowarn val y = implicitly[Embed[({ type L[x[_]] = Stream[x, A] })#L]]
+    ()
+  }
+
+  def summonFs2ContextInstances[R, F[_]: ({ type L[x[_]] = WithContext[x, R] })#L](): Unit = {
+    implicitly[WithContext[Stream[F, *], R]]
+    ()
+  }
+
+  def summonFs2LocalInstances[R, F[_]: ({ type L[x[_]] = WithLocal[x, R] })#L](): Unit = {
+    implicitly[WithLocal[Stream[F, *], R]]
+    ()
+  }
+
+  def summonFs2ProvideInstances[R, G[_], F[_]: ({ type L[x[_]] = WithProvide[x, G, R] })#L](): Unit = {
+    implicitly[WithProvide[Stream[F, _], Stream[G, _], R]]
+    ()
+  }
+
+  def summonFs2RunInstances[R, G[_], F[_]: ({ type L[x[_]] = WithRun[x, G, R] })#L](): Unit = {
+    implicitly[WithRun[Stream[F, _], Stream[G, _], R]]
+    ()
+  }
+
+}

modules/streams/src/main/scala-3/tofu/streams/internal/internal.scala

@@ -0,0 +1,5 @@
+package tofu.streams
+
+package object internal {
+  private[tofu] type Factory[-A, +C] = scala.collection.Factory[A, C]
+}

modules/streams/src/main/scala-2/tofu/streams/Emits.scala → modules/streams/src/main/scala/tofu/streams/Emits.scala

@@ -28,7 +28,7 @@ trait Evals[F[_], G[_]] extends Emits[F] with Lift[G, F] {
 
   implicit val monad: Monad[F]
 
-  lazy val applicative: Applicative[F] = monad
+  val applicative: Applicative[F] = monad
 
   def eval[A](ga: G[A]): F[A] = lift(ga)
 

modules/streams/src/test/scala-2/tofu/streams/SyntaxTypeInferenceSuite.scala → modules/streams/src/test/scala/tofu/streams/SyntaxTypeInferenceSuite.scala

@@ -7,7 +7,7 @@ import tofu.syntax.streams.all._
 
 object SyntaxTypeInferenceSuite {
 
-  class EvalsInference[S[_]: Evals[*[_], F], F[_]: Applicative] {
+  class EvalsInference[S[_]: ({ type L[x[_]] = Evals[x, F] })#L, F[_]: Applicative] {
     def foo: S[Unit] = eval(().pure[F])
     def bar: S[Unit] = evals(List((), (), ()).pure[F])
   }