Migrate code to ce3

Author: VladKopanievITV

core/src/main/scala/com.itv/servicebox/algebra/InMemoryServiceRegistry.scala

@@ -2,24 +2,20 @@ package com.itv.servicebox.algebra
 
 import java.net.Socket
 import java.util.concurrent.atomic.AtomicReference
-
 import cats.{Applicative, MonadError}
 import cats.data.{NonEmptyList, StateT}
-import cats.effect.Effect
+import cats.effect.kernel.Sync
 import cats.instances.list._
 import cats.instances.map._
 import cats.instances.set._
-import cats.instances.stream._
 import cats.syntax.all._
 
-import scala.util.Try
-
 object InMemoryServiceRegistry {
   val DefaultPortRange = 49162 to 49262
 }
 
 class InMemoryServiceRegistry[F[_]](range: Range, logger: Logger[F])(implicit tag: AppTag,
-                                                                     E: Effect[F],
+                                                                     S: Sync[F],
                                                                      M: MonadError[F, Throwable])
     extends ServiceRegistry[F](logger) {
 
@@ -31,7 +27,7 @@ class InMemoryServiceRegistry[F[_]](range: Range, logger: Logger[F])(implicit ta
 
   private def updateRange(n: Int)(implicit A: Applicative[F]): PortAllocation[F, Unit] =
     StateT.modifyF[F, AtomicReference[Range]] { ref =>
-      E.delay(ref.getAndUpdate(_.drop(n))) *> A.pure(ref)
+      S.delay(ref.getAndUpdate(_.drop(n))) *> A.pure(ref)
     }
 
   private def lift[F[_]: Applicative, A](fa: F[A]): PortAllocation[F, A] =
@@ -65,7 +61,7 @@ class InMemoryServiceRegistry[F[_]](range: Range, logger: Logger[F])(implicit ta
     } yield registered
 
   private def checkPortNotBound(port: Int): F[Boolean] =
-    E.delay {
+    S.delay {
         new Socket("localhost", port).close()
       }
       .attempt
@@ -74,7 +70,7 @@ class InMemoryServiceRegistry[F[_]](range: Range, logger: Logger[F])(implicit ta
   override def register(service: Service.Spec[F]) =
     for {
       registeredContainers <- service.containers.toList
-        .traverse[PortAllocation[F, ?], Container.Registered](c => allocate(c, service.ref))
+        .traverse[PortAllocation[F, *], Container.Registered](c => allocate(c, service.ref))
         .runA(portRange)
 
       err = ServiceRegistry.EmptyPortList(registeredContainers.map(_.ref))
@@ -99,15 +95,15 @@ class InMemoryServiceRegistry[F[_]](range: Range, logger: Logger[F])(implicit ta
       portMappings = rs.containers.foldMap(c => Map(c.ref -> c.portMappings))
 
       _ <- logger.debug(s"registering containers with port ranges:\n\t$summary")
-      _ <- E.delay(registry.getAndUpdate(_.updated(rs.ref, portMappings)))
+      _ <- S.delay(registry.getAndUpdate(_.updated(rs.ref, portMappings)))
 
     } yield rs
 
   override def deregister(id: Service.Ref) =
-    E.delay(registry.getAndUpdate(_ - id)).void
+    S.delay(registry.getAndUpdate(_ - id)).void
 
   override def deregister(id: Service.Ref, cRef: Container.Ref) =
-    E.delay(registry.getAndUpdate { m =>
+    S.delay(registry.getAndUpdate { m =>
         m.get(id)
           .map { mappings =>
             m.updated(id, mappings - cRef)
@@ -117,10 +113,10 @@ class InMemoryServiceRegistry[F[_]](range: Range, logger: Logger[F])(implicit ta
       .void
 
   override def lookup(id: Service.Ref) =
-    E.delay(registry.get).map(_.get(id).filter(_.nonEmpty))
+    S.delay(registry.get).map(_.get(id).filter(_.nonEmpty))
 
   override def updatePortMappings(id: Service.Ref, cId: Container.Ref, mappings: Set[PortMapping]) =
-    E.delay(registry.getAndUpdate { data =>
+    S.delay(registry.getAndUpdate { data =>
         val updated = data
           .get(id)
           .map { m =>

core/src/main/scala/com.itv/servicebox/algebra/Logger.scala

@@ -1,7 +1,5 @@
 package com.itv.servicebox.algebra
 
-import cats.MonadError
-
 abstract class Logger[F[_]] {
   def debug(msg: String): F[Unit]
   def info(msg: String): F[Unit]

core/src/main/scala/com.itv/servicebox/algebra/Runner.scala

@@ -2,18 +2,17 @@ package com.itv.servicebox.algebra
 
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicReference
-
 import cats.MonadError
 import cats.syntax.all._
 import cats.instances.all._
 import Service._
-import cats.effect.Effect
+import cats.effect.Sync
 
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration.FiniteDuration
 
 class Runner[F[_]](srvCtrl: ServiceController[F], networkCtrl: NetworkController[F], registry: ServiceRegistry[F])(
-    serviceSeq: Spec[F]*)(implicit M: MonadError[F, Throwable], E: Effect[F], tag: AppTag) {
+    serviceSeq: Spec[F]*)(implicit S: Sync[F], M: MonadError[F, Throwable], tag: AppTag) {
 
   private val services           = serviceSeq.toList
   private val tearedDownServices = new AtomicReference[Set[Service.Ref]](Set.empty[Service.Ref])
@@ -50,7 +49,7 @@ class Runner[F[_]](srvCtrl: ServiceController[F], networkCtrl: NetworkController
 
     for {
       sortedRefs <- Dag(incomingEdges).topologicalSort.map(_.reverse)
-      sortedSpecs <- sortedRefs.traverse[Either[Throwable, ?], Service.Spec[F]](ref =>
+      sortedSpecs <- sortedRefs.traverse[Either[Throwable, *], Service.Spec[F]](ref =>
         servicesByRef.get(ref).toRight(refNotFound(ref)))
 
     } yield sortedSpecs
@@ -68,11 +67,11 @@ class Runner[F[_]](srvCtrl: ServiceController[F], networkCtrl: NetworkController
 
   private def setupWithRuntimeInfo(spec: Spec[F])(implicit ec: ExecutionContext): F[(Registered[F], RuntimeInfo)] =
     for {
-      t1         <- E.delay(System.currentTimeMillis())
+      t1         <- S.delay(System.currentTimeMillis())
       registered <- srvCtrl.start(spec)
-      t2         <- E.delay(System.currentTimeMillis())
+      t2         <- S.delay(System.currentTimeMillis())
       _          <- srvCtrl.waitUntilReady(registered)
-      t3         <- E.delay(System.currentTimeMillis())
+      t3         <- S.delay(System.currentTimeMillis())
     } yield {
       val setupTime      = FiniteDuration(t2 - t1, TimeUnit.MILLISECONDS)
       val readyCheckTime = FiniteDuration(t3 - t2, TimeUnit.MILLISECONDS)
@@ -81,15 +80,15 @@ class Runner[F[_]](srvCtrl: ServiceController[F], networkCtrl: NetworkController
 
   private def tearDownOnce(spec: Spec[F]): F[Unit] =
     for {
-      alreadyDone     <- E.delay(tearedDownServices.get())
+      alreadyDone     <- S.delay(tearedDownServices.get())
       maybeRegistered <- registry.lookup(spec)
       _ <- maybeRegistered.filterNot(srv => alreadyDone(srv.ref)).fold(M.unit) { srv =>
-        srvCtrl.tearDown(srv) *> E.delay(tearedDownServices.getAndUpdate(_ + srv.ref))
+        srvCtrl.tearDown(srv) *> S.delay(tearedDownServices.getAndUpdate(_ + srv.ref))
       }
-      _ <- E
+      _ <- S
         .delay(tearedDownServices.get)
         .map(_ == services.map(_.ref).toSet)
-        .ifM(E.delay(Thread.sleep(2000)) *> networkCtrl.removeNetwork, E.unit)
+        .ifM(S.delay(Thread.sleep(2000)) *> networkCtrl.removeNetwork, S.unit)
 
     } yield ()
 }

core/src/main/scala/com.itv/servicebox/algebra/package.scala

@@ -3,9 +3,9 @@ package com.itv.servicebox
 import java.nio.file.{Files, Path}
 import java.util.UUID
 import java.util.concurrent.atomic.AtomicReference
-
 import cats.data.{NonEmptyList, StateT}
-import cats.effect.{Effect, IO}
+import cats.derived.semiauto
+import cats.effect.{IO, Sync}
 import cats.instances.all._
 import cats.syntax.all._
 import cats.{ApplicativeError, Eq, Monad, Show}
@@ -39,21 +39,21 @@ package object algebra {
     }
 
     def fromTmpFileContent[F[_]](baseDir: Path)(to: Path, ro: Boolean = false)(
-        files: (String, Array[Byte])*)(implicit E: Effect[F], M: Monad[F]): F[BindMount] = {
+        files: (String, Array[Byte])*)(implicit S: Sync[F], M: Monad[F]): F[BindMount] = {
 
       val mountDir = baseDir.resolve(UUID.randomUUID().toString)
 
       for {
 
-        _ <- E
+        _ <- S
           .delay(baseDir.toFile.exists())
-          .ifM(E.unit, E.delay(baseDir.toFile.mkdirs()).void)
+          .ifM(S.unit, S.delay(baseDir.toFile.mkdirs()).void)
 
-        _ <- E.delay(Files.createDirectory(mountDir))
+        _ <- S.delay(Files.createDirectory(mountDir))
 
         _ <- files.toList.traverse_ {
           case (fileName, content) =>
-            E.delay(Files.write(mountDir.resolve(fileName), content))
+            S.delay(Files.write(mountDir.resolve(fileName), content))
         }
       } yield BindMount(mountDir, to, ro)
     }
@@ -135,7 +135,8 @@ package object algebra {
     }
 
     object Spec {
-      implicit val eq: Eq[Spec] = cats.derived.semi.eq[Spec]
+
+      implicit val eqSpec: Eq[Spec] = semiauto.eq[Spec]
     }
 
     case class Registered(ref: Container.Ref,
@@ -435,6 +436,7 @@ package object algebra {
   }
 
   object UnsafeBlocking {
+    import cats.effect.unsafe.implicits._
     implicit val ioRunSync: UnsafeBlocking[IO] = new UnsafeBlocking[IO] {
       override def runSync[A](io: IO[A]): A = io.unsafeRunSync()
     }

core/src/main/scala/com/itv/servicebox/interpreter/package.scala

@@ -1,9 +1,10 @@
 package com.itv.servicebox
 
 import java.util.concurrent.{TimeUnit, TimeoutException}
-
-import cats.effect.{ContextShift, IO, Timer}
+import cats.effect.IO
+import cats.effect.kernel.Clock
 import cats.syntax.apply._
+
 import scala.concurrent.duration._
 import com.itv.servicebox.algebra._
 
@@ -13,15 +14,11 @@ import scala.concurrent.duration.FiniteDuration
 package object interpreter {
   implicit val ioLogger: Logger[IO] = IOLogger
 
-  implicit def ioScheduler(implicit logger: Logger[IO]): Scheduler[IO] = new Scheduler[IO](logger) {
+  implicit def ioScheduler(implicit logger: Logger[IO], clock: Clock[IO]): Scheduler[IO] = new Scheduler[IO](logger) {
     override def retry[A](f: () => IO[A], checkTimeout: FiniteDuration, totalTimeout: FiniteDuration, label: String)(implicit ec: ExecutionContext): IO[A] = {
 
-      implicit val timer: Timer[IO] = IO.timer(ec)
-      implicit val cs: ContextShift[IO] = IO.contextShift(ec)
-
-      def currentTimeMs = timer.clock.monotonic(TimeUnit.MILLISECONDS).map(FiniteDuration(_, TimeUnit.MILLISECONDS))
-
-      def lapseTime(startTime: FiniteDuration) = currentTimeMs.map(_ - startTime)
+      val currentTime = clock.monotonic
+      def lapseTime(startTime: FiniteDuration) = currentTime.map(_ - startTime)
 
       def attemptAction(startTime: FiniteDuration): IO[A] =
         for {
@@ -30,18 +27,18 @@ package object interpreter {
             IO(logger.warn(s"exiting loop. Elapsed time: $elapsed")) *>
               IO.raiseError(new TimeoutException(s"Ready check timed out for $label after $totalTimeout"))
           } else IO.unit
-          attemptBegin <- currentTimeMs
+          attemptBegin <- currentTime
           _ <- logger.debug(
             s"running ready-check for $label [time taken so far: $elapsed, check timeout: $checkTimeout, total timeout: $totalTimeout]")
           result <- f().timeout(checkTimeout).attempt
           sleepRemainder <- lapseTime(attemptBegin).map(elapsedTime => List(checkTimeout - elapsedTime, 0.millis).max)
           outcome <- result.fold(
-            err => logger.warn(s"Ready check failed for $label: $err...") *> IO.sleep(sleepRemainder) *> attemptAction(startTime),
+            err => logger.debug(s"Ready check failed for $label: $err...") *> IO.sleep(sleepRemainder) *> attemptAction(startTime),
             out => IO(logger.debug(s"done! total elapsed time: $elapsed")) *> IO.pure(out)
           )
         } yield outcome
 
-      currentTimeMs flatMap attemptAction
+      currentTime flatMap attemptAction
     }
   }
 }

core/src/test/scala/com/itv/servicebox/algebra/packageTest.scala

@@ -2,14 +2,11 @@ package com.itv.servicebox.algebra
 
 import java.nio.file.{Files, Paths}
 import java.util.UUID
-
 import cats.effect.IO
-import cats.instances.future._
+import cats.effect.unsafe.implicits.global
 import org.scalatest.freespec.AnyFreeSpec
 import org.scalatest.matchers.should.Matchers
-import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.duration._
-import scala.concurrent.{Await, Future}
+
 
 class packageTest extends AnyFreeSpec with Matchers {
   "BindMount" - {

core/src/test/scala/com/itv/servicebox/fake/ContainerController.scala

@@ -1,10 +1,9 @@
 package com.itv.servicebox.fake
 
 import java.util.concurrent.atomic.AtomicReference
-
 import cats.MonadError
 import cats.data.NonEmptyList
-import cats.effect.Effect
+import cats.effect.Sync
 import cats.instances.list._
 import cats.syntax.flatMap._
 import cats.syntax.functor._
@@ -21,7 +20,7 @@ class ContainerController[F[_]](
     imageRegistry: ImageRegistry[F],
     logger: Logger[F],
     network: Option[NetworkName],
-    initialState: ContainerStates = Map.empty)(implicit tag: AppTag, E: Effect[F], M: MonadError[F, Throwable])
+    initialState: ContainerStates = Map.empty)(implicit tag: AppTag, S: Sync[F])
     extends algebra.ContainerController[F](imageRegistry, logger, network) {
 
   private val containersByRef = new AtomicReference[ContainerStates](initialState)
@@ -34,7 +33,7 @@ class ContainerController[F[_]](
       containers <- spec.containers.toList
         .traverse[F, Option[ContainerWithState]] { c =>
           val ref = c.ref(spec.ref)
-          E.delay(containersByRef.get)
+          S.delay(containersByRef.get)
             .map(_.get(ref).filter(containerWithState => containerWithState.container.toSpec === c.toSpec))
         }
         .map(_.flatten)
@@ -52,7 +51,7 @@ class ContainerController[F[_]](
         s"starting container ${container.ref.show} service ref: ${serviceRef.show} with port mappings: ${container.portMappings
           .map { case (host, guest) => s"$host -> $guest" }
           .mkString(", ")}")
-      _ <- E.delay(
+      _ <- S.delay(
         containersByRef.getAndUpdate(_.updated(container.ref, ContainerWithState(container, isRunning = true))))
     } yield ()
 
@@ -64,7 +63,7 @@ class ContainerController[F[_]](
 
   private def shutdownContainer(ref: Container.Ref): F[Unit] =
     for {
-      _ <- E.delay(containersByRef.getAndUpdate { containers =>
+      _ <- S.delay(containersByRef.getAndUpdate { containers =>
         containers
           .get(ref)
           .map { c =>

core/src/test/scala/com/itv/servicebox/fake/InMemoryImageRegistry.scala

@@ -1,25 +1,24 @@
 package com.itv.servicebox.fake
 
 import java.util.concurrent.atomic.AtomicReference
-
 import cats.MonadError
-import cats.effect.Effect
+import cats.effect.Sync
 import cats.syntax.functor._
 import cats.syntax.monadError._
 import com.itv.servicebox.algebra
 
 class InMemoryImageRegistry[F[_]](
     logger: algebra.Logger[F],
     imagesDownloaded: Set[String] = Set.empty,
-    imagesAvaliable: Option[Set[String]] = None)(implicit E: Effect[F], M: MonadError[F, Throwable])
+    imagesAvaliable: Option[Set[String]] = None)(implicit S: Sync[F])
     extends algebra.ImageRegistry[F](logger) {
   private val images = new AtomicReference[Set[String]](imagesDownloaded)
 
-  def addImage(imageName: String) = E.delay(images.getAndUpdate(_ + imageName)).void
+  def addImage(imageName: String) = S.delay(images.getAndUpdate(_ + imageName)).void
 
   override def fetchImage(name: String) =
     imagesAvaliable.fold(addImage(name))(available =>
       addImage(name).ensure(new IllegalArgumentException(s"image $name is not available"))(_ => available(name)))
 
-  override def imageExists(imageName: String): F[Boolean] = E.delay(images.get()).map(_(imageName))
+  override def imageExists(imageName: String): F[Boolean] = S.delay(images.get()).map(_(imageName))
 }