KM8S · octavz · Jun 25, 2022 · Jun 30, 2022 · Jul 21, 2022 · Aug 5, 2022
diff --git a/build.sbt b/build.sbt
@@ -3,15 +3,15 @@ ThisBuild / resolvers += Resolver.sonatypeRepo("snapshots")
 lazy val ProjectName = "KM8"
 lazy val ProjectOrganization = "KM8S"
 lazy val ProjectVersion = "0.2.0-SNAPSHOT"
-lazy val ProjectScalaVersion = "3.1.2"
+lazy val ProjectScalaVersion = "3.1.3"
 
 lazy val Versions = new {
 
-  val zio = "1.0.13"
-  val zioKafka = "0.17.5"
-  val zioJson = "0.2.0-M3"
-  val zioLogging = "0.5.14"
-  val zioPrelude = "1.0.0-RC8"
+  val zio = "2.0.0"
+  val zioKafka = "2.0.0"
+  val zioJson = "0.3.0-RC9"
+  val zioLogging = "2.0.0"
+  val zioPrelude = "1.0.0-RC15"
 
   val kafkaProtobuf = "7.1.0"
   val javaFx = "16"

diff --git a/common/common.iml b/common/common.iml
@@ -0,0 +1,50 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module external.linked.project.id="common [file:/Users/octav/work/KM8/]" external.linked.project.path="$MODULE_DIR$" external.root.project.path="$MODULE_DIR$/.." external.system.id="SBT" type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" LANGUAGE_LEVEL="JDK_17">
+    <output url="file://$MODULE_DIR$/target/scala-3.1.3/classes" />
+    <output-test url="file://$MODULE_DIR$/target/scala-3.1.3/test-classes" />
+    <exclude-output />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/src/main/scala" isTestSource="false" />
+      <excludeFolder url="file://$MODULE_DIR$/target/scala-3.1.2" />
+      <excludeFolder url="file://$MODULE_DIR$/target/streams" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+    <orderEntry type="library" name="sbt: dev.zio:izumi-reflect-thirdparty-boopickle-shaded_3:2.1.0:jar" level="project" />
+    <orderEntry type="library" name="sbt: dev.zio:izumi-reflect_3:2.1.0:jar" level="project" />
+    <orderEntry type="library" name="sbt: dev.zio:zio-internal-macros_3:2.0.0:jar" level="project" />
+    <orderEntry type="library" name="sbt: dev.zio:zio-prelude-macros_3:1.0.0-RC15:jar" level="project" />
+    <orderEntry type="library" name="sbt: dev.zio:zio-prelude_3:1.0.0-RC15:jar" level="project" />
+    <orderEntry type="library" name="sbt: dev.zio:zio-stacktracer_3:2.0.0:jar" level="project" />
+    <orderEntry type="library" name="sbt: dev.zio:zio-streams_3:2.0.0:jar" level="project" />
+    <orderEntry type="library" scope="TEST" name="sbt: dev.zio:zio-test-sbt_3:2.0.0:jar" level="project" />
+    <orderEntry type="library" scope="TEST" name="sbt: dev.zio:zio-test_3:2.0.0:jar" level="project" />
+    <orderEntry type="library" name="sbt: dev.zio:zio_3:2.0.0:jar" level="project" />
+    <orderEntry type="library" name="sbt: io.circe:circe-core_3:0.14.1:jar" level="project" />
+    <orderEntry type="library" name="sbt: io.circe:circe-generic_3:0.14.1:jar" level="project" />
+    <orderEntry type="library" name="sbt: io.circe:circe-numbers_3:0.14.1:jar" level="project" />
+    <orderEntry type="library" scope="TEST" name="sbt: org.portable-scala:portable-scala-reflect_2.13:1.1.2:jar" level="project" />
+    <orderEntry type="library" name="sbt: org.scala-lang:scala-library:2.13.8:jar" level="project" />
+    <orderEntry type="library" name="sbt: org.scala-lang:scala3-library_3:3.1.3:jar" level="project" />
+    <orderEntry type="library" scope="TEST" name="sbt: org.scala-sbt:test-interface:1.0:jar" level="project" />
+    <orderEntry type="library" name="sbt: org.typelevel:cats-core_3:2.6.1:jar" level="project" />
+    <orderEntry type="library" name="sbt: org.typelevel:cats-kernel_3:2.6.1:jar" level="project" />
+    <orderEntry type="library" name="sbt: org.typelevel:simulacrum-scalafix-annotations_3:0.5.4:jar" level="project" />
+    <orderEntry type="module-library">
+      <library name="sbt: unmanaged-sources-and-docs">
+        <CLASSES />
+        <JAVADOC>
+          <root url="jar://$USER_HOME$/Library/Caches/Coursier/v1/https/repo1.maven.org/maven2/org/openjfx/javafx-fxml/16/javafx-fxml-16-javadoc.jar!/" />
+          <root url="jar://$USER_HOME$/Library/Caches/Coursier/v1/https/repo1.maven.org/maven2/org/openjfx/javafx-swing/16/javafx-swing-16-javadoc.jar!/" />
+          <root url="jar://$USER_HOME$/Library/Caches/Coursier/v1/https/repo1.maven.org/maven2/org/openjfx/javafx-web/16/javafx-web-16-javadoc.jar!/" />
+        </JAVADOC>
+        <SOURCES>
+          <root url="jar://$USER_HOME$/Library/Caches/Coursier/v1/https/repo1.maven.org/maven2/org/openjfx/javafx-fxml/16/javafx-fxml-16-sources.jar!/" />
+          <root url="jar://$USER_HOME$/Library/Caches/Coursier/v1/https/repo1.maven.org/maven2/org/openjfx/javafx-swing/16/javafx-swing-16-sources.jar!/" />
+          <root url="jar://$USER_HOME$/Library/Caches/Coursier/v1/https/repo1.maven.org/maven2/org/openjfx/javafx-web/16/javafx-web-16-sources.jar!/" />
+        </SOURCES>
+      </library>
+    </orderEntry>
+  </component>
+</module>
diff --git a/core/src/it/it.iml b/core/src/it/it.iml
@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/scala" isTestSource="false" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
diff --git a/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala b/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala
@@ -1,54 +1,51 @@
 package io.km8.core.kafka
 
 import zio.*
-import zio.blocking.Blocking
-import zio.clock.Clock
-import zio.duration.*
+import zio.test.*
 import zio.kafka.consumer.*
 import zio.kafka.serde.Serde
-import zio.test.DefaultRunnableSpec
 import zio.test.*
 import zio.test.Assertion.*
 
 import io.km8.common.*
 
-object KafkaExplorerSpec extends DefaultRunnableSpec:
+object KafkaExplorerSpec extends ZIOSpecDefault:
 
   override def spec =
     suite("Kafka explorer spec")(
-      testM("test listConsumerGroups & listConsumerOffsets") {
+      test("test listConsumerGroups & listConsumerOffsets") {
         val clusterId = "123"
         val consumerGroup = "group-km8"
 
         val layer =
-          Clock.live >+>
-            Blocking.live >+>
-            itlayers.kafkaContainer >+>
+          itlayers.kafkaContainer >+>
             itlayers.clusterConfig(clusterId) >+>
             KafkaExplorer.liveLayer >+>
             itlayers.consumerLayer(consumerGroup) >+>
             KafkaProducer.liveLayer
 
         val test =
           for {
-            topic <- UIO("topic10")
+            topic <- ZIO.succeed("topic10")
             _ <- KafkaProducer.produce(topic, "123", "test")(clusterId)
             _ <- Consumer
                    .subscribeAnd(Subscription.topics(topic))
                    .plainStream(Serde.string, Serde.string)
-                   .mapM(cr =>
+                   .mapZIO(cr =>
                      ZIO.debug(s"${"-" * 10}> key: ${cr.key}, value: ${cr.value}") *>
                        cr.offset.commit.as(cr.value)
                    )
                    .take(1)
                    .runCollect
-            r1 <- KafkaExplorer(_.listConsumerGroups(clusterId))
-            r2 <- KafkaExplorer(_.listConsumerOffsets(clusterId, consumerGroup))
+            r1 <- KafkaExplorer.listConsumerGroups(clusterId)
+            r2 <- KafkaExplorer.listConsumerOffsets(clusterId, consumerGroup)
           } yield {
-            assert(r1.groups.map(_.groupId))(equalTo(List(consumerGroup))) &&
-            assert(r2.offsets)(equalTo(Map(TopicPartitionInternal(topic, 0) -> 1)))
+            assertTrue(
+              r1.groups.map(_.groupId) == List(consumerGroup),
+              r2.offsets == Map(TopicPartitionInternal(topic, 0) -> 1L)
+            )
           }
 
         test.provideLayer(layer)
       } @@ TestAspect.timeout(30.seconds)
-    )
+    ).provide(zio.test.liveEnvironment, zio.test.Live.default)
diff --git a/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala b/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala
@@ -1,38 +1,34 @@
 package io.km8.core.kafka
 
 import zio.*
-import zio.blocking.Blocking
-import zio.clock.Clock
-import zio.duration.*
 import zio.kafka.consumer.*
 import zio.kafka.serde.Serde
 import zio.test.*
 import zio.test.Assertion.*
 
 import com.dimafeng.testcontainers.KafkaContainer
 import io.km8.common.*
-import io.km8.core.kafka.KafkaProducer.KafkaProducer
 import org.apache.kafka.clients.consumer.OffsetResetStrategy
 
-object KafkaProducerSpec extends DefaultRunnableSpec:
+object KafkaProducerSpec extends ZIOSpecDefault:
 
-  private val consumerLayer: ZLayer[Has[KafkaContainer], Nothing, Has[KafkaConsumer]] =
-    Clock.live ++ Blocking.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaConsumer.liveLayer
+  private val consumerLayer: ZLayer[KafkaContainer, Nothing, KafkaConsumer] =
+    itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaConsumer.liveLayer
 
-  private val producerLayer: ZLayer[Has[KafkaContainer], Nothing, KafkaProducer] =
-    Blocking.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaProducer.liveLayer
+  private val producerLayer: ZLayer[KafkaContainer, Nothing, KafkaProducer] =
+    itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaProducer.liveLayer
 
-  val specLayer: ZLayer[Has[KafkaContainer], Nothing, Has[KafkaConsumer] & KafkaProducer] =
+  val specLayer: ZLayer[KafkaContainer, Nothing, KafkaConsumer & KafkaProducer] =
     consumerLayer ++ producerLayer
 
-  override def spec: ZSpec[_root_.zio.test.environment.TestEnvironment, Any] =
+  override def spec =
     mainSpec
-      .provideSomeLayer[environment.TestEnvironment & Has[KafkaContainer]](specLayer ++ Clock.live)
-      .provideCustomLayerShared(Blocking.live >>> itlayers.kafkaContainer)
+      .provideSomeLayer[TestEnvironment & KafkaContainer](specLayer)
+      .provideCustomLayerShared(itlayers.kafkaContainer)
 
   private val mainSpec =
     suite("Kafka services")(
-      testM("KafkaProducer sends a message and KafkaConsumer reads it correctly ") {
+      test("KafkaProducer sends a message and KafkaConsumer reads it correctly ") {
         for
           f1 <- KafkaConsumer
                   .consume(
@@ -49,7 +45,7 @@ object KafkaProducerSpec extends DefaultRunnableSpec:
                   .fork
           _ <- KafkaProducer.produce("test_topic", "key", "value")("cluster_id")
           maybeValue <- f1.join
-        yield assert(maybeValue.map(_.value))(isSome(equalTo("value")))
+        yield assertTrue(maybeValue.map(_.value).get == "value")
 
       }
     )
diff --git a/core/src/it/scala/io/km8/core/kafka/itlayers.scala b/core/src/it/scala/io/km8/core/kafka/itlayers.scala
@@ -2,8 +2,6 @@ package io.km8.core.kafka
 
 import com.dimafeng.testcontainers.KafkaContainer
 import zio.*
-import zio.clock.Clock
-import zio.blocking.*
 import zio.kafka.consumer.Consumer
 import zio.kafka.consumer.Consumer.{AutoOffsetStrategy, OffsetRetrieval}
 import zio.kafka.consumer.ConsumerSettings
@@ -12,50 +10,52 @@ import io.km8.core.config.{ClusterConfig, ClusterProperties, ClusterSettings}
 
 object itlayers:
 
-  val kafkaContainer: ZLayer[Blocking, Nothing, Has[KafkaContainer]] =
-    ZManaged.make {
-      effectBlocking {
+  val kafkaContainer: ZLayer[Any, Nothing, KafkaContainer] =
+    ZLayer.scoped {
+      ZIO.acquireRelease(ZIO.attemptBlocking {
         val container = new KafkaContainer()
         container.start()
         container
-      }.orDie
-    }(container => effectBlocking(container.stop()).orDie).toLayer
+      })(container => ZIO.attemptBlocking(container.stop()).orDie)
+    }.orDie
 
-  def consumerSettings(cg: String): ZManaged[Has[KafkaContainer], Nothing, ConsumerSettings] =
+  def consumerSettings(cg: String): ZIO[KafkaContainer with Scope, Nothing, ConsumerSettings] =
     ZIO
       .service[KafkaContainer]
       .map(c =>
         ConsumerSettings(List(c.bootstrapServers))
           .withGroupId(cg)
           .withOffsetRetrieval(OffsetRetrieval.Auto(AutoOffsetStrategy.Earliest))
       )
-      .toManaged_
 
-  def consumerLayer(cgroup: String): ZLayer[Has[KafkaContainer] with Clock with Blocking, Nothing, Has[Consumer]] =
-    consumerSettings(cgroup).flatMap(Consumer.make(_)).orDie.toLayer
-
-  def clusterConfig(clusterId: String): ZLayer[Has[KafkaContainer], Nothing, Has[ClusterConfig]] =
-    ZIO
-      .service[KafkaContainer]
-      .map(kafkaContainer =>
-        new ClusterConfig {
-
-          override def readClusters: Task[ClusterProperties] = Task(
-            ClusterProperties(clusters =
-              List(
-                ClusterSettings(
-                  id = clusterId,
-                  name = kafkaContainer.containerName,
-                  kafkaHosts = List(kafkaContainer.bootstrapServers),
-                  schemaRegistryUrl = None
+  def consumerLayer(cgroup: String): ZLayer[KafkaContainer with Clock, Nothing, Consumer] =
+    ZLayer.scoped {
+      consumerSettings(cgroup).flatMap(Consumer.make(_)).orDie
+    }
+
+  def clusterConfig(clusterId: String): ZLayer[KafkaContainer, Nothing, ClusterConfig] =
+    ZLayer.fromZIO {
+      ZIO
+        .service[KafkaContainer]
+        .map(kafkaContainer =>
+          new ClusterConfig {
+
+            override def readClusters: Task[ClusterProperties] = ZIO.attempt(
+              ClusterProperties(clusters =
+                List(
+                  ClusterSettings(
+                    id = clusterId,
+                    name = kafkaContainer.containerName,
+                    kafkaHosts = List(kafkaContainer.bootstrapServers),
+                    schemaRegistryUrl = None
+                  )
                 )
               )
             )
-          )
 
-          override def writeClusters(cluster: ClusterSettings): Task[Unit] = ???
+            override def writeClusters(cluster: ClusterSettings): Task[Unit] = ???
 
-          override def deleteCluster(clusterId: String): Task[ClusterProperties] = ???
-        }
-      )
-      .toLayer
+            override def deleteCluster(clusterId: String): Task[ClusterProperties] = ???
+          }
+        )
+    }
diff --git a/core/src/main/main.iml b/core/src/main/main.iml
@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/scala" isTestSource="false" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
diff --git a/core/src/main/scala/io/km8/core/config/ClustersConfig.scala b/core/src/main/scala/io/km8/core/config/ClustersConfig.scala
@@ -48,29 +48,30 @@ trait ClusterConfig {
 
 object ClustersConfig {
 
-  def readClusters = ZIO.serviceWith[ClusterConfig](_.readClusters)
-  def writeClusters(cluster: ClusterSettings) = ZIO.serviceWith[ClusterConfig](_.writeClusters(cluster))
-  def deleteCluster(clusterId: String) = ZIO.serviceWith[ClusterConfig](_.deleteCluster(clusterId))
+  def readClusters = ZIO.serviceWithZIO[ClusterConfig](_.readClusters)
+  def writeClusters(cluster: ClusterSettings) = ZIO.serviceWithZIO[ClusterConfig](_.writeClusters(cluster))
+  def deleteCluster(clusterId: String) = ZIO.serviceWithZIO[ClusterConfig](_.deleteCluster(clusterId))
 
-  lazy val liveLayer: URLayer[Has[ConfigPath] with Logging, Has[ClusterConfig]] = (ClusterConfigLive(_, _)).toLayer
+  lazy val liveLayer: URLayer[ConfigPath, ClusterConfig] =
+    ZLayer(ZIO.service[ConfigPath].map(ClusterConfigLive.apply))
 
-  case class ClusterConfigLive(configPath: ConfigPath, log: Logger[String]) extends ClusterConfig {
+  case class ClusterConfigLive(configPath: ConfigPath) extends ClusterConfig {
     private val configFilepath = configPath.path
     private def emptyProperties = ClusterProperties(List.empty)
     private def emptyPropertiesJson = emptyProperties.toJsonPretty
 
     private def writeJson(json: => String) =
-      Task(os.write.over(configFilepath, json, createFolders = true))
+      ZIO.attempt(os.write.over(configFilepath, json, createFolders = true))
 
     def readClusters: Task[ClusterProperties] =
       for {
-        b <- Task(os.exists(configFilepath))
+        b <- ZIO.attempt(os.exists(configFilepath))
         _ <- writeJson(emptyPropertiesJson).unless(b)
-        s <- Task(os.read(configFilepath))
+        s <- ZIO.attempt(os.read(configFilepath))
         r <- ZIO
                .fromEither(s.fromJson[ClusterProperties])
                .catchAll { err =>
-                 log.warn(s"Parsing error: $err") *>
+                 ZIO.logWarning(s"Parsing error: $err") *>
                    writeJson(emptyPropertiesJson).as(emptyProperties)
                }
       } yield r
@@ -85,7 +86,7 @@ object ClustersConfig {
     def deleteCluster(clusterId: String): Task[ClusterProperties] =
       for {
         c <- readClusters
-        ls <- ZIO.filterNotPar(c.clusters)(s => Task(s.id == clusterId))
+        ls <- ZIO.filterNotPar(c.clusters)(s => ZIO.attempt(s.id == clusterId))
         json = ClusterProperties(ls).toJsonPretty
         _ <- writeJson(json)
         r <- readClusters

diff --git a/core/src/main/scala/io/km8/core/config/ConfigPathService.scala b/core/src/main/scala/io/km8/core/config/ConfigPathService.scala
@@ -2,15 +2,16 @@ package io.km8.core
 package config
 
 import zio.*
-import zio.system.*
+import zio.System
+import zio.System.env
 
 lazy val EnvKey = "KAFKAMATE_ENV"
 lazy val FileName = "kafkamate.json"
 
 case class ConfigPath(path: os.Path)
 
-lazy val liveLayer: URLayer[System, Has[ConfigPath]] =
-  env(EnvKey).map {
+lazy val liveLayer: URLayer[System, ConfigPath] =
+  ZLayer.fromZIO(env(EnvKey).map {
     case Some("prod") => ConfigPath(os.root / FileName)
     case _            => ConfigPath(os.pwd / FileName)
-  }.orDie.toLayer
+  }.orDie)