libraryDependencies += "dev.zio" %% "zio-interop-cats" % "23.0.x"
ZIO integrates with Typelevel libraries by providing an instance of Concurrent
, Temporal
and Async
for Task
as required, for instance, by fs2
, doobie
and http4s
.
For convenience, the ZIO library defines an alias as follows:
type Task[A] = ZIO[Any, Throwable, A]
Therefore, we provide Cats Effect instances based on this specific datatype.
In order to get a Concurrent[Task]
or Concurrent[RIO[R, *]]
(note *
is kind-projector notation) we need an
implicit Runtime[R]
in scope. The easiest way to get it is using ZIO.runtime
:
import cats.effect._
import zio._
import zio.interop.catz._
def ceConcurrentForTaskExample =
ZIO.runtime.flatMap { implicit r: Runtime[Any] =>
// the presence of a runtime allows you to summon Cats Effect Typeclasses
val F: cats.effect.Concurrent[Task] = implicitly
F.racePair(F.unit, F.unit)
}
Temporal
requires a Runtime
with Clock
.
import cats.effect._
import zio._
import zio.interop.catz._
def ceTemporal =
ZIO.runtime.flatMap { implicit r: Runtime[Clock] =>
val F: cats.effect.Temporal[Task] = implicitly
F.sleep(1.second) *> F.unit
}
Similar to the other examples, we require a Runtime
with the Clock
layer.
import cats.effect._
import zio._
import zio.interop.catz._
def ceAsync =
ZIO.runtime.flatMap { implicit r: Runtime[Clock] =>
val F: cats.effect.Async[Task] = implicitly
F.racePair(F.unit, F.sleep(1.second) *> F.unit)
}
There are many other typeclasses and useful conversions that this library provides implementations for:
- See
zio/interop/cats.scala
file to see all available typeclass implementations for the Cats Effect 3 typeclasses - See
zio/stream/interop/cats.scala
for ZStream typeclass implementations - See
zio/stream/interop/FS2StreamSyntax.scala
for FS2 <-> ZStream conversions
In the examples above, we had to bring the Runtime[Clock]
via the ZIO.runtime
combinator. This may
not be ideal since everywhere you use these typeclasses, you will now be required to feed in the Runtime
.
For example, with FS2:
def example(implicit rts: Runtime[Clock]): Task[Unit] =
fs2.Stream
.awakeDelay[Task](10.seconds) // this type annotation is mandatory
.evalTap(in => cats.effect.std.Console.make[Task].println(s"Hello $in"))
.compile
.drain
Rather than requiring the runtime implicit, we can add an import (if we don't mind depending on Runtime.default
):
import zio.interop.catz._
import zio.interop.catz.implicits._
val example: Task[Unit] =
fs2.Stream
.awakeDelay[Task](10.seconds)
.evalTap(in => cats.effect.std.Console.make[Task].println(s"Hello $in"))
.compile
.drain
The major downside to doing this is you rely on live implementations of Clock
to summon instances which
makes testing much more difficult for any Cats Effect code that you use.
If you only need instances for cats-core
typeclasses, not cats-effect
import zio.interop.catz.core._
:
import zio.interop.catz.core._
Note that this library only has an Optional
dependency on cats-effect โ if you or your libraries don't depend on it, this library will not add it to the classpath.
The following example shows how to use ZIO with Doobie (a library for JDBC access) and FS2 (a streaming library), which both rely on Cats Effect instances (cats.effect.Async
and cats.effect.Temporal
):
import zio._
import zio.interop.catz._
import doobie._
import doobie.implicits._
import zio.clock.Clock
import scala.concurrent.duration._
object DoobieH2Example extends App {
override def run(args: List[String]): URIO[zio.ZEnv, ExitCode] =
ZIO.runtime.flatMap { implicit r: Runtime[Clock] =>
val xa: Transactor[Task] =
Transactor.fromDriverManager[Task]("org.h2.Driver", "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1", "user", "")
sql"SELECT 42"
.query[Int]
.stream
.transact(xa)
.delayBy(1.second)
.evalTap(i => ZIO.succeedBlocking(println(s"Data $i"))))
.compile
.drain
.exitCode
}
}
Learn more on the ZIO Interop Cats homepage!
For the general guidelines, see ZIO contributor's guide.
See the Code of Conduct