Perform, a successor to UnliftIO

.github/workflows/ci.yml

@@ -73,7 +73,7 @@ jobs:
         run: sbt ++${{ matrix.scala }} docs/mdoc
 
       - name: Compress target directories
-        run: tar cf targets.tar modules/logging/interop/shapeless/target modules/kernel/target modules/logging/target examples/target modules/concurrent/target modules/logging/layout/target modules/optics/core/target modules/kernelCE2Interop/target modules/doobie/core/target modules/zio/target tofu-docs/target modules/zio/logging/target modules/core3/target modules/kernel/interop/cats-mtl/target modules/config/target modules/optics/macro/target modules/kernelCE3Interop/target modules/zio/core/target modules/streams/target modules/logging/structured/target modules/logging/interop/log4cats/target modules/doobie/logging/target modules/enums/target target modules/fs2/target modules/logging/derivation/target modules/logging/interop/refined/target modules/higherKindCore/target modules/derivation/target modules/core/target modules/observable/target modules/optics/interop/target modules/memo/target modules/env/target project/target
+        run: tar cf targets.tar modules/logging/interop/shapeless/target modules/kernel/target modules/logging/target examples/target modules/doobie/core-ce3/target modules/concurrent/target modules/logging/layout/target modules/optics/core/target modules/kernelCE2Interop/target modules/doobie/core/target modules/zio/target examples-ce3/target tofu-docs/target modules/zio/logging/target modules/core3/target modules/kernel/interop/cats-mtl/target modules/config/target modules/optics/macro/target modules/doobie/logging-ce3/target modules/kernelCE3Interop/target modules/zio/core/target modules/streams/target modules/logging/structured/target modules/logging/interop/log4cats/target modules/doobie/logging/target modules/enums/target target modules/fs2/target modules/logging/derivation/target modules/logging/interop/refined/target modules/higherKindCore/target modules/derivation/target modules/core/target modules/observable/target modules/optics/interop/target modules/memo/target modules/env/target project/target
 
       - name: Upload target directories
         uses: actions/upload-artifact@v2

.scalafix.conf

@@ -1,4 +1,5 @@
 OrganizeImports {
-  groupedImports = Merge
   removeUnused = true
+  groups = ["re:javax?\\.", "scala.", "tofu." "*"]
+  groupedImports = AggressiveMerge
 }

build.sbt

@@ -300,12 +300,28 @@ lazy val doobie = project
 lazy val doobieLogging = project
   .in(file("modules/doobie/logging"))
   .settings(
-    libraryDependencies ++= List(doobieCore),
     defaultSettings,
     name := "tofu-doobie-logging",
   )
   .dependsOn(doobie, loggingStr)
 
+lazy val doobieCE3 = project
+  .in(file("modules/doobie/core-ce3"))
+  .settings(
+    libraryDependencies ++= List(doobieCoreCE3, derevo),
+    defaultSettings,
+    name := "tofu-doobie-ce3",
+  )
+  .dependsOn(core3, derivation)
+
+lazy val doobieLoggingCE3 = project
+  .in(file("modules/doobie/logging-ce3"))
+  .settings(
+    defaultSettings,
+    name := "tofu-doobie-logging-ce3",
+  )
+  .dependsOn(doobieCE3, loggingStr)
+
 lazy val examples = project
   .in(file("examples"))
   .settings(
@@ -317,6 +333,16 @@ lazy val examples = project
   )
   .dependsOn(mainModuleDeps: _*)
 
+lazy val examplesCE3 = project
+  .in(file("examples-ce3"))
+  .settings(
+    libraryDependencies ++= List(doobieCoreCE3, doobieH2CE3, derevo, groovy),
+    defaultSettings,
+    name := "tofu-examples-ce3",
+    noPublishSettings,
+  )
+  .dependsOn(ce3MainModuleDeps: _*)
+
 lazy val streams = project
   .in(file("modules/streams"))
   .settings(
@@ -350,9 +376,15 @@ lazy val ce3CoreModules = Vector(
 lazy val commonModules =
   Vector(observable, opticsInterop, logging, enums, config, zioInterop, fs2Interop, doobie, doobieLogging)
 
+lazy val ce3CommonModules =
+  Vector(loggingStr, loggingDer, loggingLayout, doobieCE3, doobieLoggingCE3)
+
 lazy val allModuleRefs  = (coreModules ++ commonModules).map(x => x: ProjectReference)
 lazy val mainModuleDeps = (coreModules ++ commonModules).map(x => x: ClasspathDep[ProjectReference])
 
+lazy val ce3AllModuleRefs  = (ce3CoreModules ++ ce3CommonModules).map(x => x: ProjectReference)
+lazy val ce3MainModuleDeps = (ce3CoreModules ++ ce3CommonModules).map(x => x: ClasspathDep[ProjectReference])
+
 lazy val docs = project // new documentation project
   .in(file("tofu-docs"))
   .settings(
@@ -375,7 +407,11 @@ lazy val tofu = project
     defaultSettings,
     name := "tofu"
   )
-  .aggregate((coreModules ++ commonModules ++ ce3CoreModules :+ docs :+ examples).map(x => x: ProjectReference): _*)
+  .aggregate(
+    (coreModules ++ commonModules ++ ce3CoreModules ++ ce3CommonModules :+ docs :+ examples :+ examplesCE3).map(x =>
+      x: ProjectReference
+    ): _*
+  )
   .dependsOn(coreModules.map(x => x: ClasspathDep[ProjectReference]): _*)
 
 lazy val defaultScalacOptions = scalacOptions := {

examples-ce3/src/main/resources/logback.groovy

@@ -0,0 +1,19 @@
+import tofu.logging.logback.ConsoleContextLayout
+import tofu.logging.ELKLayout
+
+//puts messages as plain text
+appender("PLAIN-COLORED", ConsoleAppender) {
+    encoder(LayoutWrappingEncoder) {
+        layout(ConsoleContextLayout) {
+            pattern = "%cyan(%d{HH:mm:ss} %-5level %logger{36} - %msg%n [%mdc]%n)"
+        }
+    }
+}
+//puts messages as JSONs
+appender("STRUCTURED", ConsoleAppender) {
+    encoder(LayoutWrappingEncoder) {
+        layout(ELKLayout)
+    }
+}
+
+root(DEBUG, ["PLAIN-COLORED", "STRUCTURED"])

examples-ce3/src/main/scala/tofu/example/doobie/TofuDoobieExample.scala

@@ -0,0 +1,149 @@
+package tofu.example.doobie
+
+import cats.data.ReaderT
+import cats.effect.std.Dispatcher
+import cats.effect.{Async, IO, IOApp, Sync}
+import cats.tagless.syntax.functorK._
+import cats.{Apply, Monad}
+import derevo.derive
+import doobie._
+import doobie.implicits._
+import doobie.util.log.LogHandler
+import tofu.doobie.LiftConnectionIO
+import tofu.doobie.log.{EmbeddableLogHandler, LogHandlerF}
+import tofu.doobie.transactor.Txr
+import tofu.higherKind.derived.representableK
+import tofu.kernel.types.PerformThrow
+import tofu.lift.Lift
+import tofu.logging.derivation.{loggable, loggingMidTry}
+import tofu.logging.{Logging, LoggingCompanion}
+import tofu.syntax.context._
+import tofu.syntax.doobie.log.handler._
+import tofu.syntax.doobie.log.string._
+import tofu.syntax.monadic._
+import tofu.{Delay, Tries, WithContext, WithLocal, WithRun}
+
+import scala.annotation.unused
+
+// Simple context
+@derive(loggable)
+final case class Ctx(traceId: String)
+
+// Model
+@derive(loggable)
+final case class Person(id: Long, name: String, deptId: Long)
+
+@derive(loggable)
+final case class Dept(id: Long, name: String)
+
+// Person SQL algebra
+@derive(representableK, loggingMidTry)
+trait PersonSql[F[_]] {
+  def init: F[Unit]
+  def create(person: Person): F[Unit]
+  def read(id: Long): F[Option[Person]]
+}
+
+object PersonSql extends LoggingCompanion[PersonSql] {
+  def make[DB[_]: Monad: LiftConnectionIO: EmbeddableLogHandler]: PersonSql[DB] = {
+    EmbeddableLogHandler[DB].embedLift(implicit lh => new Impl)
+  }
+
+  final class Impl(implicit lh: LogHandler) extends PersonSql[ConnectionIO] {
+    def init: ConnectionIO[Unit]                     =
+      lsql"create table if not exists person(id int8, name varchar(50), dept_id int8)".update.run.void
+    def create(p: Person): ConnectionIO[Unit]        =
+      lsql"insert into person values(${p.id}, ${p.name}, ${p.deptId})".update.run.void
+    def read(id: Long): ConnectionIO[Option[Person]] =
+      lsql"select id, name, dept_id from person where id = $id"
+        .query[Person]
+        .option
+  }
+}
+
+// Department SQL algebra
+@derive(representableK, loggingMidTry)
+trait DeptSql[F[_]] {
+  def init: F[Unit]
+  def create(dept: Dept): F[Unit]
+  def read(id: Long): F[Option[Dept]]
+}
+
+object DeptSql extends LoggingCompanion[DeptSql] {
+  def make[DB[_]: Monad: LiftConnectionIO: EmbeddableLogHandler]: DeptSql[DB] = {
+    EmbeddableLogHandler[DB].embedLift(implicit lh => new Impl)
+  }
+
+  final class Impl(implicit lh: LogHandler) extends DeptSql[ConnectionIO] {
+    def init: ConnectionIO[Unit]                   =
+      lsql"create table if not exists department(id int8, name varchar(50))".update.run.void
+    def create(d: Dept): ConnectionIO[Unit]        =
+      lsql"insert into department values(${d.id}, ${d.name})".update.run.void
+    def read(id: Long): ConnectionIO[Option[Dept]] =
+      lsql"select id, name from department where id = $id"
+        .query[Dept]
+        .option
+  }
+}
+
+// Storage algebra encapsulates database transactional logic
+@derive(representableK, loggingMidTry)
+trait PersonStorage[F[_]] {
+  def init: F[Unit]
+  def store(person: Person, dept: Dept): F[Unit]
+}
+
+object PersonStorage extends LoggingCompanion[PersonStorage] {
+  def make[F[_]: Apply, DB[_]: Monad: Txr[F, *[_]]](
+      persSql: PersonSql[DB],
+      deptSql: DeptSql[DB]
+  ): PersonStorage[F] = {
+    val impl = new Impl[DB](persSql, deptSql): PersonStorage[DB]
+    val tx   = Txr[F, DB].trans
+    impl.mapK(tx)
+  }
+
+  final class Impl[DB[_]: Monad](persSql: PersonSql[DB], deptSql: DeptSql[DB]) extends PersonStorage[DB] {
+    def init: DB[Unit]                              =
+      deptSql.init >> persSql.init
+    def store(person: Person, dept: Dept): DB[Unit] =
+      deptSql.create(dept) >> persSql.create(person)
+  }
+}
+
+object TofuDoobieExample extends IOApp.Simple {
+
+  val run: IO[Unit] = Dispatcher[IO].use { (disp: Dispatcher[IO]) =>
+    @unused implicit val withDispatcher: WithContext[ReaderT[IO, Ctx, *], Dispatcher[IO]] = WithContext.const(disp)
+    runF[IO, ReaderT[IO, Ctx, *]]
+  }
+
+  def runF[I[_]: Async, F[_]: Sync: PerformThrow: WithRun[*[_], I, Ctx]]: I[Unit] = {
+    // Simplified wiring below
+    implicit val loggingF = Logging.Make.contextual[F, Ctx]
+
+    val transactor   = Transactor.fromDriverManager[I](
+      driver = "org.h2.Driver",
+      url = "jdbc:h2:./test"
+    )
+    implicit val txr = Txr.continuational(transactor.mapK(Lift.trans[I, F]))
+
+    def initStorage[
+        DB[_]: Tries: Txr[F, *[_]]: Delay: Monad: LiftConnectionIO: WithLocal[*[_], Ctx]: PerformThrow
+    ]: PersonStorage[F] = {
+      implicit val loggingDB = Logging.Make.contextual[DB, Ctx]
+
+      implicit val elh = EmbeddableLogHandler.async(LogHandlerF.loggable[DB](Logging.Debug))
+
+      val personSql = PersonSql.make[DB].attachErrLogs
+      val deptSql   = DeptSql.make[DB].attachErrLogs
+
+      PersonStorage.make[F, DB](personSql, deptSql).attachErrLogs
+    }
+
+    val storage = initStorage[txr.DB]
+    val program = storage.init >> storage.store(Person(13L, "Alex", 42L), Dept(42L, "Marketing"))
+    val launch  = runContext(program)(Ctx("715a-562a-4da5-a6e0"))
+    launch
+  }
+}

examples/src/main/scala/tofu/example/doobie/TofuDoobieExample.scala

@@ -1,24 +1,25 @@
 package tofu.example.doobie
 
 import cats.data.ReaderT
-import cats.effect.{ContextShift, Effect, IO, IOApp, Sync}
-import cats.{Apply, Monad}
+import cats.effect.{Async, ContextShift, IO, IOApp, Sync}
 import cats.tagless.syntax.functorK._
+import cats.{Apply, Monad}
 import derevo.derive
 import doobie._
 import doobie.implicits._
 import doobie.util.log.LogHandler
+import tofu.doobie.LiftConnectionIO
 import tofu.doobie.log.{EmbeddableLogHandler, LogHandlerF}
 import tofu.doobie.transactor.Txr
-import tofu.doobie.LiftConnectionIO
 import tofu.higherKind.derived.representableK
-import tofu.lift.{Lift, UnliftIO}
+import tofu.kernel.types.PerformThrow
+import tofu.lift.Lift
 import tofu.logging.derivation.{loggable, loggingMidTry}
-import tofu.logging.{Logging, LoggingCompanion, Logs}
+import tofu.logging.{Logging, LoggingCompanion}
 import tofu.syntax.context._
-import tofu.syntax.monadic._
 import tofu.syntax.doobie.log.handler._
 import tofu.syntax.doobie.log.string._
+import tofu.syntax.monadic._
 import tofu.{Delay, Tries, WithLocal, WithRun}
 
 // Simple context
@@ -108,12 +109,11 @@ object PersonStorage extends LoggingCompanion[PersonStorage] {
 }
 
 object TofuDoobieExample extends IOApp.Simple {
-  val run: IO[Unit] =
-    runF[IO, ReaderT[IO, Ctx, *]]
+  val run: IO[Unit] = runF[IO, ReaderT[IO, Ctx, *]]
 
-  def runF[I[_]: Effect: ContextShift, F[_]: Sync: UnliftIO: WithRun[*[_], I, Ctx]]: I[Unit] = {
+  def runF[I[_]: Async: ContextShift, F[_]: Sync: PerformThrow: WithRun[*[_], I, Ctx]]: I[Unit] = {
     // Simplified wiring below
-    implicit val loggingF = Logs.contextual[F, Ctx]
+    implicit val loggingF = Logging.Make.contextual[F, Ctx]
 
     val transactor   = Transactor.fromDriverManager[I](
       driver = "org.h2.Driver",
@@ -122,11 +122,11 @@ object TofuDoobieExample extends IOApp.Simple {
     implicit val txr = Txr.continuational(transactor.mapK(Lift.trans[I, F]))
 
     def initStorage[
-        DB[_]: Tries: Txr[F, *[_]]: Delay: Monad: LiftConnectionIO: WithLocal[*[_], Ctx]: UnliftIO
+        DB[_]: Tries: Txr[F, *[_]]: Delay: Monad: LiftConnectionIO: WithLocal[*[_], Ctx]: PerformThrow
     ]: PersonStorage[F] = {
-      implicit val loggingDB = Logs.contextual[DB, Ctx]
+      implicit val loggingDB = Logging.Make.contextual[DB, Ctx]
 
-      implicit val elh = EmbeddableLogHandler.sync(LogHandlerF.loggable[DB](Logging.Debug))
+      implicit val elh = EmbeddableLogHandler.async(LogHandlerF.loggable[DB](Logging.Debug))
 
       val personSql = PersonSql.make[DB].attachErrLogs
       val deptSql   = DeptSql.make[DB].attachErrLogs

modules/core3/src/test/scala/tofu/concurrent/PerformSuite.scala

@@ -0,0 +1,67 @@
+package tofu.concurrent
+
+import scala.annotation.unchecked.uncheckedVariance
+import scala.concurrent.duration._
+
+import cats.Monad
+import cats.data.ReaderT
+import cats.effect.std.Dispatcher
+import cats.effect.{IO, Resource}
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.must.Matchers
+import tofu.generate.GenRandom
+import tofu.syntax.monadic._
+import tofu.syntax.scoped._
+import tofu.syntax.time._
+import tofu.time.Sleep
+import tofu.{Execute, PerformThrow, WithContext}
+
+case class MyContext(dispatcher: Dispatcher[IO], traceId: Long)
+object MyContext {
+  type Reader[+A] = ReaderT[IO, MyContext, A @uncheckedVariance]
+  implicit val withDispatcher: WithContext[Reader, Dispatcher[IO]] =
+    WithContext[Reader, MyContext].extract(_.dispatcher)
+}
+
+class PerformSuite extends AnyFunSuite with Matchers {
+  type Eff[+A] = MyContext.Reader[A]
+
+  val p = PerformThrow[Eff]
+
+  val genTraceId = for {
+    random  <- GenRandom.instance[IO, IO]()
+    traceId <- random.nextLong
+  } yield traceId
+
+  val init = for {
+    dispatcher <- Dispatcher[IO]
+    traceId    <- Resource.eval(genTraceId)
+  } yield MyContext(dispatcher, traceId)
+
+  def waitUpdate[F[_]: Sleep: Monad](atom: Atom[F, Int]): F[Int] = for {
+    v <- atom.get
+    _ <- sleep(100.millis)
+    _ <- atom.update(_ + 1)
+  } yield v
+
+  def program[F[_]: PerformThrow: Monad: Execute, A](fa: F[A]) = for {
+    performer <- PerformThrow[F].performer
+    (fut1, _)  = performer.toFuture(fa)
+    (fut2, _)  = performer.toFuture(fa)
+    (fut3, _)  = performer.toFuture(fa)
+    res1      <- deferFuture(_ => fut1)
+    res2      <- deferFuture(_ => fut2)
+    res3      <- deferFuture(_ => fut3)
+  } yield List(res1, res2, res3)
+
+  val run = for {
+    atom <- MakeAtom[Eff, Eff].of(1)
+    ress <- program(waitUpdate(atom))
+  } yield ress
+
+  test("parallize over performed futures") {
+    import cats.effect.unsafe.implicits.global
+    val ts = init.use(run.run).unsafeRunSync()
+    all(ts) mustEqual 1
+  }
+}