IC-481 Migrates to cats-effect 3

fernandomora · Darren8098 · commit 379bd3046437 · 2021-11-19T09:04:16.000+01:00
diff --git a/.gitignore b/.gitignore
@@ -1,48 +1,4 @@
-# javascript specific
-node_modules
-
-# scala specific
-*.class
-*.log
-
-# sbt specific
-.cache
-.history
-.lib/
-dist/*
-target/
-lib_managed/
-src_managed/
-project/boot/
-project/plugins/project/
-
-# intellij
+.bsp/
 .idea/
-*.ipr
-*.iml
-
-# Scala IDE specific
-.scala_dependencies
-.worksheet
-
-# Visual Studio Code specific
-.bloop/
-.metals/
-project/metals.sbt
-.history/
-.vscode/
-
-# environment specific
-.envrc
-
-# Mac files
-.DS_Store
-
-
-# deployment secrets
-devel.secrets.json
-/dashboard/certs/
-/docker/dashboard/dependencies/certs
-
-build.sbt.repl
-build.sbt.original
+project/target/
+target/
diff --git a/build.sbt b/build.sbt
@@ -6,9 +6,11 @@ ThisBuild / homepage := Some(url("https://github.com/intenthq/action-processor-i
 ThisBuild / developers := List(Developer("intenthq", "Intent HQ", null, url("https://www.intenthq.com/")))
 ThisBuild / licenses := Seq(("MIT", url("http://opensource.org/licenses/MIT")))
 
-ThisBuild / scalaVersion := "2.13.3"
+ThisBuild / scalaVersion := "2.13.7"
+ThisBuild / semanticdbEnabled := true
+ThisBuild / semanticdbVersion := scalafixSemanticdb.revision
 
-addCompilerPlugin("org.typelevel" % "kind-projector" % "0.11.0" cross CrossVersion.full)
+addCompilerPlugin("org.typelevel" % "kind-projector" % "0.13.2" cross CrossVersion.full)
 
 lazy val root = (project in file("."))
   .settings(
@@ -18,17 +20,16 @@ lazy val root = (project in file("."))
     testFrameworks += new TestFramework("weaver.framework.CatsEffect"),
     scalafmtOnCompile := true,
     libraryDependencies ++= Seq(
-      "co.fs2" %% "fs2-core" % "2.5.5",
-      "co.fs2" %% "fs2-io" % "2.5.5",
-      "com.google.guava" % "guava" % "30.0-jre",
-      "com.propensive" %% "magnolia" % "0.17.0",
+      "co.fs2" %% "fs2-core" % "3.2.2",
+      "co.fs2" %% "fs2-io" % "3.2.2",
+      "com.softwaremill.magnolia1_2" %% "magnolia" % "1.0.0-M7",
       "de.siegmar" % "fastcsv" % "1.0.3",
       "org.mapdb" % "mapdb" % "3.0.8",
-      "org.tpolecat" %% "doobie-core" % "0.12.1",
-      "org.tpolecat" %% "doobie-hikari" % "0.12.1",
-      "com.disneystreaming" %% "weaver-cats" % "0.6.2" % Test,
-      "com.disneystreaming" %% "weaver-core" % "0.6.2" % Test,
-      "org.tpolecat" %% "doobie-h2" % "0.12.1" % Test
+      "org.tpolecat" %% "doobie-core" % "1.0.0-RC1",
+      "org.tpolecat" %% "doobie-hikari" % "1.0.0-RC1",
+      "com.disneystreaming" %% "weaver-cats" % "0.7.7" % Test,
+      "com.disneystreaming" %% "weaver-core" % "0.7.7" % Test,
+      "org.tpolecat" %% "doobie-h2" % "1.0.0-RC1" % Test
     ),
     /*
     https://github.com/sbt/sbt/issues/3249#issuecomment-534757714
diff --git a/project/build.properties b/project/build.properties
@@ -1 +1 @@
-sbt.version=1.3.13
+sbt.version=1.5.5
diff --git a/project/plugins.sbt b/project/plugins.sbt
@@ -2,3 +2,4 @@ addSbtPlugin("com.geirsson" % "sbt-ci-release" % "1.5.3")
 addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.5.1")
 addSbtPlugin("io.github.davidgregory084" % "sbt-tpolecat" % "0.1.13")
 addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.2")
+addSbtPlugin("ch.epfl.scala" % "sbt-scalafix" % "0.9.32")
diff --git a/src/main/scala/com/intenthq/StreamOps.scala b/src/main/scala/com/intenthq/StreamOps.scala
@@ -1,12 +1,11 @@
 package com.intenthq
 
-import cats.effect.{Concurrent, Sync, Timer}
+import cats.effect.{Async, Concurrent, Sync, Temporal}
 import cats.implicits._
 import fs2.{Chunk, Stream}
 
 import scala.concurrent.duration.FiniteDuration
 import StreamOps._
-import cats.Parallel
 
 class StreamChunkOps[F[_], O](s: fs2.Stream[F, Chunk[O]]) {
   private val parallelismEnabled: Boolean = scala.util.Properties.envOrNone("FS2_PARALLELISM").map(_.toBoolean).getOrElse(true)
@@ -25,37 +24,37 @@ class StreamChunkOps[F[_], O](s: fs2.Stream[F, Chunk[O]]) {
       s.parEvalMap[F2, Stream[F2, O2]](parallelism)(chunk => f(chunk).map(fs2.Stream.chunk)).flatten
 }
 
-class StreamOps[F[_]: Sync, O](s: fs2.Stream[F, O]) {
+class StreamOps[F[_], O](s: fs2.Stream[F, O]) {
 
   def debugEvery(
     d: FiniteDuration
-  )(formatter: O => String = _.toString, logger: String => Unit = println(_))(implicit C: Concurrent[F], T: Timer[F]): Stream[F, O] =
+  )(formatter: O => String = _.toString, logger: String => Unit = println(_))(implicit C: Temporal[F]): Stream[F, O] =
     s.observe(_.debounce(d).debug(formatter, logger).drain)
 
   object preservingChunks {
     // Parallelizes different chunks processing but not elements of same chunk (traverse instead of parTraverse)
-    def parEvalMap[F2[x] >: F[x]: Concurrent: Parallel, O2](parallelism: Int)(f: O => F2[O2]): Stream[F2, O2] =
+    def parEvalMap[F2[x] >: F[x]: Concurrent, O2](parallelism: Int)(f: O => F2[O2]): Stream[F2, O2] =
       s.chunks.parEvalMapChunks[F2, O2](parallelism)(chunk => chunk.traverse(f))
 
     // Alias for parallezing non-effectful operations
-    def parMap[O2](parallelism: Int)(f: O => O2)(implicit C: Concurrent[F], P: Parallel[F]): Stream[F, O2] =
+    def parMap[O2](parallelism: Int)(f: O => O2)(implicit A: Async[F]): Stream[F, O2] =
       preservingChunks.parEvalMap(parallelism)(o => Sync[F].delay(f(o)))
 
     // Parallelizes different chunks processing but not elements of same chunk (traverse instead of parTraverse)
-    def parEvalMapUnordered[F2[x] >: F[x]: Concurrent: Parallel, O2](parallelism: Int)(f: O => F2[O2]): Stream[F2, O2] =
+    def parEvalMapUnordered[F2[x] >: F[x]: Concurrent, O2](parallelism: Int)(f: O => F2[O2]): Stream[F2, O2] =
       s.chunks.parEvalMapChunksUnordered[F2, O2](parallelism)(chunk => chunk.traverse(f))
 
     // Alias for parallezing non-effectful operations
-    def parMapUnordered[O2](parallelism: Int)(f: O => O2)(implicit C: Concurrent[F], P: Parallel[F]): Stream[F, O2] =
+    def parMapUnordered[O2](parallelism: Int)(f: O => O2)(implicit A: Async[F]): Stream[F, O2] =
       preservingChunks.parEvalMapUnordered(parallelism)(o => Sync[F].delay(f(o)))
   }
 }
 
 trait ToStreamOps {
-  implicit def toStreamOps[F[_]: Sync, O](fo: Stream[F, O]): StreamOps[F, O] =
+  implicit def toStreamOps[F[_], O](fo: Stream[F, O]): StreamOps[F, O] =
     new StreamOps(fo)
 
-  implicit def toStreamChunkOps[F[_]: Sync, O](fo: fs2.Stream[F, Chunk[O]]): StreamChunkOps[F, O] =
+  implicit def toStreamChunkOps[F[_], O](fo: fs2.Stream[F, Chunk[O]]): StreamChunkOps[F, O] =
     new StreamChunkOps(fo)
 }
 
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/aggregations/Aggregate.scala b/src/main/scala/com/intenthq/action_processor/integrations/aggregations/Aggregate.scala
@@ -1,28 +1,21 @@
 package com.intenthq.action_processor.integrations.aggregations
 
-import java.util.concurrent.ConcurrentMap
-
-import scala.jdk.CollectionConverters._
-
-import cats.effect.{Blocker, ContextShift, IO, Resource, SyncIO}
-
+import cats.effect.{IO, Resource}
 import com.intenthq.action_processor.integrations.config.MapDbSettings
 import com.intenthq.action_processor.integrations.feeds.FeedContext
 import com.intenthq.action_processor.integrations.repositories.MapDBRepository
-
-import org.mapdb.{DataInput2, DataOutput2, HTreeMap, Serializer}
 import org.mapdb.elsa.{ElsaMaker, ElsaSerializer}
 import org.mapdb.serializer.GroupSerializerObjectArray
+import org.mapdb.{DataInput2, DataOutput2, HTreeMap, Serializer}
 
-object Aggregate {
+import java.util.concurrent.ConcurrentMap
+import scala.jdk.CollectionConverters._
 
-  private lazy val blocker = Blocker[SyncIO].allocated.unsafeRunSync()._1
-  private val ec = scala.concurrent.ExecutionContext.global
-  implicit private val contextShift: ContextShift[IO] = IO.contextShift(ec)
+object Aggregate {
 
   def noop[I]: fs2.Pipe[IO, I, (I, Long)] = _.map(_ -> 1L)
 
-  private def loadAggRepository[K](mapDbSettings: MapDbSettings)(blocker: Blocker): Resource[IO, HTreeMap[K, Long]] = {
+  private def loadAggRepository[K](mapDbSettings: MapDbSettings): Resource[IO, HTreeMap[K, Long]] = {
 
     val serializer: Serializer[K] = new GroupSerializerObjectArray[K] {
       val elsaSerializer: ElsaSerializer = new ElsaMaker().make
@@ -31,7 +24,7 @@ object Aggregate {
     }
 
     MapDBRepository
-      .load(mapDbSettings)(blocker)
+      .load(mapDbSettings)
       .map(db =>
         db.hashMap("stuff", serializer, Serializer.LONG.asInstanceOf[Serializer[Long]])
           .layout(mapDbSettings.segments, mapDbSettings.nodeSize, mapDbSettings.levels)
@@ -46,9 +39,9 @@ object Aggregate {
       val aggregateInRepository: fs2.Pipe[IO, I, ConcurrentMap[K, Long]] =
         in => {
           fs2.Stream
-            .resource[IO, ConcurrentMap[K, Long]](loadAggRepository(feedContext.mapDbSettings)(blocker))
+            .resource[IO, ConcurrentMap[K, Long]](loadAggRepository(feedContext.mapDbSettings))
             .flatMap { aggRepository =>
-              fs2.Stream.eval_(IO.delay(println("Starting aggregation"))) ++
+              fs2.Stream.exec(IO.delay(println("Starting aggregation"))) ++
                 in.evalMapChunk { o =>
                   IO.delay {
                     keys(o).foreach { value =>
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/feeds/LocalFileCSVFeed.scala b/src/main/scala/com/intenthq/action_processor/integrations/feeds/LocalFileCSVFeed.scala
@@ -1,23 +1,20 @@
 package com.intenthq.action_processor.integrations.feeds
 
-import java.nio.file.Paths
-
-import cats.effect.{Blocker, ContextShift, IO}
+import cats.effect.IO
+import fs2.io.file.{Files, Flags, Path}
 import fs2.text
 
+import java.nio.file.Paths
 import scala.util.Properties
 
 trait LocalFileCSVFeed[O] extends CSVFeed[O] {
 
-  implicit protected val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
   protected val localFilePath: String = Properties.envOrElse("CSV_RESOURCE", "/data.csv")
 
   override protected def rows: fs2.Stream[IO, String] =
-    fs2.Stream.resource(Blocker[IO]).flatMap { blocker =>
-      fs2.io.file
-        .readAll[IO](Paths.get(localFilePath), blocker, 4096)
-        .through(text.utf8Decode)
-        .through(text.lines)
-        .dropLastIf(_.isEmpty)
-    }
+    Files[IO]
+      .readAll(Path.fromNioPath(Paths.get(localFilePath)), 4096, Flags.Read)
+      .through(text.utf8.decode)
+      .through(text.lines)
+      .dropLastIf(_.isEmpty)
 }
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/feeds/SQLFeed.scala b/src/main/scala/com/intenthq/action_processor/integrations/feeds/SQLFeed.scala
@@ -1,6 +1,6 @@
 package com.intenthq.action_processor.integrations.feeds
 
-import cats.effect.{ContextShift, IO}
+import cats.effect.IO
 import com.intenthq.action_processor.integrations.DoobieImplicits
 import doobie.util.query.Query0
 import doobie.util.transactor.Transactor
@@ -12,8 +12,6 @@ trait SQLFeed[I, O] extends Feed[I, O] with DoobieImplicits {
   protected val jdbcUrl: String
   protected def query(sourceContext: FeedContext[IO]): Query0[I]
 
-  implicit private val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
-
   protected lazy val transactor: Transactor[IO] = createTransactor
   protected val chunkSize: Int = doobie.util.query.DefaultChunkSize
 
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/repositories/MapDBRepository.scala b/src/main/scala/com/intenthq/action_processor/integrations/repositories/MapDBRepository.scala
@@ -4,17 +4,17 @@ import java.io.File
 import java.nio.file.{Files, Paths}
 import java.time.LocalDateTime
 
-import cats.effect.{Blocker, ContextShift, IO, Resource}
+import cats.effect.{IO, Resource}
 import com.intenthq.action_processor.integrations.config.MapDbSettings
 import org.mapdb.{DB, DBMaker}
 
 object MapDBRepository {
 
-  def load(mapDBSettings: MapDbSettings)(blocker: Blocker)(implicit cs: ContextShift[IO]): Resource[IO, DB] = {
+  def load(mapDBSettings: MapDbSettings): Resource[IO, DB] = {
     val dbInitOp = for {
       now <- IO.delay(LocalDateTime.now())
       dbFile <- IO.delay(new File(Paths.get(mapDBSettings.dbPath.toString, s"dbb-${now.toLocalDate}-${now.toLocalTime}").toUri))
-      createDb <- IO.delay {
+      createDb <- IO.blocking {
         DBMaker
           .fileDB(dbFile.getAbsolutePath)
           .allocateStartSize(mapDBSettings.startDbSize)
@@ -26,6 +26,6 @@ object MapDBRepository {
           .make()
       }
     } yield (createDb, dbFile)
-    Resource.make(blocker.blockOn(dbInitOp))(db => IO.delay(db._1.close()).guarantee(IO.delay(Files.deleteIfExists(db._2.toPath)).void)).map(_._1)
+    Resource.make(dbInitOp)(db => IO.delay(db._1.close()).guarantee(IO.delay(Files.deleteIfExists(db._2.toPath)).void)).map(_._1)
   }
 }
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/serializations/csv/Csv.scala b/src/main/scala/com/intenthq/action_processor/integrations/serializations/csv/Csv.scala
@@ -2,7 +2,7 @@ package com.intenthq.action_processor.integrations.serializations.csv
 
 import java.time._
 
-import magnolia._
+import magnolia1._
 
 trait Csv[A] {
   def toCSV(a: A): Seq[String]
@@ -13,9 +13,9 @@ object Csv {
 
   type Typeclass[A] = Csv[A]
 
-  def combine[A](ctx: CaseClass[Csv, A]): Csv[A] = (a: A) => ctx.parameters.flatMap(p => p.typeclass.toCSV(p.dereference(a)))
+  def join[A](ctx: CaseClass[Csv, A]): Csv[A] = (a: A) => ctx.parameters.flatMap(p => p.typeclass.toCSV(p.dereference(a)))
 
-  def dispatch[A](ctx: SealedTrait[Csv, A]): Csv[A] = (a: A) => ctx.dispatch(a)(sub => sub.typeclass.toCSV(sub.cast(a)))
+  def split[A](ctx: SealedTrait[Csv, A]): Csv[A] = (a: A) => ctx.split(a)(sub => sub.typeclass.toCSV(sub.cast(a)))
 
   implicit def csvOpt[T: Csv]: Csv[Option[T]] = (a: Option[T]) => a.fold(Seq(""))(Csv[T].toCSV)
   implicit def csvIterable[T: Csv]: Csv[Iterable[T]] = (a: Iterable[T]) => Seq(a.map(Csv[T].toCSV).mkString(","))
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/CSVFeedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/CSVFeedSpec.scala
@@ -1,16 +1,15 @@
 package com.intenthq.action_processor.integrations
 
-import java.io.{File, FileWriter}
-import java.nio.charset.StandardCharsets
-import java.nio.file.Files
-
 import cats.effect.{IO, Resource}
-import cats.implicits._
 import com.intenthq.action_processor.integrations.aggregations.NoAggregate
 import com.intenthq.action_processor.integrations.feeds.LocalFileCSVFeed
 import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
 import weaver.IOSuite
 
+import java.io.{File, FileWriter}
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+
 object CSVFeedSpec extends IOSuite with CSVFeedSpecResources {
 
   override val csvFeedContent: String =
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala
@@ -2,21 +2,19 @@ package com.intenthq.action_processor.integrations
 
 import java.time.temporal.ChronoUnit
 import java.time.{Instant, LocalDate, LocalTime}
-
-import cats.effect.{Async, Blocker, Bracket, ContextShift, IO, Resource}
+import cats.effect.{Async, IO, MonadCancel, Resource}
 import cats.implicits.toFunctorOps
-
 import com.intenthq.action_processor.integrations.aggregations.NoAggregate
 import com.intenthq.action_processor.integrations.feeds.{FeedContext, SQLFeed}
 import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
-
 import doobie.h2.H2Transactor
 import doobie.implicits.{toConnectionIOOps, toSqlInterpolator}
 import doobie.util.query.Query0
 import doobie.util.transactor.Transactor
 import doobie.util.update.Update
 import doobie.implicits.javatimedrivernative._
 import weaver.IOSuite
+
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
 
 object SQLFeedSpec extends IOSuite with SQLFeedSpecResources {
@@ -52,19 +50,14 @@ trait SQLFeedSpecResources { self: IOSuite =>
 
   protected val exampleRows: Seq[ExampleCsvFeedRow]
 
-  override def sharedResource: Resource[IO, Res] =
-    for {
-      blocker <- Blocker[IO]
-      // Given a transactor for an exanpleRows fixture populated database
-      transactor <- transactorResource[IO](blocker, ExecutionContext.global)
-    } yield transactor
+  override def sharedResource: Resource[IO, Res] = transactorResource[IO](ExecutionContext.global)
 
-  private def transactorResource[F[_]: Async: ContextShift](blocker: Blocker, ec: ExecutionContextExecutor): Resource[F, H2Transactor[F]] = {
-    val transactor = H2Transactor.newH2Transactor[F]("jdbc:h2:mem:test;DB_CLOSE_DELAY=-1", "sa", "", ec, blocker)
+  private def transactorResource[F[_]: Async](ec: ExecutionContextExecutor): Resource[F, H2Transactor[F]] = {
+    val transactor = H2Transactor.newH2Transactor[F]("jdbc:h2:mem:test;DB_CLOSE_DELAY=-1", "sa", "", ec)
     transactor.evalTap(insertDatabaseFixtures[F])
   }
 
-  private def insertDatabaseFixtures[F[_]: Bracket[*[_], Throwable]](transactor: Transactor[F]): F[Unit] = {
+  private def insertDatabaseFixtures[F[_]: MonadCancel[*[_], Throwable]](transactor: Transactor[F]): F[Unit] = {
     val createTable: doobie.ConnectionIO[Unit] =
       sql"""CREATE TABLE example(
            |  integer INTEGER NOT NULL,

Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-sbt.version=1.3.13`
	`1`	`+sbt.version=1.5.5`