diff --git a/build.sbt b/build.sbt
index 3191a48..6f195b3 100644
--- 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
new file mode 100644
index 0000000..601aad8
--- /dev/null
+++ b/common/common.iml
@@ -0,0 +1,50 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/core/src/it/it.iml b/core/src/it/it.iml
new file mode 100644
index 0000000..3acf350
--- /dev/null
+++ b/core/src/it/it.iml
@@ -0,0 +1,11 @@
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala b/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala
index 569b303..7682924 100644
--- a/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala
+++ b/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala
@@ -1,29 +1,24 @@
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) >+>
@@ -31,24 +26,26 @@ object KafkaExplorerSpec extends DefaultRunnableSpec:
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
index 818c41a..d0b44aa 100644
--- a/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala
+++ b/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala
@@ -1,9 +1,6 @@
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.*
@@ -11,28 +8,27 @@ 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
index ff96bfe..04bfe85 100644
--- 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,16 +10,16 @@ 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 =>
@@ -29,33 +27,35 @@ object itlayers:
.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
new file mode 100644
index 0000000..3acf350
--- /dev/null
+++ b/core/src/main/main.iml
@@ -0,0 +1,11 @@
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/core/src/main/scala/io/km8/core/config/ClustersConfig.scala b/core/src/main/scala/io/km8/core/config/ClustersConfig.scala
index 3f5a18f..66a8f92 100644
--- 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
index 6749268..4fa1f73 100644
--- 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)
diff --git a/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala b/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala
index f1dddc4..eecf985 100644
--- a/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala
+++ b/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala
@@ -2,9 +2,7 @@ package io.km8.core
package kafka
import zio.*
-import zio.blocking.Blocking
-import zio.clock.Clock
-import zio.duration.*
+
import zio.kafka.consumer.*
import zio.kafka.consumer.Consumer.*
import zio.kafka.serde.Deserializer
@@ -28,19 +26,19 @@ trait KafkaConsumer {
object KafkaConsumer {
- lazy val liveLayer: URLayer[Clock with Blocking with Has[ClusterConfig], Has[KafkaConsumer]] =
- (KafkaConsumerLive(_, _, _)).toLayer
+ lazy val liveLayer: URLayer[ClusterConfig, KafkaConsumer] =
+ ZLayer {
+ for {
+ config <- ZIO.service[ClusterConfig]
+ } yield KafkaConsumerLive(config)
+ }
- def consume(request: ConsumeRequest): ZStream[Has[KafkaConsumer], Throwable, Message] =
- ZStream.accessStream[Has[KafkaConsumer]](_.get.consume(request))
+ def consume(request: ConsumeRequest): ZStream[KafkaConsumer, Throwable, Message] =
+ ZStream.environmentWithStream[KafkaConsumer](_.get.consume(request))
case class KafkaConsumerLive(
- clock: Clock.Service,
- blocking: Blocking.Service,
clustersConfigService: ClusterConfig)
extends KafkaConsumer {
- private val clockLayer = ZLayer.succeed(clock)
- private val blockingLayer = ZLayer.succeed(blocking)
// TODO Ciprian transform to enum or reuse the Consumer enum + serdes
private def extractOffsetStrategy(offsetValue: String): AutoOffsetStrategy =
@@ -55,7 +53,7 @@ object KafkaConsumer {
.map(_.asTry)
private def consumerSettings(config: ClusterSettings, offsetStrategy: String): Task[ConsumerSettings] =
- Task {
+ ZIO.attempt {
val uuid = UUID.randomUUID().toString
ConsumerSettings(config.kafkaHosts)
.withGroupId(s"group-kafkamate-$uuid")
@@ -68,11 +66,11 @@ object KafkaConsumer {
private def makeConsumerLayer(
clusterId: String,
offsetStrategy: String
- ): ZLayer[Clock & Blocking, Throwable, Has[Consumer]] =
- ZLayer.fromManaged {
+ ): ZLayer[Any, Throwable, Consumer] =
+ ZLayer.scoped {
for {
- cs <- clustersConfigService.getCluster(clusterId).toManaged_
- settings <- consumerSettings(cs, offsetStrategy).toManaged_
+ cs <- clustersConfigService.getCluster(clusterId)
+ settings <- consumerSettings(cs, offsetStrategy)
consumer <- Consumer.make(settings)
} yield consumer
}
@@ -80,7 +78,7 @@ object KafkaConsumer {
def consume(request: ConsumeRequest): ZStream[Any, Throwable, Message] = {
def consumer[T](
valueDeserializer: Deserializer[Any, Try[T]]
- ): ZStream[Has[Consumer], Throwable, Message] = Consumer
+ ): ZStream[Consumer, Throwable, Message] = Consumer
.subscribeAnd(Subscription.topics(request.topicName))
.plainStream(Deserializer.string, valueDeserializer)
.collect {
@@ -98,7 +96,7 @@ object KafkaConsumer {
.orElseFail(new Exception("SchemaRegistry url was not provided!"))
)
ZStream
- .fromEffect(protoSettings.flatMap(protobufDeserializer))
+ .fromZIO(protoSettings.flatMap(protobufDeserializer))
.flatMap(consumer)
case _ => consumer(Deserializer.string.asTry)
}
@@ -114,8 +112,7 @@ object KafkaConsumer {
else withFilter.take(request.maxResults)
withFilterLimit.provideLayer(
- clockLayer ++ blockingLayer >>>
- makeConsumerLayer(request.clusterId, request.offsetStrategy)
+ makeConsumerLayer(request.clusterId, request.offsetStrategy)
)
}
}
diff --git a/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala b/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala
index 2fe4b2a..afeaee8 100644
--- a/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala
+++ b/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala
@@ -2,9 +2,7 @@ package io.km8.core
package kafka
import zio.*
-import zio.blocking.Blocking
-import zio.clock.Clock
-import zio.duration.*
+
import zio.kafka.admin.*
import zio.kafka.admin.AdminClient.*
@@ -19,131 +17,144 @@ trait KafkaExplorer:
def listConsumerGroups(clusterId: String): Task[ConsumerGroupsResponse]
def listConsumerOffsets(clusterId: String, groupId: String): Task[ConsumerGroupOffsetsResponse]
-object KafkaExplorer extends Accessible[KafkaExplorer]:
+object KafkaExplorer:
val CleanupPolicyKey = "cleanup.policy"
val RetentionMsKey = "retention.ms"
- lazy val liveLayer: URLayer[Has[ClusterConfig] with Clock with Blocking, Has[KafkaExplorer]] =
- KafkaExplorerLive.apply.toLayer
-
- case class KafkaExplorerLive(
- clock: Clock.Service,
- blocking: Blocking.Service,
- clustersConfigService: ClusterConfig)
- extends KafkaExplorer {
- val clockLayer = ZLayer.succeed(clock)
- val blockingLayer = ZLayer.succeed(blocking)
-
- private def adminClientLayer(clusterId: String) =
- ZLayer.fromManaged {
- for {
- cs <- clustersConfigService.getCluster(clusterId).toManaged_
- client <- AdminClient.make(AdminClientSettings(cs.kafkaHosts, 2.seconds, Map.empty))
- } yield client
- }
-
- def withAdminClient[A](clusterId: String)(eff: AdminClient => RIO[Blocking with Clock, A]): Task[A] =
+ def listConsumerGroups(clusterId: String): ZIO[KafkaExplorer, Throwable, ConsumerGroupsResponse] =
+ ZIO.serviceWithZIO(_.listConsumerGroups(clusterId))
+
+ def listConsumerOffsets(
+ clusterId: String,
+ groupId: String
+ ): ZIO[KafkaExplorer, Throwable, ConsumerGroupOffsetsResponse] =
+ ZIO.serviceWithZIO(_.listConsumerOffsets(clusterId, groupId))
+
+ lazy val liveLayer: ZLayer[ClusterConfig, Nothing, KafkaExplorer] =
+ ZLayer {
+ for {
+ cc <- ZIO.service[ClusterConfig]
+ } yield KafkaExplorerLive(cc)
+ }
+
+case class KafkaExplorerLive(clustersConfigService: ClusterConfig) extends KafkaExplorer {
+
+ private def adminClientLayer(clusterId: String) =
+ ZLayer.fromZIO {
+ for {
+ cs <- clustersConfigService.getCluster(clusterId)
+ client <- AdminClient.make(AdminClientSettings(cs.kafkaHosts, 2.seconds, Map.empty))
+ } yield client
+ }
+
+ def withAdminClient[A](clusterId: String)(eff: AdminClient => RIO[Any, A]): ZIO[Any, Throwable, A] =
+ ZIO.scoped {
ZIO
.service[AdminClient]
.flatMap(eff(_).timeoutFail(new Exception("Timed out"))(6.seconds))
- .provideLayer(blockingLayer >+> adminClientLayer(clusterId) ++ clockLayer)
-
- override def listBrokers(clusterId: String) =
- withAdminClient(clusterId) { ac =>
- for {
- (nodes, controllerId) <- ac.describeClusterNodes() <&> ac.describeClusterController().map(_.map(_.id))
- brokers = nodes.map { n =>
- val nodeId = n.id
- if (controllerId.contains(nodeId)) BrokerDetails(nodeId, isController = true)
- else BrokerDetails(id = nodeId, isController = false)
- }
- // resources = nodes.map(n => new ConfigResource(ConfigResource.Type.BROKER, n.idString()))
- // _ <- ac.describeConfigs(resources)
- } yield brokers
- }
-
- override def listTopics(clusterId: String) =
- withAdminClient(clusterId) { ac =>
- ac.listTopics()
- .map(_.keys.toList)
- .flatMap(ls => ZIO.filterNotPar(ls)(t => UIO(t.startsWith("_"))))
- .flatMap(ls =>
- ac.describeTopics(ls) <&> ac.describeConfigs(ls.map(ConfigResource(ConfigResourceType.Topic, _)))
- )
- .map { case (nameDescriptionMap, topicConfigMap) =>
- val configs = topicConfigMap.map { case (res, conf) => (res.name, conf) }
- nameDescriptionMap.map { case (name, description) =>
- val conf = configs.get(name).map(_.entries)
- def getConfig(key: String) = conf.flatMap(_.get(key).map(_.value())).getOrElse("unknown")
- TopicDetails(
- name = name,
- partitions = description.partitions.size,
- replication = description.partitions.headOption.map(_.replicas.size).getOrElse(0),
- cleanupPolicy = getConfig(CleanupPolicyKey),
- retentionMs = getConfig(RetentionMsKey),
- size = 0
- )
- }.toList.sortBy(_.name)
- }
- }
-
- override def addTopic(req: AddTopicRequest) =
- withAdminClient(req.clusterId) { ac =>
- ac
- .createTopic(
- AdminClient.NewTopic(
- req.name,
- req.partitions,
- req.replication.toShort,
- Map(
- CleanupPolicyKey -> req.cleanupPolicy,
- RetentionMsKey -> req.retentionMs
- )
- )
- )
- .as(
+ .provideLayer(adminClientLayer(clusterId))
+ }
+
+ override def listBrokers(clusterId: String) =
+ withAdminClient(clusterId) { ac =>
+ for {
+ nodes <- ac.describeClusterNodes()
+ controllerId <- ac.describeClusterController().map(_.map(_.id))
+ brokers = nodes.map { n =>
+ val nodeId = n.id
+ if (controllerId.contains(nodeId)) BrokerDetails(nodeId, isController = true)
+ else BrokerDetails(id = nodeId, isController = false)
+ }
+ // resources = nodes.map(n => new ConfigResource(ConfigResource.Type.BROKER, n.idString()))
+ // _ <- ac.describeConfigs(resources)
+ } yield brokers
+ }
+
+ override def listTopics(clusterId: String) =
+ withAdminClient(clusterId) { ac =>
+ ac.listTopics()
+ .map(_.keys.toList)
+ .flatMap(ls => ZIO.filterNotPar(ls)(t => ZIO.succeed(t.startsWith("_"))))
+ .flatMap(ls =>
+ ac.describeTopics(ls) <&> ac.describeConfigs(ls.map(ConfigResource(ConfigResourceType.Topic, _)))
+ )
+ .map { case (nameDescriptionMap, topicConfigMap) =>
+ val configs = topicConfigMap.map { case (res, conf) => (res.name, conf) }
+ nameDescriptionMap.map { case (name, description) =>
+ val conf = configs.get(name).map(_.entries)
+ def getConfig(key: String) = conf.flatMap(_.get(key).map(_.value())).getOrElse("unknown")
TopicDetails(
- name = req.name,
- partitions = req.partitions,
- replication = req.replication,
- cleanupPolicy = req.cleanupPolicy,
- retentionMs = req.retentionMs,
+ name = name,
+ partitions = description.partitions.size,
+ replication = description.partitions.headOption.map(_.replicas.size).getOrElse(0),
+ cleanupPolicy = getConfig(KafkaExplorer.CleanupPolicyKey),
+ retentionMs = getConfig(KafkaExplorer.RetentionMsKey),
size = 0
)
+ }.toList.sortBy(_.name)
+ }
+ }
+
+ override def addTopic(req: AddTopicRequest) =
+ withAdminClient(req.clusterId) { ac =>
+ ac
+ .createTopic(
+ AdminClient.NewTopic(
+ req.name,
+ req.partitions,
+ req.replication.toShort,
+ Map(
+ KafkaExplorer.CleanupPolicyKey -> req.cleanupPolicy,
+ KafkaExplorer.RetentionMsKey -> req.retentionMs
+ )
)
- }
-
- override def deleteTopic(req: DeleteTopicRequest) =
- withAdminClient(req.clusterId) {
- _.deleteTopic(req.topicName)
- .as(DeleteTopicResponse(req.topicName))
- }
-
- override def listConsumerGroups(clusterId: String): Task[ConsumerGroupsResponse] =
- def mapConsumerGroup(state: Option[ConsumerGroupState]): ConsumerGroupInternalState = state match {
- case None | Some(ConsumerGroupState.Unknown) => ConsumerGroupInternalState.Unknown
- case Some(ConsumerGroupState.PreparingRebalance) => ConsumerGroupInternalState.PreparingRebalance
- case Some(ConsumerGroupState.CompletingRebalance) => ConsumerGroupInternalState.CompletingRebalance
- case Some(ConsumerGroupState.Stable) => ConsumerGroupInternalState.Stable
- case Some(ConsumerGroupState.Dead) => ConsumerGroupInternalState.Dead
- case Some(ConsumerGroupState.Empty) => ConsumerGroupInternalState.Empty
- }
-
- withAdminClient(clusterId) {
- _.listConsumerGroups()
- .map(lst => ConsumerGroupsResponse(lst.map(r => ConsumerGroupInternal(r.groupId, mapConsumerGroup(r.state)))))
- }
-
- end listConsumerGroups
-
- override def listConsumerOffsets(clusterId: String, groupId: String): Task[ConsumerGroupOffsetsResponse] =
- withAdminClient(clusterId) {
- _.listConsumerGroupOffsets(groupId).map(res =>
- ConsumerGroupOffsetsResponse(res.map { case (tp, offMeta) =>
- TopicPartitionInternal(tp.name, tp.partition) -> offMeta.offset
- })
)
- }
-
- }
+ .as(
+ TopicDetails(
+ name = req.name,
+ partitions = req.partitions,
+ replication = req.replication,
+ cleanupPolicy = req.cleanupPolicy,
+ retentionMs = req.retentionMs,
+ size = 0
+ )
+ )
+ }
+
+ override def deleteTopic(req: DeleteTopicRequest) =
+ withAdminClient(req.clusterId) {
+ _.deleteTopic(req.topicName)
+ .as(DeleteTopicResponse(req.topicName))
+ }
+
+ override def listConsumerGroups(clusterId: String): Task[ConsumerGroupsResponse] =
+ def mapConsumerGroup(state: Option[ConsumerGroupState]): ConsumerGroupInternalState = state match {
+ case None | Some(ConsumerGroupState.Unknown) => ConsumerGroupInternalState.Unknown
+ case Some(ConsumerGroupState.PreparingRebalance) => ConsumerGroupInternalState.PreparingRebalance
+ case Some(ConsumerGroupState.CompletingRebalance) => ConsumerGroupInternalState.CompletingRebalance
+ case Some(ConsumerGroupState.Stable) => ConsumerGroupInternalState.Stable
+ case Some(ConsumerGroupState.Dead) => ConsumerGroupInternalState.Dead
+ case Some(ConsumerGroupState.Empty) => ConsumerGroupInternalState.Empty
+ }
+
+ withAdminClient(clusterId) {
+ _.listConsumerGroups()
+ .map(lst => ConsumerGroupsResponse(lst.map(r => ConsumerGroupInternal(r.groupId, mapConsumerGroup(r.state)))))
+ }
+
+ end listConsumerGroups
+
+ override def listConsumerOffsets(
+ clusterId: String,
+ groupId: String
+ ): ZIO[Any, Throwable, ConsumerGroupOffsetsResponse] =
+ withAdminClient(clusterId) {
+ _.listConsumerGroupOffsets(groupId).map(res =>
+ ConsumerGroupOffsetsResponse(res.map { case (tp, offMeta) =>
+ TopicPartitionInternal(tp.name, tp.partition) -> offMeta.offset
+ })
+ )
+ }
+
+}
diff --git a/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala b/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala
index 6c2e2b4..f6ea311 100644
--- a/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala
+++ b/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala
@@ -2,26 +2,25 @@ package io.km8.core
package kafka
import zio.*
-import zio.blocking.*
+
import zio.kafka.producer.*
import zio.kafka.serde.*
import io.km8.core.config.*
import io.km8.core.config.ClustersConfig.*
-object KafkaProducer {
- type KafkaProducer = Has[Service]
+trait KafkaProducer {
- trait Service {
+ def produce(
+ topic: String,
+ key: String,
+ value: String
+ )(
+ clusterId: String
+ ): ZIO[Scope, Throwable, Unit]
+}
- def produce(
- topic: String,
- key: String,
- value: String
- )(
- clusterId: String
- ): Task[Unit]
- }
+object KafkaProducer {
def produce(
topic: String,
@@ -29,39 +28,39 @@ object KafkaProducer {
value: String
)(
clusterId: String
- ): RIO[KafkaProducer, Unit] = ZIO.accessM[KafkaProducer](_.get.produce(topic, key, value)(clusterId))
-
- lazy val liveLayer: URLayer[Has[ClusterConfig] with Blocking, KafkaProducer] =
- ZLayer.fromServices[ClusterConfig, Blocking.Service, KafkaProducer.Service](
- (clusterConfigService: ClusterConfig, blocking: Blocking.Service) =>
- new Service {
+ ): ZIO[KafkaProducer, Throwable, Unit] =
+ ZIO.scoped {
+ ZIO.environmentWithZIO[KafkaProducer](_.get.produce(topic, key, value)(clusterId))
+ }
- lazy val serdeLayer: ULayer[Has[Serializer[Any, String]]] =
- ZLayer.succeed(Serde.string)
+ lazy val liveLayer: URLayer[ClusterConfig, KafkaProducer] = ZLayer.fromZIO(for {
+ clusterConfigService <- ZIO.service[ClusterConfig]
+ } yield new KafkaProducer {
- def settingsLayer(clusterId: String): Task[ProducerSettings] =
- clusterConfigService
- .getCluster(clusterId)
- .map(c => ProducerSettings(c.kafkaHosts))
+ lazy val serdeLayer: ULayer[Serializer[Any, String]] =
+ ZLayer.succeed(Serde.string)
- def producerLayer(clusterId: String): ZLayer[Any, Throwable, Has[Producer]] =
- settingsLayer(clusterId).toManaged_
- .flatMap(settings => Producer.make(settings))
- .provideLayer(ZLayer.succeed(blocking) ++ serdeLayer ++ ZLayer.succeed(clusterConfigService))
- .toLayer
+ def settingsLayer(clusterId: String): Task[ProducerSettings] =
+ clusterConfigService
+ .getCluster(clusterId)
+ .map(c => ProducerSettings(c.kafkaHosts))
- def produce(
- topic: String,
- key: String,
- value: String
- )(
- clusterId: String
- ): Task[Unit] =
- Producer
- .produce[Any, String, String](topic, key, value, Serde.string, Serde.string)
- .unit
- .provideLayer(producerLayer(clusterId))
- }
- )
+ def producerLayer(clusterId: String): ZIO[Scope, Throwable, Producer] =
+ ZIO.scoped {
+ settingsLayer(clusterId)
+ .flatMap(settings => Producer.make(settings))
+ }
+ def produce(
+ topic: String,
+ key: String,
+ value: String
+ )(
+ clusterId: String
+ ): ZIO[Scope, Throwable, Unit] =
+ Producer
+ .produce[Any, String, String](topic, key, value, Serde.string, Serde.string)
+ .unit
+ .provideLayer(ZLayer.fromZIO(producerLayer(clusterId)))
+ })
}
diff --git a/core/src/main/scala/io/km8/core/utils/Logger.scala b/core/src/main/scala/io/km8/core/utils/Logger.scala
index b064390..3ed33cf 100644
--- a/core/src/main/scala/io/km8/core/utils/Logger.scala
+++ b/core/src/main/scala/io/km8/core/utils/Logger.scala
@@ -2,16 +2,18 @@ package io.km8.core
package utils
import zio.URLayer
-import zio.clock.Clock
-import zio.console.Console
+import zio.Clock
import zio.logging.*
+import zio.Console
object Logger {
+ /*
lazy val liveLayer: URLayer[Console with Clock, Logging] =
- Logging.console(
+ console(
logLevel = LogLevel.Info,
format = LogFormat.ColoredLogFormat()
) >>> Logging.withRootLoggerName("kafkamate")
+ */
}
diff --git a/core/src/test/scala/io/kafkamate/BasicSpec.scala b/core/src/test/scala/io/kafkamate/BasicSpec.scala
index 9456aca..3073da2 100644
--- a/core/src/test/scala/io/kafkamate/BasicSpec.scala
+++ b/core/src/test/scala/io/kafkamate/BasicSpec.scala
@@ -3,16 +3,17 @@ package io.kafkamate
import zio.ZIO
import zio.test.*
import zio.test.Assertion.equalTo
+import zio.test.ZIOSpecDefault
/**
* This is a sample test for the CI; TODO delete after we start creating real tests for the app
*/
-object BasicSpec extends DefaultRunnableSpec {
+object BasicSpec extends ZIOSpecDefault {
- override def spec: ZSpec[_root_.zio.test.environment.TestEnvironment, Any] =
+ override def spec =
suite("Simple test")(
- testM("Check zio result") {
- assertM(ZIO.succeed(1))(equalTo(1))
+ test("Check zio result") {
+ assertTrue(1 == 1)
}
)
}
diff --git a/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala b/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala
index 66fe5fe..7cb3580 100644
--- a/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala
+++ b/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala
@@ -1,31 +1,24 @@
package io.kafkamate
package kafka
package consumer
-/*
+/*
import zio.*
-import zio.blocking.Blocking
-import zio.clock.Clock
-import zio.duration.*
import zio.test.Assertion.*
import zio.test.TestAspect.*
-import zio.test.environment.*
import zio.logging.*
-import zio.console.*
-import zio.test.{DefaultRunnableSpec, _}
import util.{HelperSpec, KafkaEmbedded}
-object ConsumerSpec extends DefaultRunnableSpec with HelperSpec {
+object ConsumerSpec extends ZIOSpecDefault with HelperSpec {
import KafkaConsumer.*
import utils.Logger
import messages.*
val testLayer
- : Layer[TestFailure[Throwable], Clock with Blocking with Logging with StringProducer with KafkaConsumer] =
+ : Layer[TestFailure[Throwable], StringProducer with KafkaConsumer] =
(Clock.live >+>
Console.live >+>
- Blocking.live >+>
KafkaEmbedded.Kafka.embedded >+>
stringProducer >+>
testConfigLayer >+>
@@ -40,8 +33,8 @@ object ConsumerSpec extends DefaultRunnableSpec with HelperSpec {
kvs = (1 to 5).toList.map(i => (s"key$i", s"msg$i"))
_ <- produceMany(topic, kvs)
records <- KafkaConsumer.consume(ConsumeRequest("test-id", topic, 5, "earliest", "")).runCollect
- } yield assert(records.map(v => (v.key, v.value)).toList)(equalTo(kvs.map(v => (v._1, v._2))))
+ } yield assertTrue(records.map(v => (v.key, v.value)).toList == kvs.map(v => (v._1, v._2)))
}
).provideLayerShared(testLayer) @@ timeout(30.seconds)
}
- */
+*/
diff --git a/core/src/test/scala/io/kafkamate/util/HelperSpec.scala b/core/src/test/scala/io/kafkamate/util/HelperSpec.scala
index be630a4..4835283 100644
--- a/core/src/test/scala/io/kafkamate/util/HelperSpec.scala
+++ b/core/src/test/scala/io/kafkamate/util/HelperSpec.scala
@@ -4,7 +4,6 @@ package util
/*
import org.apache.kafka.clients.producer.{ProducerRecord, RecordMetadata}
import zio.*
-import zio.blocking.Blocking
import zio.clock.Clock
import zio.kafka.producer.{Producer, ProducerSettings}
import zio.kafka.serde.{Serde, Serializer}
@@ -18,13 +17,12 @@ trait HelperSpec {
val producerSettings: URIO[Kafka, ProducerSettings] =
ZIO.access[Kafka](_.get.bootstrapServers).map(ProducerSettings(_))
- val stringProducer: ZLayer[Kafka with Blocking, Throwable, StringProducer] =
- (Blocking.any ++ producerSettings.toLayer ++ ZLayer.succeed(Serde.string: Serializer[Any, String])) >>>
+ val stringProducer: ZLayer[Kafka, Throwable, StringProducer] =
+ (producerSettings.toLayer ++ ZLayer.succeed(Serde.string: Serializer[Any, String])) >>>
Producer.live[Any, String, String]
- val testConfigLayer: URLayer[Clock with Blocking with Kafka, Clock with Blocking with ClustersConfigService] =
+ val testConfigLayer: URLayer[Clock with Kafka, Clock with ClustersConfigService] =
ZLayer.requires[Clock] ++
- ZLayer.requires[Blocking] ++
ZLayer.fromService[Kafka.Service, ClustersConfig.Service] { kafka =>
new ClustersConfig.Service {
def readClusters: Task[ClusterProperties] =
@@ -39,7 +37,7 @@ trait HelperSpec {
def produceMany(
topic: String,
kvs: Iterable[(String, String)]
- ): RIO[Blocking with StringProducer, Chunk[RecordMetadata]] =
+ ): RIO[StringProducer, Chunk[RecordMetadata]] =
Producer
.produceChunk[Any, String, String](Chunk.fromIterable(kvs.map { case (k, v) => new ProducerRecord(topic, k, v) }))
}
diff --git a/core/src/test/scala/io/kafkamate/util/KafkaEmbedded.scala b/core/src/test/scala/io/kafkamate/util/KafkaEmbedded.scala
index 3ea06bd..334609b 100644
--- a/core/src/test/scala/io/kafkamate/util/KafkaEmbedded.scala
+++ b/core/src/test/scala/io/kafkamate/util/KafkaEmbedded.scala
@@ -6,7 +6,7 @@ import net.manub.embeddedkafka.{EmbeddedK, EmbeddedKafka, EmbeddedKafkaConfig}
import zio.*
object KafkaEmbedded {
- type Kafka = Has[Kafka.Service]
+ type Kafka = Kafka.Service
object Kafka {
diff --git a/core/src/test/test.iml b/core/src/test/test.iml
new file mode 100644
index 0000000..0c150a6
--- /dev/null
+++ b/core/src/test/test.iml
@@ -0,0 +1,11 @@
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/fx/fx.iml b/fx/fx.iml
new file mode 100644
index 0000000..04c6faa
--- /dev/null
+++ b/fx/fx.iml
@@ -0,0 +1,91 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/fx/src/main/scala/io/km8/fx/Main.scala b/fx/src/main/scala/io/km8/fx/Main.scala
index 7fa9d73..c433127 100644
--- a/fx/src/main/scala/io/km8/fx/Main.scala
+++ b/fx/src/main/scala/io/km8/fx/Main.scala
@@ -1,9 +1,9 @@
package io.km8.fx
+import com.sun.javafx.application.PlatformImpl
import javafx.application.Platform
-import scala.collection.mutable.{Queue => SQueue}
+
import zio.*
-import zio.duration.*
import javafx.beans.property.ObjectProperty
import scalafx.application.JFXApp3
import scalafx.collections.ObservableBuffer
@@ -13,72 +13,40 @@ import scalafx.scene.control.*
import scalafx.scene.layout.*
import scalafx.scene.paint.*
import scalafx.scene.text.*
-import io.km8.fx.models.*
-import io.km8.fx.models.given
-import io.km8.fx.ui.{given, *}
-import io.km8.fx.ui.components.{given, *}
+import io.km8.fx.models.{*, given}
+import io.km8.fx.ui.{*, given}
+import io.km8.fx.ui.components.{*, given}
+import io.km8.fx.views.{*, given}
-import java.util.concurrent.Executor
import scala.concurrent.ExecutionContext
import scalafx.scene.input.KeyEvent
import scalafx.Includes.*
import scalafx.scene.input.KeyCode
-object Main extends JFXApp3 with BootstrapRuntime:
-
- val currentThreadEC = ExecutionContext.fromExecutor(new Executor {
- override def execute(command: Runnable): Unit = command.run()
- })
-
- private lazy val mkWindow =
- for
- header <- HeaderControl().render
- navigator <- NavigatorControl().render
- mainContent <- MainContentControl().render
- pane <- ZIO(new SplitPane {
- dividerPositions = 0
- id = "page-splitpane"
- items.addAll(navigator, mainContent)
- })
- p <- ZIO(new BorderPane {
- top = new VBox {
- vgrow = Priority.Always
- hgrow = Priority.Always
- children = header
- }
- center = new BorderPane {
- center = pane
- }
- })
- yield p
+object Main extends JFXApp3:
- private def mkScene(sceneRoot: Parent): ZIO[Has[EventsHub], Throwable, Scene] =
- for dispatchFocusOmni <- dispatchEvent
- yield new Scene(1366, 768) {
- stylesheets = List("css/app.css")
- root = sceneRoot
- onKeyReleased = k =>
- k.code match
- case KeyCode.Slash =>
- k.consume()
- dispatchFocusOmni(UIEvent.FocusOmni)
- case _ => ()
- }
+ val currentExe =
+ zio.Executor.fromJavaExecutor((command: Runnable) => command.run())
override def start(): Unit =
- val io: ZIO[Any, Throwable, JFXApp3.PrimaryStage] =
+
+ val io =
for
- h <- Hub.unbounded[UIEvent]
- hubLayer = ZLayer.succeed(h)
- main <- mkWindow.provideLayer(UI.make("") +!+ hubLayer)
- s <- mkScene(main).provideLayer(hubLayer)
- ret <- ZIO(new JFXApp3.PrimaryStage {
- title = "KM8"
- scene = s
- })
+ layers <- App.initialize(MainView)
+ (msgLayer, fxLayer) = layers
+ s <- MainStage()
+ .render
+ .tapError(e => ZIO.succeed(e.printStackTrace))
+ .orDie
+ .provide(msgLayer, fxLayer)
+ ret = new JFXApp3.PrimaryStage {
+ title = s"KM8 - ${Thread.currentThread()}"
+ scene = s
+ }
+ _ <- MsgBus.signal(Backend.LoadClusters, ViewState.empty).provide(msgLayer)
yield ret
- unsafeRun(
- io
- .on(currentThreadEC)
- .exitCode
- )
+
+ stage =
+ Unsafe.unsafeCompat { implicit u: Unsafe =>
+ Runtime.default.unsafe.run(io.onExecutor(currentExe)).getOrThrow()
+ }
diff --git a/fx/src/main/scala/io/km8/fx/models/UI.scala b/fx/src/main/scala/io/km8/fx/models/UI.scala
index 21fb6e2..2340ebc 100644
--- a/fx/src/main/scala/io/km8/fx/models/UI.scala
+++ b/fx/src/main/scala/io/km8/fx/models/UI.scala
@@ -1,13 +1,14 @@
package io.km8.fx
package models
+import io.km8.fx.views.ViewState
+import javafx.application.Platform
import zio.*
-import zio.duration.*
-import zio.clock.*
-import zio.prelude.*
-import Assertion.*
+import zio.prelude.NonEmptyList
+import zio.stream.ZStream
import scala.annotation.implicitNotFound
+import scala.collection.mutable.Queue as SQueue
type TestSeed = String | Int | Long
@@ -60,18 +61,13 @@ def switch[P <: Page, S <: Page](p: Path[P, S]) = p match {
}
*/
-inline def isNotEmpty: Assertion[String] = hasLength(greaterThan(0))
+//inline def isNotEmpty: Assertion[String] = hasLength(greaterThan(0))
-object Offset extends Subtype[Long]
-type Offset = Offset.Type
+opaque type Offset = Long
-object Config extends Newtype[(String, String)]
-type Config = Config.Type
+opaque type Config = (String, String)
-object TopicName extends Newtype[String]:
- override inline def assertion = isNotEmpty
-
-type TopicName = TopicName.Type
+opaque type TopicName = String
/*
Default type class
@@ -80,7 +76,7 @@ trait Def[T]:
def apply(seed: TestSeed): T
given Def[TopicName] with
- def apply(seed: TestSeed): TopicName = TopicName.make(s"topic $seed").getOrElse(TopicName("nothing"))
+ def apply(seed: TestSeed): TopicName = s"topic $seed"
def gen[T: Def](seed: TestSeed = "test"): T = summon[Def[T]].apply(seed)
@@ -108,7 +104,7 @@ case class Partition(
size: Long)
given Def[Partition] with
- def apply(seed: TestSeed) = Partition(0, gen(seed), NonEmptyList(gen(seed)), Offset(0), Offset(0), 0)
+ def apply(seed: TestSeed) = Partition(0, gen(seed), NonEmptyList(gen(seed)), 0L, 0L, 0)
case class Topic(
name: TopicName,
@@ -121,7 +117,7 @@ given Def[Topic] with
def apply(seed: TestSeed) =
Topic(
- TopicName.make(s"topic $seed").getOrElse(TopicName("topic")),
+ s"topic $seed",
NonEmptyList(gen(seed)),
Nil,
MessageEncoding.String,
@@ -131,7 +127,7 @@ given Def[Topic] with
case class PartitionOffset(partition: Partition, offset: Offset)
given Def[PartitionOffset] with
- def apply(seed: TestSeed) = PartitionOffset(gen(seed), Offset(0))
+ def apply(seed: TestSeed) = PartitionOffset(gen(seed), 0L)
case class ConsumerGroup(
name: String,
@@ -144,7 +140,7 @@ given Def[ConsumerGroup] with
ConsumerGroup(s"name $seed", gen(seed), List.range(0, 10).map(gen))
given Def[Config] with
- def apply(seed: TestSeed) = Config(s"key_$seed" -> s"value $seed")
+ def apply(seed: TestSeed): Config = s"key_$seed" -> s"value $seed"
case class Cluster(
name: String,
@@ -158,7 +154,6 @@ case class Cluster(
given Def[Cluster] with
def apply(seed: TestSeed) =
- val brokers = NonEmptyList.fromIterable(1, List.range(2, 5)).map(gen[Broker])
Cluster(
name = s"Cluster $seed",
kafkaHosts = NonEmptyList(gen(seed)),
@@ -175,22 +170,21 @@ case class UI(
data: List[Cluster],
config: UIConfig)
-enum UIEvent:
- case FocusOmni
-
-type EventsHub = Hub[UIEvent]
-
-val EventsHub = Hub
-
object UI:
- def make(seed: TestSeed = ""): URLayer[Any, Has[UI]] =
- UIO(
+ def make(seed: TestSeed = ""): UI =
+ UI(
+ data = List.range(1, 4).map(gen),
+ config = UIConfig(leftWidth = 300)
+ )
+
+ def makeLayer(seed: TestSeed = ""): URLayer[Any, UI] =
+ ZLayer.succeed(
UI(
data = List.range(1, 4).map(gen),
config = UIConfig(leftWidth = 300)
)
- ).toLayer
+ )
case class Message(
key: Array[Byte],
@@ -199,7 +193,101 @@ case class Message(
case class MessageHeader(key: String, value: Array[Byte])
-type UIEnv = Has[UI] & Has[EventsHub]
+object Test {
+ val stuff: String = ""
+}
+
+trait Msg
+
+enum Backend extends Msg:
+ case LoadClusters
+ case LoadConfig
+ case FocusOmni
+ case Search(search: String)
+ case LoadTopics
+ case KeyPressed(key: Char)
+
+enum Signal extends Msg:
+ case Nop
+ case Search
+ case ChangedClusters
+
+case class EventData[+S](state: Option[S], msg: Option[Msg])
+
+object EventData:
+ def apply[S](state: S): EventData[S] = EventData(Some(state), None)
+ def apply(msg: Msg): EventData[Nothing] = EventData(None, Some(msg))
+
+type MsgBus[S] = Hub[EventData[S]]
+
+object MsgBus:
+ def layer[S: Tag] = ZLayer.fromZIO(Hub.unbounded[(Option[S], Option[Msg])])
+
+ def signal[S: Tag](m: Msg, s: S) =
+ ZIO.service[MsgBus[S]].flatMap { hub =>
+ hub.publish(EventData(Some(s), Some(m)))
+ }
+
+type EventsQ[S] = SQueue[EventData[S]]
+
+def fireFX[S](m: Option[Msg], s: Option[S]): EventsQ[S] ?=> Unit =
+ summon[EventsQ[S]].enqueue(EventData(s, m))
+
+def fireFX[S](m: Msg, s: Option[S] = None): EventsQ[S] ?=> Unit =
+ fireFX(Some(m), s)
+
+import io.km8.fx.views.*
+
+object App:
+ val schedule = Schedule.spaced(100.millis)
+
+ def initialize[S: Tag, V <: View[S]](v: View[S]): UIO[(ULayer[MsgBus[S]], ULayer[EventsQ[S]])] =
+ for
+ hub <- Hub.unbounded[EventData[S]]
+ hubLayer = ZLayer.succeed(hub)
+ _ <- v.init.forkDaemon.provide(hubLayer)
+ eventsQ <- ZIO.succeed(SQueue.empty[EventData[S]])
+ _ <- ZIO
+ .succeed(eventsQ.dequeueAll(_ => true))
+ .flatMap(hub.publishAll)
+ .repeat(schedule)
+ .forkDaemon
+ yield (hubLayer, ZLayer.succeed(eventsQ))
+
+extension (c: => Unit) def fx = ZIO.succeed(Platform.runLater(() => c))
+
+type Update[S] = EventData[S] => UIO[EventData[S]]
+
+object Update:
+ def state[S](state: S): UIO[(Option[S], Option[Msg])] =
+ ZIO.succeed(Some(state), None)
+
+ def stateZIO[S](state: UIO[S]): UIO[(Option[S], Option[Msg])] =
+ state.map(s => Some(s) -> None)
+
+ def apply[S](state: S, msg: Msg): UIO[EventData[S]] =
+ ZIO.succeed(EventData(Some(state), Some(msg)))
+
+ def none[S]: UIO[EventData[S]] = ZIO.succeed(EventData(None , None))
+
+def publishMessage[S: Tag](msg: EventData[S]): URIO[MsgBus[S], Unit] =
+ ZIO.service[MsgBus[S]].flatMap(_.publish(msg)).unit
+
+def registerCallbackAsync[S: Tag](sender: Object, cb: Update[S]) =
+ registerCallback(sender, cb).forkDaemon
+
+def registerCallback[S: Tag](sender: Object, cb: Update[S]): URIO[MsgBus[S], Unit] =
+ for
+ hub <- ZIO.service[MsgBus[S]]
+ _ <- ZIO.debug(s"registering ${sender.toString} - ${Thread.currentThread()}")
+ _ <- ZStream.fromHub(hub).foreach { m =>
+ for
+ res <- cb(m)
+ _ <- res match
+ case EventData(s, m) if s.isDefined || m.isDefined =>
+ ZIO.debug(s"Sending $res") *> hub.publish(res)
+ case _ => Update.none
-def dispatchEvent: ZIO[Has[EventsHub], Nothing, UIEvent => Unit] =
- ZIO.service[EventsHub].map(hub => event => Runtime.global.unsafeRun(hub.publish(event)))
+ yield ()
+ }
+ yield ()
diff --git a/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala b/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala
index 0addfde..8ee78cd 100644
--- a/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala
+++ b/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala
@@ -1,13 +1,16 @@
package io.km8.fx
package ui
+import io.km8.fx.models._
import scalafx.scene.Node
import scalafx.scene.control.Alert.AlertType
import scalafx.scene.control.{Alert, ButtonType}
import zio.*
-trait BaseControl[R <: Has[_]]:
- def render: ZIO[R, Throwable, Node]
+trait BaseControl[S: Tag, R <: MsgBus[S]]:
+
+ def render: ZIO[R , Throwable, Node]
def alert(text: Any) =
new Alert(AlertType.None, text.toString, ButtonType.OK).show()
+
diff --git a/fx/src/main/scala/io/km8/fx/ui/components/ClusterListControl.scala b/fx/src/main/scala/io/km8/fx/ui/components/ClusterListControl.scala
index 2680ae8..7d3dd95 100644
--- a/fx/src/main/scala/io/km8/fx/ui/components/ClusterListControl.scala
+++ b/fx/src/main/scala/io/km8/fx/ui/components/ClusterListControl.scala
@@ -8,8 +8,9 @@ import zio.prelude.*
import scalafx.scene.Node
import models.*
+import views.*
-class ClusterListControl extends BaseControl[Has[UI]] {
+class ClusterListControl extends BaseControl[ViewState, UI & MsgBus[ViewState]] {
val mkList = for {
ui <- ZIO.service[UI]
@@ -18,5 +19,9 @@ class ClusterListControl extends BaseControl[Has[UI]] {
private[components] val view = mkList
- override def render: RIO[Has[UI], Node] = view
+ override def render =
+ registerCallbackAsync(this, update ) *> view
+
+ val update: Update[ViewState] =
+ case s => ZIO.debug(s"${this.getClass} - $s") *> Update.none
}
diff --git a/fx/src/main/scala/io/km8/fx/ui/components/HeaderControl.scala b/fx/src/main/scala/io/km8/fx/ui/components/HeaderControl.scala
index c797b65..c43f028 100644
--- a/fx/src/main/scala/io/km8/fx/ui/components/HeaderControl.scala
+++ b/fx/src/main/scala/io/km8/fx/ui/components/HeaderControl.scala
@@ -8,11 +8,12 @@ import scalafx.scene.control.*
import scalafx.scene.image.{Image, ImageView}
import zio.*
import zio.stream.*
-
import models.*
import scalafx.beans.property.ReadOnlyDoubleProperty
+import io.km8.fx.views.*
+import javafx.application.Platform
-class HeaderControl extends BaseControl[UIEnv]:
+class HeaderControl extends BaseControl[ViewState, MsgBus[ViewState] & EventsQ[ViewState]]:
lazy val omniBar =
new TextField {
@@ -20,41 +21,45 @@ class HeaderControl extends BaseControl[UIEnv]:
prefWidth = 600
}
- lazy val button = new Button {
- id = "search-omni"
- text = "Search"
- onMouseClicked = _ => alert("test")
+ val buttonZIO =
+ for {
+ given EventsQ[ViewState] <- ZIO.service[EventsQ[ViewState]]
+ button <- ZIO.attempt(new Button {
+ id = "search-omni"
+ text = "Search"
+ onMouseClicked = _ =>
+ fireFX(Backend.Search(omniBar.text.value))
+ })
+ } yield button
+
+ val update: Update[ViewState] = {
+ case EventData(_ , Some(Backend.FocusOmni)) =>
+ omniBar.requestFocus().fx *> Update.none
+ case _ => Update.none
}
- private[ui] val view =
- for
- hub <- ZIO.service[EventsHub]
- _ <- ZStream
- .fromHub(hub)
- .foreach { case UIEvent.FocusOmni =>
- ZIO {
- omniBar.requestFocus()
- }
- }
- .forkDaemon
- yield new ToolBar {
- prefHeight = 76
- maxHeight = 76
- id = "mainToolBar"
- content = List(
- new ImageView {
- image = new Image(
- this.getClass.getResourceAsStream("/images/logo.png"),
- 200,
- 100,
- true,
- true
+ override def render =
+ for {
+ button <- buttonZIO
+ toolBar =
+ new ToolBar {
+ prefHeight = 76
+ maxHeight = 76
+ id = "mainToolBar"
+ content = List(
+ new ImageView {
+ image = new Image(
+ this.getClass.getResourceAsStream("/images/logo.png"),
+ 200,
+ 100,
+ true,
+ true
+ )
+ margin = Insets(0, 100, 0, 10)
+ },
+ omniBar,
+ button
)
- margin = Insets(0, 100, 0, 10)
- },
- omniBar,
- button
- )
- }
-
- override def render: RIO[UIEnv, Node] = view
+ }
+ _ <- registerCallbackAsync(this, update)
+ } yield toolBar
diff --git a/fx/src/main/scala/io/km8/fx/ui/components/MainContentControl.scala b/fx/src/main/scala/io/km8/fx/ui/components/MainContentControl.scala
index 6162bd1..5672eae 100644
--- a/fx/src/main/scala/io/km8/fx/ui/components/MainContentControl.scala
+++ b/fx/src/main/scala/io/km8/fx/ui/components/MainContentControl.scala
@@ -7,9 +7,14 @@ import scalafx.scene.control.{ScrollPane, TreeItem, TreeView}
import zio.*
import models.*
+import io.km8.fx.views.*
-class MainContentControl extends BaseControl[Has[UI]]:
+class MainContentControl extends BaseControl[ViewState, MsgBus[ViewState]]:
- private lazy val view = ZIO(new ScrollPane())
+ private lazy val view = ZIO.attempt(new ScrollPane())
- override def render: ZIO[Has[UI], Throwable, Node] = view
+ override def render =
+ registerCallbackAsync(this, update) *> view
+
+ val update: Update[ViewState] =
+ case EventData(_, m) => ZIO.debug(s"${this.getClass} - $m") *> Update.none
diff --git a/fx/src/main/scala/io/km8/fx/ui/components/MainStage.scala b/fx/src/main/scala/io/km8/fx/ui/components/MainStage.scala
new file mode 100644
index 0000000..54c4f76
--- /dev/null
+++ b/fx/src/main/scala/io/km8/fx/ui/components/MainStage.scala
@@ -0,0 +1,60 @@
+package io.km8.fx
+package ui
+package components
+
+import zio.*
+import scalafx.scene.control.SplitPane
+import scalafx.scene.layout.{BorderPane, Priority, VBox}
+import scalafx.scene.*
+import io.km8.fx.ui.BaseControl
+import scalafx.scene.input._
+import scalafx.Includes.*
+import views.*
+import models.*
+
+class MainStage:
+
+ private def mkWindow =
+ for
+// _ <- ZIO.debug(s"mkWindow - ${Thread.currentThread()}")
+ header <- HeaderControl().render
+ navigator <- NavigatorControl().render
+ mainContent <- MainContentControl().render
+ pane <- ZIO.attempt(new SplitPane {
+ dividerPositions = 0
+ id = "page-splitpane"
+ items.addAll(navigator, mainContent)
+ })
+ p <- ZIO.attempt(new BorderPane {
+ top = new VBox {
+ vgrow = Priority.Always
+ hgrow = Priority.Always
+ children = header
+ }
+ center = new BorderPane {
+ center = pane
+ }
+ })
+ yield p
+
+ def handler: Update[ViewState] =
+ case m => ZIO.debug(s"UI-$m from ${Thread.currentThread()}") *> Update.none
+
+
+ def render =
+ for {
+ given EventsQ[ViewState] <- ZIO.service[EventsQ[ViewState]]
+ _ <- registerCallbackAsync(this, handler)
+ sceneRoot <- mkWindow
+ res <- ZIO.attempt(
+ new Scene(1366, 768) {
+ stylesheets = List("css/app.css")
+ root = sceneRoot
+ onKeyReleased = k =>
+ k.code match
+ case KeyCode.Slash =>
+ k.consume()
+ fireFX(Backend.FocusOmni)
+ case _ => ()
+ })
+ } yield res
diff --git a/fx/src/main/scala/io/km8/fx/ui/components/NavigatorControl.scala b/fx/src/main/scala/io/km8/fx/ui/components/NavigatorControl.scala
index f5c04c9..f8e4c7b 100644
--- a/fx/src/main/scala/io/km8/fx/ui/components/NavigatorControl.scala
+++ b/fx/src/main/scala/io/km8/fx/ui/components/NavigatorControl.scala
@@ -2,27 +2,20 @@ package io.km8.fx
package ui
package components
+import javafx.application.*
+
+import scala.jdk.CollectionConverters.*
import scalafx.scene.Node
-import scalafx.scene.control.{ScrollPane, TreeItem, TreeView}
+import scalafx.scene.control.*
+import scalafx.Includes.*
import zio.*
import models.*
+import io.km8.fx.views.*
+import javafx.scene.layout.*
+import javafx.scene.paint.*
+import javafx.geometry.*
-class NavigatorControl extends BaseControl[Has[UI]]:
-
- private lazy val view =
- for
- ui <- ZIO.service[UI]
- tv <- treeView
- ret <- ZIO(
- new ScrollPane {
- minWidth = ui.config.leftWidth
- fitToWidth = true
- fitToHeight = true
- id = "page-tree"
- content = tv
- }
- )
- yield ret
+class NavigatorControl extends BaseControl[ViewState, MsgBus[ViewState]]:
private val clusterNode = (cluster: Cluster) =>
new TreeItem[String](cluster.name) {
@@ -35,25 +28,38 @@ class NavigatorControl extends BaseControl[Has[UI]]:
)
}
- private val treeView =
- for
- ui <- ZIO.service[UI]
- tv <- ZIO {
- new TreeView[String] {
- showRoot = false
- id = "left-tree"
- root = new TreeItem[String]("Clusters") {
- expanded = true
- children = ui.data.map(clusterNode).toSeq
- }
- }
- }
+ val update: Update[ViewState] =
+ case EventData(Some(state), Some(Signal.ChangedClusters)) =>
+ ZIO.attempt {
+ val nodes = state.clusterDetails.map(s => clusterNode(s).delegate)
+ treeView.root.value.getChildren.addAll(nodes: _*)
+ }.catchAll(e => ZIO.succeed(e.printStackTrace())) *> Update.none
+ case _ => Update.none
+
+ private lazy val treeView =
+ new TreeView[String] {
+ showRoot = true
+ id = "left-tree"
+ root = new TreeItem[String]("Clusters") {
+ expanded = true
+ }
+ }
// _ = tv.getSelectionModel.selectedItemProperty().addListener((obs, oldVal, newVal) => alert(newVal.getValue))
+/*
_ = tv.getSelectionModel
.selectedItemProperty()
- .addListener((obs, oldVal, newVal) => alert(ui.data.map(_.consumerGroups)))
- yield tv
+ .addListener((obs, oldVal, newVal) => ())
+*/
+
+ lazy val scrollPane =
+ new ScrollPane {
+ minWidth = 300
+ fitToWidth = true
+ fitToHeight = true
+ background = Background(BackgroundFill(Color.RED, CornerRadii(0), Insets.EMPTY))
+ id = "page-tree"
+ content = treeView
+ }
- override def render: ZIO[Has[UI], Throwable, Node] =
- for ret <- view
- yield ret
+ override def render =
+ registerCallbackAsync(this, update).as(scrollPane)
diff --git a/fx/src/main/scala/io/km8/fx/views/Views.scala b/fx/src/main/scala/io/km8/fx/views/Views.scala
new file mode 100644
index 0000000..a56e5e9
--- /dev/null
+++ b/fx/src/main/scala/io/km8/fx/views/Views.scala
@@ -0,0 +1,74 @@
+package io.km8.fx.views
+
+import zio.*
+import zio.stream.ZStream
+import io.km8.common.ClusterDetails
+import io.km8.fx.models.*
+import io.km8.fx.models.given
+
+object Data:
+
+ def loadClusters =
+ ZIO.succeed {
+ val c1 = gen[Cluster]()
+ val c2 = gen[Cluster]()
+ c1 :: c2 :: Nil
+ }
+
+trait View[S: Tag]:
+ val children: List[View[S]] = Nil
+
+ def init: ZIO[MsgBus[S], Nothing, Unit] =
+ for
+ _ <- registerCallbackAsync(this, update)
+ _ <- ZIO.foreach(children)(_.init)
+ yield ()
+
+ def update: Update[S]
+
+case class ViewState(
+ clusterDetails: List[Cluster],
+ currentCluster: Option[Cluster])
+
+object ViewState:
+ def empty = ViewState(Nil, None)
+
+object MainView extends View[ViewState]:
+ override val children = ClustersView :: SearchView :: Nil
+
+ override def update =
+ case _ => Update.none
+
+object ClustersView extends View[ViewState]:
+ override val children = TitleView :: Nil
+
+ override def update =
+ case EventData(Some(state), Some(Backend.LoadClusters)) =>
+ Data.loadClusters.flatMap { newClusters =>
+ ZIO
+ .debug("creating clusters") *>
+ Update(state.copy(clusterDetails = newClusters), Signal.ChangedClusters)
+ }
+ case EventData(_, Some(Backend.Search(search))) =>
+ ZIO.debug(s"Searched $search") *> Update.none
+ case EventData(_, m) =>
+ ZIO.debug(s"ClusterView update: $m") *> Update.none
+
+/*
+
+object TextInput extends View[String]:
+ override def update =
+ case state -> Backend.KeyPressed(k) =>
+ ZIO.succeed(state ++ 'c'.toString).
+*/
+
+
+object SearchView extends View[ViewState]:
+
+ override def update =
+ case EventData(_, m) => ZIO.debug(s"SearchView update: $m") *> Update.none
+
+object TitleView extends View[ViewState]:
+
+ override def update =
+ case EventData(_ , m) => ZIO.debug(s"TitleView update: $m") *> Update.none
diff --git a/project/plugins.sbt b/project/plugins.sbt
index b3d626e..8d769e4 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -7,3 +7,5 @@ addSbtPlugin("org.jmotor.sbt" % "sbt-dependency-updates" % "1.2.2")
addSbtPlugin("se.marcuslonnberg" % "sbt-docker" % "1.9.0")
addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "1.2.0")
addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.6")
+// project/plugins.sbt
+addSbtPlugin("ch.epfl.scala" % "sbt-scalafix" % "0.10.1")
\ No newline at end of file