From 8600fc55684d805596523cf10270011d41cf82e1 Mon Sep 17 00:00:00 2001 From: Octav Zaharia Date: Sat, 25 Jun 2022 09:58:55 +0300 Subject: [PATCH 1/6] Upgrade to ZIO 2 --- build.sbt | 12 +- .../io/km8/core/kafka/KafkaExplorerSpec.scala | 2 - .../io/km8/core/kafka/KafkaProducerSpec.scala | 15 +- .../it/scala/io/km8/core/kafka/itlayers.scala | 9 +- .../io/km8/core/config/ClustersConfig.scala | 21 +- .../km8/core/config/ConfigPathService.scala | 9 +- .../io/km8/core/kafka/KafkaConsumer.scala | 40 +-- .../io/km8/core/kafka/KafkaExplorer.scala | 240 +++++++++--------- .../io/km8/core/kafka/KafkaProducer.scala | 81 +++--- .../main/scala/io/km8/core/utils/Logger.scala | 8 +- .../test/scala/io/kafkamate/BasicSpec.scala | 7 +- .../kafka/consumer/ConsumerSpec.scala | 4 +- .../scala/io/kafkamate/util/HelperSpec.scala | 10 +- .../io/kafkamate/util/KafkaEmbedded.scala | 2 +- fx/src/main/scala/io/km8/fx/Main.scala | 23 +- fx/src/main/scala/io/km8/fx/models/UI.scala | 44 ++-- .../main/scala/io/km8/fx/ui/BaseControl.scala | 2 +- .../fx/ui/components/ClusterListControl.scala | 4 +- .../km8/fx/ui/components/HeaderControl.scala | 2 +- .../fx/ui/components/MainContentControl.scala | 6 +- .../fx/ui/components/NavigatorControl.scala | 8 +- .../main/scala/io/km8/fx/views/Clusters.scala | 3 + project/plugins.sbt | 2 + 23 files changed, 278 insertions(+), 276 deletions(-) create mode 100644 fx/src/main/scala/io/km8/fx/views/Clusters.scala diff --git a/build.sbt b/build.sbt index 3191a48..5355422 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-RC5" + val zioKafka = "2.0.0-M3" + val zioJson = "0.3.0-RC7" + val zioLogging = "2.0.0-RC8" + val zioPrelude = "1.0.0-RC13" val kafkaProtobuf = "7.1.0" val javaFx = "16" 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..a8c32ee 100644 --- a/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala +++ b/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala @@ -1,7 +1,6 @@ package io.km8.core.kafka import zio.* -import zio.blocking.Blocking import zio.clock.Clock import zio.duration.* import zio.kafka.consumer.* @@ -22,7 +21,6 @@ object KafkaExplorerSpec extends DefaultRunnableSpec: val layer = Clock.live >+> - Blocking.live >+> itlayers.kafkaContainer >+> itlayers.clusterConfig(clusterId) >+> KafkaExplorer.liveLayer >+> 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..400a185 100644 --- a/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala +++ b/core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala @@ -1,7 +1,6 @@ package io.km8.core.kafka import zio.* -import zio.blocking.Blocking import zio.clock.Clock import zio.duration.* import zio.kafka.consumer.* @@ -16,19 +15,19 @@ import org.apache.kafka.clients.consumer.OffsetResetStrategy object KafkaProducerSpec extends DefaultRunnableSpec: - 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] = + Clock.live ++ 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] = mainSpec - .provideSomeLayer[environment.TestEnvironment & Has[KafkaContainer]](specLayer ++ Clock.live) - .provideCustomLayerShared(Blocking.live >>> itlayers.kafkaContainer) + .provideSomeLayer[environment.TestEnvironment & KafkaContainer](specLayer ++ Clock.live) + .provideCustomLayerShared(itlayers.kafkaContainer) private val mainSpec = suite("Kafka services")( 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..ecbc6ed 100644 --- a/core/src/it/scala/io/km8/core/kafka/itlayers.scala +++ b/core/src/it/scala/io/km8/core/kafka/itlayers.scala @@ -3,7 +3,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,7 +11,7 @@ import io.km8.core.config.{ClusterConfig, ClusterProperties, ClusterSettings} object itlayers: - val kafkaContainer: ZLayer[Blocking, Nothing, Has[KafkaContainer]] = + val kafkaContainer: ZLayer[Any, Nothing, KafkaContainer] = ZManaged.make { effectBlocking { val container = new KafkaContainer() @@ -21,7 +20,7 @@ object itlayers: }.orDie }(container => effectBlocking(container.stop()).orDie).toLayer - def consumerSettings(cg: String): ZManaged[Has[KafkaContainer], Nothing, ConsumerSettings] = + def consumerSettings(cg: String): ZManaged[KafkaContainer, Nothing, ConsumerSettings] = ZIO .service[KafkaContainer] .map(c => @@ -31,10 +30,10 @@ object itlayers: ) .toManaged_ - def consumerLayer(cgroup: String): ZLayer[Has[KafkaContainer] with Clock with Blocking, Nothing, Has[Consumer]] = + def consumerLayer(cgroup: String): ZLayer[KafkaContainer with Clock , Nothing, Consumer] = consumerSettings(cgroup).flatMap(Consumer.make(_)).orDie.toLayer - def clusterConfig(clusterId: String): ZLayer[Has[KafkaContainer], Nothing, Has[ClusterConfig]] = + def clusterConfig(clusterId: String): ZLayer[KafkaContainer, Nothing, ClusterConfig] = ZIO .service[KafkaContainer] .map(kafkaContainer => 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..999d789 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..19721b8 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,9 @@ package io.km8.core package kafka import zio.* -import zio.blocking.Blocking -import zio.clock.Clock -import zio.duration.* + +import zio.Clock + import zio.kafka.consumer.* import zio.kafka.consumer.Consumer.* import zio.kafka.serde.Deserializer @@ -28,19 +28,23 @@ trait KafkaConsumer { object KafkaConsumer { - lazy val liveLayer: URLayer[Clock with Blocking with Has[ClusterConfig], Has[KafkaConsumer]] = - (KafkaConsumerLive(_, _, _)).toLayer + lazy val liveLayer: URLayer[Clock with ClusterConfig, KafkaConsumer] = + ZLayer { + for { + clock <- ZIO.service[Clock] + config <- ZIO.service[ClusterConfig] + } yield KafkaConsumerLive(clock, 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, + clock: Clock, 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 +59,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 +72,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 +84,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 +102,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,7 +118,7 @@ object KafkaConsumer { else withFilter.take(request.maxResults) withFilterLimit.provideLayer( - clockLayer ++ blockingLayer >>> + clockLayer >>> 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..3316f66 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,9 @@ package io.km8.core package kafka import zio.* -import zio.blocking.Blocking -import zio.clock.Clock -import zio.duration.* + +import zio.Clock + import zio.kafka.admin.* import zio.kafka.admin.AdminClient.* @@ -24,126 +24,130 @@ object KafkaExplorer extends Accessible[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] = + lazy val liveLayer: ZLayer[Clock & ClusterConfig, Nothing, KafkaExplorer] = + ZLayer{ + for { + cl <- ZIO.service[Clock] + cc <- ZIO.service[ClusterConfig] + } yield KafkaExplorerLive(cl, cc) + } + + +case class KafkaExplorerLive(clock: Clock, clustersConfigService: ClusterConfig) extends KafkaExplorer { + val clockLayer: ULayer[Clock] = ZLayer.succeed(clock) + + 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[Clock, 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) ++ clockLayer) + } + + 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..c82628b 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,35 @@ object KafkaProducer { value: String )( clusterId: String - ): RIO[KafkaProducer, Unit] = ZIO.accessM[KafkaProducer](_.get.produce(topic, key, value)(clusterId)) + ): ZIO[Scope & KafkaProducer, Throwable, Unit] = ZIO.environmentWithZIO[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 { + lazy val liveLayer: URLayer[ClusterConfig & Scope, KafkaProducer] =ZLayer.fromZIO( for { + clusterConfigService <- ZIO.service[ClusterConfig] + } yield new KafkaProducer { - lazy val serdeLayer: ULayer[Has[Serializer[Any, String]]] = - ZLayer.succeed(Serde.string) + lazy val serdeLayer: ULayer[Serializer[Any, String]] = + ZLayer.succeed(Serde.string) - def settingsLayer(clusterId: String): Task[ProducerSettings] = - clusterConfigService - .getCluster(clusterId) - .map(c => ProducerSettings(c.kafkaHosts)) + def settingsLayer(clusterId: String): Task[ProducerSettings] = + clusterConfigService + .getCluster(clusterId) + .map(c => ProducerSettings(c.kafkaHosts)) - 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 producerLayer(clusterId: String): ZIO[Scope, Throwable, Producer] = + settingsLayer(clusterId) + .flatMap(settings => Producer.make(settings)) - 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 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))) + } + ) +} \ No newline at end of file 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..3936324 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..cb94812 100644 --- a/core/src/test/scala/io/kafkamate/BasicSpec.scala +++ b/core/src/test/scala/io/kafkamate/BasicSpec.scala @@ -3,15 +3,16 @@ 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") { + test("Check zio result") { assertM(ZIO.succeed(1))(equalTo(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..f3aa435 100644 --- a/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala +++ b/core/src/test/scala/io/kafkamate/kafka/consumer/ConsumerSpec.scala @@ -4,7 +4,6 @@ package consumer /* import zio.* -import zio.blocking.Blocking import zio.clock.Clock import zio.duration.* import zio.test.Assertion.* @@ -22,10 +21,9 @@ object ConsumerSpec extends DefaultRunnableSpec with HelperSpec { import messages.* val testLayer - : Layer[TestFailure[Throwable], Clock with Blocking with Logging with StringProducer with KafkaConsumer] = + : Layer[TestFailure[Throwable], Clock with StringProducer with KafkaConsumer] = (Clock.live >+> Console.live >+> - Blocking.live >+> KafkaEmbedded.Kafka.embedded >+> stringProducer >+> testConfigLayer >+> 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/fx/src/main/scala/io/km8/fx/Main.scala b/fx/src/main/scala/io/km8/fx/Main.scala index 7fa9d73..ba78694 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 javafx.application.Platform -import scala.collection.mutable.{Queue => SQueue} + +import scala.collection.mutable.Queue as SQueue import zio.* -import zio.duration.* import javafx.beans.property.ObjectProperty import scalafx.application.JFXApp3 import scalafx.collections.ObservableBuffer @@ -15,8 +15,8 @@ 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.ui.{*, given} +import io.km8.fx.ui.components.{*, given} import java.util.concurrent.Executor import scala.concurrent.ExecutionContext @@ -24,7 +24,7 @@ import scalafx.scene.input.KeyEvent import scalafx.Includes.* import scalafx.scene.input.KeyCode -object Main extends JFXApp3 with BootstrapRuntime: +object Main extends JFXApp3: val currentThreadEC = ExecutionContext.fromExecutor(new Executor { override def execute(command: Runnable): Unit = command.run() @@ -35,12 +35,12 @@ object Main extends JFXApp3 with BootstrapRuntime: header <- HeaderControl().render navigator <- NavigatorControl().render mainContent <- MainContentControl().render - pane <- ZIO(new SplitPane { + pane <- ZIO.attempt(new SplitPane { dividerPositions = 0 id = "page-splitpane" items.addAll(navigator, mainContent) }) - p <- ZIO(new BorderPane { + p <- ZIO.attempt(new BorderPane { top = new VBox { vgrow = Priority.Always hgrow = Priority.Always @@ -52,7 +52,7 @@ object Main extends JFXApp3 with BootstrapRuntime: }) yield p - private def mkScene(sceneRoot: Parent): ZIO[Has[EventsHub], Throwable, Scene] = + private def mkScene(sceneRoot: Parent): ZIO[EventsHub, Throwable, Scene] = for dispatchFocusOmni <- dispatchEvent yield new Scene(1366, 768) { stylesheets = List("css/app.css") @@ -72,13 +72,14 @@ object Main extends JFXApp3 with BootstrapRuntime: hubLayer = ZLayer.succeed(h) main <- mkWindow.provideLayer(UI.make("") +!+ hubLayer) s <- mkScene(main).provideLayer(hubLayer) - ret <- ZIO(new JFXApp3.PrimaryStage { + ret <- ZIO.attempt(new JFXApp3.PrimaryStage { title = "KM8" scene = s }) yield ret - unsafeRun( + + Runtime.default.unsafeRun( io - .on(currentThreadEC) + .onExecutionContext(currentThreadEC) .exitCode ) 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..66921d9 100644 --- a/fx/src/main/scala/io/km8/fx/models/UI.scala +++ b/fx/src/main/scala/io/km8/fx/models/UI.scala @@ -2,10 +2,7 @@ package io.km8.fx package models import zio.* -import zio.duration.* -import zio.clock.* -import zio.prelude.* -import Assertion.* +import zio.prelude.NonEmptyList import scala.annotation.implicitNotFound @@ -60,18 +57,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 +72,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 +100,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 +113,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 +123,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 +136,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, @@ -180,17 +172,17 @@ enum UIEvent: type EventsHub = Hub[UIEvent] -val EventsHub = Hub +val EventsHub = zio.Hub object UI: - def make(seed: TestSeed = ""): URLayer[Any, Has[UI]] = - UIO( + def make(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 +191,11 @@ case class Message( case class MessageHeader(key: String, value: Array[Byte]) -type UIEnv = Has[UI] & Has[EventsHub] +type UIEnv = UI & EventsHub -def dispatchEvent: ZIO[Has[EventsHub], Nothing, UIEvent => Unit] = +def dispatchEvent: ZIO[EventsHub, Nothing, UIEvent => Unit] = ZIO.service[EventsHub].map(hub => event => Runtime.global.unsafeRun(hub.publish(event))) + +object Test { + val stuff: String = "" +} \ No newline at end of file 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..cc36f45 100644 --- a/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala +++ b/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala @@ -6,7 +6,7 @@ import scalafx.scene.control.Alert.AlertType import scalafx.scene.control.{Alert, ButtonType} import zio.* -trait BaseControl[R <: Has[_]]: +trait BaseControl[R ]: def render: ZIO[R, Throwable, Node] def alert(text: Any) = 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..d791a41 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 @@ -9,7 +9,7 @@ import scalafx.scene.Node import models.* -class ClusterListControl extends BaseControl[Has[UI]] { +class ClusterListControl extends BaseControl[UI] { val mkList = for { ui <- ZIO.service[UI] @@ -18,5 +18,5 @@ class ClusterListControl extends BaseControl[Has[UI]] { private[components] val view = mkList - override def render: RIO[Has[UI], Node] = view + override def render: RIO[UI, Node] = view } 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..7257c57 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 @@ -32,7 +32,7 @@ class HeaderControl extends BaseControl[UIEnv]: _ <- ZStream .fromHub(hub) .foreach { case UIEvent.FocusOmni => - ZIO { + ZIO.succeed { omniBar.requestFocus() } } 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..a404185 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 @@ -8,8 +8,8 @@ import zio.* import models.* -class MainContentControl extends BaseControl[Has[UI]]: +class MainContentControl extends BaseControl[UI]: - private lazy val view = ZIO(new ScrollPane()) + private lazy val view = ZIO.succeed(new ScrollPane()) - override def render: ZIO[Has[UI], Throwable, Node] = view + override def render: ZIO[UI, Throwable, Node] = view 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..e28712f 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 @@ -7,13 +7,13 @@ import scalafx.scene.control.{ScrollPane, TreeItem, TreeView} import zio.* import models.* -class NavigatorControl extends BaseControl[Has[UI]]: +class NavigatorControl extends BaseControl[UI]: private lazy val view = for ui <- ZIO.service[UI] tv <- treeView - ret <- ZIO( + ret <- ZIO.succeed( new ScrollPane { minWidth = ui.config.leftWidth fitToWidth = true @@ -38,7 +38,7 @@ class NavigatorControl extends BaseControl[Has[UI]]: private val treeView = for ui <- ZIO.service[UI] - tv <- ZIO { + tv <- ZIO.succeed { new TreeView[String] { showRoot = false id = "left-tree" @@ -54,6 +54,6 @@ class NavigatorControl extends BaseControl[Has[UI]]: .addListener((obs, oldVal, newVal) => alert(ui.data.map(_.consumerGroups))) yield tv - override def render: ZIO[Has[UI], Throwable, Node] = + override def render: ZIO[UI, Throwable, Node] = for ret <- view yield ret diff --git a/fx/src/main/scala/io/km8/fx/views/Clusters.scala b/fx/src/main/scala/io/km8/fx/views/Clusters.scala new file mode 100644 index 0000000..d310716 --- /dev/null +++ b/fx/src/main/scala/io/km8/fx/views/Clusters.scala @@ -0,0 +1,3 @@ +package io.km8.fx.views + +case class Clusters(list: List[String]) \ No newline at end of file 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 From 2b5fb6fd9844a1b12bd24287a86b5310ed773a2d Mon Sep 17 00:00:00 2001 From: Octav Zaharia Date: Thu, 30 Jun 2022 19:22:11 +0300 Subject: [PATCH 2/6] Fix some tests --- build.sbt | 10 ++-- .../io/km8/core/kafka/KafkaExplorerSpec.scala | 27 +++++---- .../io/km8/core/kafka/KafkaProducerSpec.scala | 15 ++--- .../it/scala/io/km8/core/kafka/itlayers.scala | 57 ++++++++++--------- .../io/km8/core/config/ClustersConfig.scala | 2 +- .../io/km8/core/kafka/KafkaConsumer.scala | 13 +---- .../io/km8/core/kafka/KafkaExplorer.scala | 33 ++++++----- .../io/km8/core/kafka/KafkaProducer.scala | 18 +++--- .../main/scala/io/km8/core/utils/Logger.scala | 4 +- .../test/scala/io/kafkamate/BasicSpec.scala | 2 +- .../kafka/consumer/ConsumerSpec.scala | 15 ++--- fx/src/main/scala/io/km8/fx/Main.scala | 36 +++++++----- fx/src/main/scala/io/km8/fx/models/UI.scala | 13 ++++- .../main/scala/io/km8/fx/ui/BaseControl.scala | 2 +- .../km8/fx/ui/components/HeaderControl.scala | 2 +- .../fx/ui/components/MainContentControl.scala | 2 +- .../fx/ui/components/NavigatorControl.scala | 4 +- .../main/scala/io/km8/fx/views/Clusters.scala | 55 +++++++++++++++++- 18 files changed, 186 insertions(+), 124 deletions(-) diff --git a/build.sbt b/build.sbt index 5355422..6f195b3 100644 --- a/build.sbt +++ b/build.sbt @@ -7,11 +7,11 @@ lazy val ProjectScalaVersion = "3.1.3" lazy val Versions = new { - val zio = "2.0.0-RC5" - val zioKafka = "2.0.0-M3" - val zioJson = "0.3.0-RC7" - val zioLogging = "2.0.0-RC8" - val zioPrelude = "1.0.0-RC13" + 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/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala b/core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala index a8c32ee..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,27 +1,24 @@ package io.km8.core.kafka import zio.* -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 >+> - itlayers.kafkaContainer >+> + itlayers.kafkaContainer >+> itlayers.clusterConfig(clusterId) >+> KafkaExplorer.liveLayer >+> itlayers.consumerLayer(consumerGroup) >+> @@ -29,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 400a185..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,8 +1,6 @@ package io.km8.core.kafka import zio.* -import zio.clock.Clock -import zio.duration.* import zio.kafka.consumer.* import zio.kafka.serde.Serde import zio.test.* @@ -10,13 +8,12 @@ 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[KafkaContainer, Nothing, KafkaConsumer] = - Clock.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaConsumer.liveLayer + itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaConsumer.liveLayer private val producerLayer: ZLayer[KafkaContainer, Nothing, KafkaProducer] = itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaProducer.liveLayer @@ -24,14 +21,14 @@ object KafkaProducerSpec extends DefaultRunnableSpec: 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 & KafkaContainer](specLayer ++ Clock.live) + .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( @@ -48,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 ecbc6ed..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,7 +2,6 @@ package io.km8.core.kafka import com.dimafeng.testcontainers.KafkaContainer import zio.* -import zio.clock.Clock import zio.kafka.consumer.Consumer import zio.kafka.consumer.Consumer.{AutoOffsetStrategy, OffsetRetrieval} import zio.kafka.consumer.ConsumerSettings @@ -12,15 +11,15 @@ import io.km8.core.config.{ClusterConfig, ClusterProperties, ClusterSettings} object itlayers: val kafkaContainer: ZLayer[Any, Nothing, KafkaContainer] = - ZManaged.make { - effectBlocking { + 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[KafkaContainer, Nothing, ConsumerSettings] = + def consumerSettings(cg: String): ZIO[KafkaContainer with Scope, Nothing, ConsumerSettings] = ZIO .service[KafkaContainer] .map(c => @@ -28,33 +27,35 @@ object itlayers: .withGroupId(cg) .withOffsetRetrieval(OffsetRetrieval.Auto(AutoOffsetStrategy.Earliest)) ) - .toManaged_ - def consumerLayer(cgroup: String): ZLayer[KafkaContainer with Clock , Nothing, Consumer] = - consumerSettings(cgroup).flatMap(Consumer.make(_)).orDie.toLayer + 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] = - 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 + 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/scala/io/km8/core/config/ClustersConfig.scala b/core/src/main/scala/io/km8/core/config/ClustersConfig.scala index 999d789..66a8f92 100644 --- a/core/src/main/scala/io/km8/core/config/ClustersConfig.scala +++ b/core/src/main/scala/io/km8/core/config/ClustersConfig.scala @@ -52,7 +52,7 @@ object ClustersConfig { def writeClusters(cluster: ClusterSettings) = ZIO.serviceWithZIO[ClusterConfig](_.writeClusters(cluster)) def deleteCluster(clusterId: String) = ZIO.serviceWithZIO[ClusterConfig](_.deleteCluster(clusterId)) - lazy val liveLayer: URLayer[ConfigPath , ClusterConfig] = + lazy val liveLayer: URLayer[ConfigPath, ClusterConfig] = ZLayer(ZIO.service[ConfigPath].map(ClusterConfigLive.apply)) case class ClusterConfigLive(configPath: ConfigPath) extends ClusterConfig { 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 19721b8..eecf985 100644 --- a/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala +++ b/core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala @@ -3,8 +3,6 @@ package kafka import zio.* -import zio.Clock - import zio.kafka.consumer.* import zio.kafka.consumer.Consumer.* import zio.kafka.serde.Deserializer @@ -28,24 +26,20 @@ trait KafkaConsumer { object KafkaConsumer { - lazy val liveLayer: URLayer[Clock with ClusterConfig, KafkaConsumer] = + lazy val liveLayer: URLayer[ClusterConfig, KafkaConsumer] = ZLayer { for { - clock <- ZIO.service[Clock] config <- ZIO.service[ClusterConfig] - } yield KafkaConsumerLive(clock, config) + } yield KafkaConsumerLive(config) } def consume(request: ConsumeRequest): ZStream[KafkaConsumer, Throwable, Message] = ZStream.environmentWithStream[KafkaConsumer](_.get.consume(request)) case class KafkaConsumerLive( - clock: Clock, clustersConfigService: ClusterConfig) extends KafkaConsumer { - private val clockLayer = ZLayer.succeed(clock) - // TODO Ciprian transform to enum or reuse the Consumer enum + serdes private def extractOffsetStrategy(offsetValue: String): AutoOffsetStrategy = offsetValue.toLowerCase match { @@ -118,8 +112,7 @@ object KafkaConsumer { else withFilter.take(request.maxResults) withFilterLimit.provideLayer( - clockLayer >>> - 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 3316f66..afeaee8 100644 --- a/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala +++ b/core/src/main/scala/io/km8/core/kafka/KafkaExplorer.scala @@ -3,8 +3,6 @@ package kafka import zio.* -import zio.Clock - import zio.kafka.admin.* import zio.kafka.admin.AdminClient.* @@ -19,22 +17,28 @@ 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: ZLayer[Clock & ClusterConfig, Nothing, KafkaExplorer] = - ZLayer{ + 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 { - cl <- ZIO.service[Clock] cc <- ZIO.service[ClusterConfig] - } yield KafkaExplorerLive(cl, cc) + } yield KafkaExplorerLive(cc) } - -case class KafkaExplorerLive(clock: Clock, clustersConfigService: ClusterConfig) extends KafkaExplorer { - val clockLayer: ULayer[Clock] = ZLayer.succeed(clock) +case class KafkaExplorerLive(clustersConfigService: ClusterConfig) extends KafkaExplorer { private def adminClientLayer(clusterId: String) = ZLayer.fromZIO { @@ -44,12 +48,12 @@ case class KafkaExplorerLive(clock: Clock, clustersConfigService: ClusterConfig) } yield client } - def withAdminClient[A](clusterId: String)(eff: AdminClient => RIO[Clock, A]): ZIO[Any, Throwable, A] = + 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(adminClientLayer(clusterId) ++ clockLayer) + .provideLayer(adminClientLayer(clusterId)) } override def listBrokers(clusterId: String) = @@ -141,7 +145,10 @@ case class KafkaExplorerLive(clock: Clock, clustersConfigService: ClusterConfig) end listConsumerGroups - override def listConsumerOffsets(clusterId: String, groupId: String): ZIO[Any, Throwable,ConsumerGroupOffsetsResponse] = + override def listConsumerOffsets( + clusterId: String, + groupId: String + ): ZIO[Any, Throwable, ConsumerGroupOffsetsResponse] = withAdminClient(clusterId) { _.listConsumerGroupOffsets(groupId).map(res => ConsumerGroupOffsetsResponse(res.map { case (tp, offMeta) => 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 c82628b..f6ea311 100644 --- a/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala +++ b/core/src/main/scala/io/km8/core/kafka/KafkaProducer.scala @@ -28,9 +28,12 @@ object KafkaProducer { value: String )( clusterId: String - ): ZIO[Scope & KafkaProducer, Throwable, Unit] = ZIO.environmentWithZIO[KafkaProducer](_.get.produce(topic, key, value)(clusterId)) + ): ZIO[KafkaProducer, Throwable, Unit] = + ZIO.scoped { + ZIO.environmentWithZIO[KafkaProducer](_.get.produce(topic, key, value)(clusterId)) + } - lazy val liveLayer: URLayer[ClusterConfig & Scope, KafkaProducer] =ZLayer.fromZIO( for { + lazy val liveLayer: URLayer[ClusterConfig, KafkaProducer] = ZLayer.fromZIO(for { clusterConfigService <- ZIO.service[ClusterConfig] } yield new KafkaProducer { @@ -43,8 +46,10 @@ object KafkaProducer { .map(c => ProducerSettings(c.kafkaHosts)) def producerLayer(clusterId: String): ZIO[Scope, Throwable, Producer] = - settingsLayer(clusterId) - .flatMap(settings => Producer.make(settings)) + ZIO.scoped { + settingsLayer(clusterId) + .flatMap(settings => Producer.make(settings)) + } def produce( topic: String, @@ -57,6 +62,5 @@ object KafkaProducer { .produce[Any, String, String](topic, key, value, Serde.string, Serde.string) .unit .provideLayer(ZLayer.fromZIO(producerLayer(clusterId))) - } - ) -} \ No newline at end of file + }) +} 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 3936324..3ed33cf 100644 --- a/core/src/main/scala/io/km8/core/utils/Logger.scala +++ b/core/src/main/scala/io/km8/core/utils/Logger.scala @@ -8,12 +8,12 @@ import zio.Console object Logger { -/* + /* lazy val liveLayer: URLayer[Console with Clock, Logging] = 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 cb94812..3073da2 100644 --- a/core/src/test/scala/io/kafkamate/BasicSpec.scala +++ b/core/src/test/scala/io/kafkamate/BasicSpec.scala @@ -13,7 +13,7 @@ object BasicSpec extends ZIOSpecDefault { override def spec = suite("Simple test")( test("Check zio result") { - assertM(ZIO.succeed(1))(equalTo(1)) + 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 f3aa435..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,27 +1,22 @@ package io.kafkamate package kafka package consumer -/* +/* import zio.* -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 StringProducer with KafkaConsumer] = + : Layer[TestFailure[Throwable], StringProducer with KafkaConsumer] = (Clock.live >+> Console.live >+> KafkaEmbedded.Kafka.embedded >+> @@ -38,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/fx/src/main/scala/io/km8/fx/Main.scala b/fx/src/main/scala/io/km8/fx/Main.scala index ba78694..383130f 100644 --- a/fx/src/main/scala/io/km8/fx/Main.scala +++ b/fx/src/main/scala/io/km8/fx/Main.scala @@ -17,6 +17,7 @@ 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.views.{*, given} import java.util.concurrent.Executor import scala.concurrent.ExecutionContext @@ -67,19 +68,24 @@ object Main extends JFXApp3: override def start(): Unit = val io: ZIO[Any, Throwable, JFXApp3.PrimaryStage] = - for - h <- Hub.unbounded[UIEvent] - hubLayer = ZLayer.succeed(h) - main <- mkWindow.provideLayer(UI.make("") +!+ hubLayer) - s <- mkScene(main).provideLayer(hubLayer) - ret <- ZIO.attempt(new JFXApp3.PrimaryStage { - title = "KM8" - scene = s - }) - yield ret + ZIO.scoped { + val hubLayer = ZLayer.fromZIO(Hub.unbounded[UIEvent]) + val hubMessages = ZLayer.fromZIO(Hub.unbounded[Msg[ViewState]]) + for + _ <- initViews.provide(hubMessages) + main <- mkWindow.provideLayer(UI.make("") +!+ hubLayer) + s <- mkScene(main).provideLayer(hubLayer) + ret <- ZIO.attempt(new JFXApp3.PrimaryStage { + title = "KM8" + scene = s + }) + yield ret + } - Runtime.default.unsafeRun( - io - .onExecutionContext(currentThreadEC) - .exitCode - ) + Unsafe.unsafe { implicit u: Unsafe => + Runtime.default.unsafe.run( + io + .onExecutionContext(currentThreadEC) + .exitCode + ) + } 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 66921d9..820b343 100644 --- a/fx/src/main/scala/io/km8/fx/models/UI.scala +++ b/fx/src/main/scala/io/km8/fx/models/UI.scala @@ -61,7 +61,7 @@ def switch[P <: Page, S <: Page](p: Path[P, S]) = p match { opaque type Offset = Long -opaque type Config = (String, String) +opaque type Config = (String, String) opaque type TopicName = String @@ -194,8 +194,15 @@ case class MessageHeader(key: String, value: Array[Byte]) type UIEnv = UI & EventsHub def dispatchEvent: ZIO[EventsHub, Nothing, UIEvent => Unit] = - ZIO.service[EventsHub].map(hub => event => Runtime.global.unsafeRun(hub.publish(event))) + ZIO + .service[EventsHub] + .map(hub => + event => + Unsafe.unsafeCompat { implicit u => + Runtime.default.unsafe.run(hub.publish(event)) + } + ) object Test { val stuff: String = "" -} \ No newline at end of file +} 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 cc36f45..9df9d6e 100644 --- a/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala +++ b/fx/src/main/scala/io/km8/fx/ui/BaseControl.scala @@ -6,7 +6,7 @@ import scalafx.scene.control.Alert.AlertType import scalafx.scene.control.{Alert, ButtonType} import zio.* -trait BaseControl[R ]: +trait BaseControl[R]: def render: ZIO[R, Throwable, Node] def alert(text: Any) = 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 7257c57..0819c56 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 @@ -32,7 +32,7 @@ class HeaderControl extends BaseControl[UIEnv]: _ <- ZStream .fromHub(hub) .foreach { case UIEvent.FocusOmni => - ZIO.succeed { + ZIO.attempt { omniBar.requestFocus() } } 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 a404185..773cbb2 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 @@ -10,6 +10,6 @@ import models.* class MainContentControl extends BaseControl[UI]: - private lazy val view = ZIO.succeed(new ScrollPane()) + private lazy val view = ZIO.attempt(new ScrollPane()) override def render: ZIO[UI, Throwable, Node] = view 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 e28712f..3f33ea1 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 @@ -13,7 +13,7 @@ class NavigatorControl extends BaseControl[UI]: for ui <- ZIO.service[UI] tv <- treeView - ret <- ZIO.succeed( + ret <- ZIO.attempt( new ScrollPane { minWidth = ui.config.leftWidth fitToWidth = true @@ -38,7 +38,7 @@ class NavigatorControl extends BaseControl[UI]: private val treeView = for ui <- ZIO.service[UI] - tv <- ZIO.succeed { + tv <- ZIO.attempt { new TreeView[String] { showRoot = false id = "left-tree" diff --git a/fx/src/main/scala/io/km8/fx/views/Clusters.scala b/fx/src/main/scala/io/km8/fx/views/Clusters.scala index d310716..5c277d2 100644 --- a/fx/src/main/scala/io/km8/fx/views/Clusters.scala +++ b/fx/src/main/scala/io/km8/fx/views/Clusters.scala @@ -1,3 +1,56 @@ package io.km8.fx.views -case class Clusters(list: List[String]) \ No newline at end of file +import zio.* +import io.km8.common.ClusterDetails + +trait BaseMessage + +enum Msg[S] extends BaseMessage: + case Init(state: S) + +enum Screen: + case Clusters, Search + +trait View[S: Tag]: + val children: List[View[S]] = Nil + + def init: ZIO[Hub[Msg[S]], Nothing, Unit] = + registerCallback(update) *> ZIO.foreach(children)(_.init).unit + + def publishMessage(msg: Msg[S]): ZIO[Hub[Msg[S]], Nothing, Unit] = + ZIO.service[Hub[Msg[S]]].flatMap(_.publish(msg)).unit + + def update(message: Msg[S]): ZIO[Any, Nothing, Option[Msg[S]]] + + def registerCallback( + cb: Msg[S] => ZIO[Any, Nothing, Option[Msg[S]]] + ): ZIO[Hub[Msg[S]], Nothing, Unit] = + ZIO.scoped { + for { + hub <- ZIO.service[Hub[Msg[S]]] + q <- hub.subscribe + msg <- q.take + res <- cb(msg) + _ <- publishMessage(res.get).when(res.isDefined) + } yield () + } + +case class ViewState( + focused: Screen, + clusterDetails: List[ClusterDetails], + currentCluster: Option[ClusterDetails]) + +case class MainView() extends View[ViewState]: + override val children = ClustersView() :: SearchView(None) :: Nil + override def update(message: Msg[ViewState]) = ZIO.none + +case class ClustersView() extends View[ViewState]: + override def init: UIO[Unit] = ZIO.debug("calling method from clusters") + override def update(message: Msg[ViewState]) = ZIO.none + +case class SearchView(search: Option[String]) extends View[ViewState]: + override def init: UIO[Unit] = ZIO.debug("calling method from search") + override def update(message: Msg[ViewState]) = ZIO.none + +def initViews = + MainView().init From e075946523d72f06d57f90f50d4384cd31300063 Mon Sep 17 00:00:00 2001 From: Octav Zaharia Date: Thu, 21 Jul 2022 19:54:14 +0300 Subject: [PATCH 3/6] wip --- common/common.iml | 50 ++++++++++ core/src/it/it.iml | 11 +++ core/src/main/main.iml | 11 +++ core/src/test/test.iml | 11 +++ fx/fx.iml | 91 ++++++++++++++++++ fx/src/main/scala/io/km8/fx/Main.scala | 96 +++++++++++-------- fx/src/main/scala/io/km8/fx/models/UI.scala | 80 ++++++++++++---- .../km8/fx/ui/components/HeaderControl.scala | 80 +++++++++------- .../fx/ui/components/NavigatorControl.scala | 4 +- .../main/scala/io/km8/fx/views/Clusters.scala | 61 +++++------- 10 files changed, 364 insertions(+), 131 deletions(-) create mode 100644 common/common.iml create mode 100644 core/src/it/it.iml create mode 100644 core/src/main/main.iml create mode 100644 core/src/test/test.iml create mode 100644 fx/fx.iml 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/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/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 383130f..2c42981 100644 --- a/fx/src/main/scala/io/km8/fx/Main.scala +++ b/fx/src/main/scala/io/km8/fx/Main.scala @@ -1,8 +1,8 @@ package io.km8.fx +import com.sun.javafx.application.PlatformImpl import javafx.application.Platform -import scala.collection.mutable.Queue as SQueue import zio.* import javafx.beans.property.ObjectProperty import scalafx.application.JFXApp3 @@ -18,6 +18,8 @@ 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 zio.stream.ZStream +import scala.collection.mutable.Queue as SQueue import java.util.concurrent.Executor import scala.concurrent.ExecutionContext @@ -27,13 +29,15 @@ import scalafx.scene.input.KeyCode object Main extends JFXApp3: - val currentThreadEC = ExecutionContext.fromExecutor(new Executor { - override def execute(command: Runnable): Unit = command.run() - }) + val currentExe = zio.Executor.fromJavaExecutor((command: Runnable) => command.run()) + val currentThreadEC = ExecutionContext.fromExecutor((command: Runnable) => command.run()) + lazy val headerControl = HeaderControl() - private lazy val mkWindow = + private def mkWindow = for - header <- HeaderControl().render + q <- ZIO.service[EventsQ] + _ <- ZIO.debug(s"mkWindow - ${Thread.currentThread()}") + header <- headerControl.render navigator <- NavigatorControl().render mainContent <- MainContentControl().render pane <- ZIO.attempt(new SplitPane { @@ -53,39 +57,53 @@ object Main extends JFXApp3: }) yield p - private def mkScene(sceneRoot: Parent): ZIO[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 _ => () - } + private def mkScene(sceneRoot: Parent) = + for { + q <- ZIO.service[EventsQ] + res <- ZIO.attempt( + new Scene(1366, 768) { + stylesheets = List("css/app.css") + root = sceneRoot + onKeyReleased = k => + k.code match + case KeyCode.Slash => + k.consume() + q.enqueue(Backend.FocusOmni) + case _ => () + }) + } yield res + + def handler: Update = + case m => + ZIO.debug(s"UI-$m from ${Thread.currentThread()}").as(None) + + override def start(): Unit = { + val ui = UI.make() + val io = + for + hub <- Hub.unbounded[Msg] + msgLayer = ZLayer.succeed(hub) + qLayer = msgLayer >>> EventsQ.toBus + _ <- MainView().init.forkDaemon.provide(msgLayer) + _ <- registerCallback(this, handler).provide(msgLayer).forkDaemon + _ <- + ZIO.debug(s"Init - ${Thread.currentThread()}") *> hub + .publish(Backend.Init) + main <- mkWindow.provide(ZLayer.succeed(ui), qLayer, msgLayer).onExecutionContext(currentThreadEC) + s <- mkScene(main).provide(qLayer) + ret <- + ZIO.attempt { + println(s"KM8 - ${Thread.currentThread()} - ${Platform.isFxApplicationThread}") + new JFXApp3.PrimaryStage { + title = s"KM8 - ${Thread.currentThread()} - ${Platform.isFxApplicationThread}" + scene = s + } + } + yield ret - override def start(): Unit = - val io: ZIO[Any, Throwable, JFXApp3.PrimaryStage] = - ZIO.scoped { - val hubLayer = ZLayer.fromZIO(Hub.unbounded[UIEvent]) - val hubMessages = ZLayer.fromZIO(Hub.unbounded[Msg[ViewState]]) - for - _ <- initViews.provide(hubMessages) - main <- mkWindow.provideLayer(UI.make("") +!+ hubLayer) - s <- mkScene(main).provideLayer(hubLayer) - ret <- ZIO.attempt(new JFXApp3.PrimaryStage { - title = "KM8" - scene = s - }) - yield ret + stage = + Unsafe.unsafeCompat { implicit u: Unsafe => + Runtime.default.unsafe.run(io.onExecutor(currentExe)).getOrThrow() } - Unsafe.unsafe { implicit u: Unsafe => - Runtime.default.unsafe.run( - io - .onExecutionContext(currentThreadEC) - .exitCode - ) - } + } 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 820b343..4410d4c 100644 --- a/fx/src/main/scala/io/km8/fx/models/UI.scala +++ b/fx/src/main/scala/io/km8/fx/models/UI.scala @@ -1,10 +1,13 @@ package io.km8.fx package models +import javafx.application.Platform import zio.* import zio.prelude.NonEmptyList +import zio.stream.ZStream import scala.annotation.implicitNotFound +import scala.collection.mutable.Queue as SQueue type TestSeed = String | Int | Long @@ -167,16 +170,14 @@ case class UI( data: List[Cluster], config: UIConfig) -enum UIEvent: - case FocusOmni - -type EventsHub = Hub[UIEvent] - -val EventsHub = zio.Hub - object UI: + def make(seed: TestSeed = ""): UI = + UI( + data = List.range(1, 4).map(gen), + config = UIConfig(leftWidth = 300) + ) - def make(seed: TestSeed = ""): URLayer[Any, UI] = + def makeLayer(seed: TestSeed = ""): URLayer[Any, UI] = ZLayer.succeed( UI( data = List.range(1, 4).map(gen), @@ -191,18 +192,61 @@ case class Message( case class MessageHeader(key: String, value: Array[Byte]) -type UIEnv = UI & EventsHub +type UIEnv = UI -def dispatchEvent: ZIO[EventsHub, Nothing, UIEvent => Unit] = - ZIO - .service[EventsHub] - .map(hub => - event => - Unsafe.unsafeCompat { implicit u => - Runtime.default.unsafe.run(hub.publish(event)) - } - ) object Test { val stuff: String = "" } + +trait Msg + +enum Backend extends Msg: + case Init + case FocusOmni + case SearchClicked(search: String) + +enum Signal extends Msg: + case Search + case ChangedClusters(value: List[Cluster]) + +type MsgBus = Hub[Msg] +type EventsQ = SQueue[Msg] + +object EventsQ { + val schedule = Schedule.spaced(100.millis) + def toBus = + ZLayer.fromZIO( + for + hub <- ZIO.service[Hub[Msg]] + eventsQ <- ZIO.attempt(SQueue.empty[Msg]) + _ <- ZIO.attempt(eventsQ.dequeueAll(_ => true)) + .flatMap(hub.publishAll) + .repeat(schedule) + .forkDaemon + yield eventsQ + ) +} + +extension (c: => Unit) + def fx = ZIO.succeed(Platform.runLater(() => c)) + +type Update = PartialFunction[Msg , ZIO[Any, Nothing, Option[Msg]]] + +def publishMessage(msg: Msg): ZIO[MsgBus, Nothing, Unit] = + ZIO.service[MsgBus].flatMap(_.publish(msg)).unit + +def registerCallback(sender: Object, cb: Update): ZIO[MsgBus , Nothing, Unit] = + for { + hub <- ZIO.service[MsgBus] + _ <- ZIO.debug(s"${sender.toString} - ${Thread.currentThread()}") + _ <- ZStream.fromHub(hub).foreach { m => + for { + res <- cb(m) + _ <- res match { + case Some(v) => ZIO.debug(s"Sending $v") *> hub.publish(v) + case None => ZIO.unit + } + } yield () + } + } yield () 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 0819c56..ce2d24b 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[UIEnv & MsgBus & EventsQ]: lazy val omniBar = new TextField { @@ -20,41 +21,46 @@ class HeaderControl extends BaseControl[UIEnv]: prefWidth = 600 } - lazy val button = new Button { - id = "search-omni" - text = "Search" - onMouseClicked = _ => alert("test") - } + lazy val buttonZIO = + for { + q <- ZIO.service[EventsQ] + button <- ZIO.attempt(new Button { + id = "search-omni" + text = "Search" + onMouseClicked = + _ => q.enqueue(Backend.SearchClicked(omniBar.text.value)) + }) + } yield button + + val update: Update = + case Backend.FocusOmni => + ZIO.debug("Focus omni") *> + omniBar.requestFocus().fx.as(None) - private[ui] val view = - for - hub <- ZIO.service[EventsHub] - _ <- ZStream - .fromHub(hub) - .foreach { case UIEvent.FocusOmni => - ZIO.attempt { - 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 + private[ui] lazy val view = + for { + button <- buttonZIO + toolBar <- ZIO.attempt( + 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 - ) - } + }) + _ <- registerCallback(this, update).forkDaemon + } yield toolBar - override def render: RIO[UIEnv, Node] = view + override def render = view 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 3f33ea1..212d404 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 @@ -51,7 +51,9 @@ class NavigatorControl extends BaseControl[UI]: // _ = tv.getSelectionModel.selectedItemProperty().addListener((obs, oldVal, newVal) => alert(newVal.getValue)) _ = tv.getSelectionModel .selectedItemProperty() - .addListener((obs, oldVal, newVal) => alert(ui.data.map(_.consumerGroups))) + .addListener((obs, oldVal, newVal) => + alert(ui.data.map(_.consumerGroups) + )) yield tv override def render: ZIO[UI, Throwable, Node] = diff --git a/fx/src/main/scala/io/km8/fx/views/Clusters.scala b/fx/src/main/scala/io/km8/fx/views/Clusters.scala index 5c277d2..cb8eec7 100644 --- a/fx/src/main/scala/io/km8/fx/views/Clusters.scala +++ b/fx/src/main/scala/io/km8/fx/views/Clusters.scala @@ -1,56 +1,45 @@ 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 -trait BaseMessage - -enum Msg[S] extends BaseMessage: - case Init(state: S) - -enum Screen: - case Clusters, Search trait View[S: Tag]: val children: List[View[S]] = Nil - def init: ZIO[Hub[Msg[S]], Nothing, Unit] = - registerCallback(update) *> ZIO.foreach(children)(_.init).unit - - def publishMessage(msg: Msg[S]): ZIO[Hub[Msg[S]], Nothing, Unit] = - ZIO.service[Hub[Msg[S]]].flatMap(_.publish(msg)).unit + def init: ZIO[MsgBus, Nothing, Unit] = + for + f <- registerCallback(this, update).forkDaemon + _ <- ZIO.foreach(children)(_.init) + yield () - def update(message: Msg[S]): ZIO[Any, Nothing, Option[Msg[S]]] - - def registerCallback( - cb: Msg[S] => ZIO[Any, Nothing, Option[Msg[S]]] - ): ZIO[Hub[Msg[S]], Nothing, Unit] = - ZIO.scoped { - for { - hub <- ZIO.service[Hub[Msg[S]]] - q <- hub.subscribe - msg <- q.take - res <- cb(msg) - _ <- publishMessage(res.get).when(res.isDefined) - } yield () - } + def update: Update case class ViewState( - focused: Screen, - clusterDetails: List[ClusterDetails], - currentCluster: Option[ClusterDetails]) + clusterDetails: List[Cluster], + currentCluster: Option[Cluster]) case class MainView() extends View[ViewState]: override val children = ClustersView() :: SearchView(None) :: Nil - override def update(message: Msg[ViewState]) = ZIO.none + override def update = { case _ => ZIO.none} case class ClustersView() extends View[ViewState]: - override def init: UIO[Unit] = ZIO.debug("calling method from clusters") - override def update(message: Msg[ViewState]) = ZIO.none + override val children = TitleView() :: Nil + + override def update = { + case Backend.Init => + val c1 = gen[Cluster]() + val c2 = gen[Cluster]() + ZIO.debug("creating clusters").as(Some(Signal.ChangedClusters(List(c1, c2)))) + case Backend.SearchClicked(search) => ZIO.debug(s"Searched $search").as(None) + case m => ZIO.debug(s"ClusterView update: $m") *> ZIO.none + } case class SearchView(search: Option[String]) extends View[ViewState]: - override def init: UIO[Unit] = ZIO.debug("calling method from search") - override def update(message: Msg[ViewState]) = ZIO.none + override def update = { case m => ZIO.debug(s"SearchView update: $m").as(None)} -def initViews = - MainView().init +case class TitleView() extends View[ViewState]: + override def update = {case m => ZIO.debug(s"TitleView update: $m").as(None)} From 60d2ef022f5dc3ed1fa6d0e0872fdd1a04abb196 Mon Sep 17 00:00:00 2001 From: Octav Zaharia Date: Fri, 5 Aug 2022 10:57:36 +0300 Subject: [PATCH 4/6] wip --- fx/src/main/scala/io/km8/fx/Main.scala | 89 +++------------ fx/src/main/scala/io/km8/fx/models/UI.scala | 108 ++++++++++-------- .../main/scala/io/km8/fx/ui/BaseControl.scala | 7 +- .../fx/ui/components/ClusterListControl.scala | 9 +- .../km8/fx/ui/components/HeaderControl.scala | 28 +++-- .../fx/ui/components/MainContentControl.scala | 9 +- .../io/km8/fx/ui/components/MainStage.scala | 60 ++++++++++ .../fx/ui/components/NavigatorControl.scala | 80 +++++++------ .../main/scala/io/km8/fx/views/Clusters.scala | 45 -------- fx/src/main/scala/io/km8/fx/views/Views.scala | 65 +++++++++++ 10 files changed, 278 insertions(+), 222 deletions(-) create mode 100644 fx/src/main/scala/io/km8/fx/ui/components/MainStage.scala delete mode 100644 fx/src/main/scala/io/km8/fx/views/Clusters.scala create mode 100644 fx/src/main/scala/io/km8/fx/views/Views.scala diff --git a/fx/src/main/scala/io/km8/fx/Main.scala b/fx/src/main/scala/io/km8/fx/Main.scala index 2c42981..c433127 100644 --- a/fx/src/main/scala/io/km8/fx/Main.scala +++ b/fx/src/main/scala/io/km8/fx/Main.scala @@ -13,15 +13,11 @@ 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.models.{*, given} import io.km8.fx.ui.{*, given} import io.km8.fx.ui.components.{*, given} import io.km8.fx.views.{*, given} -import zio.stream.ZStream -import scala.collection.mutable.Queue as SQueue -import java.util.concurrent.Executor import scala.concurrent.ExecutionContext import scalafx.scene.input.KeyEvent import scalafx.Includes.* @@ -29,81 +25,28 @@ import scalafx.scene.input.KeyCode object Main extends JFXApp3: - val currentExe = zio.Executor.fromJavaExecutor((command: Runnable) => command.run()) - val currentThreadEC = ExecutionContext.fromExecutor((command: Runnable) => command.run()) - lazy val headerControl = HeaderControl() + val currentExe = + zio.Executor.fromJavaExecutor((command: Runnable) => command.run()) - private def mkWindow = - for - q <- ZIO.service[EventsQ] - _ <- 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 + override def start(): Unit = - private def mkScene(sceneRoot: Parent) = - for { - q <- ZIO.service[EventsQ] - res <- ZIO.attempt( - new Scene(1366, 768) { - stylesheets = List("css/app.css") - root = sceneRoot - onKeyReleased = k => - k.code match - case KeyCode.Slash => - k.consume() - q.enqueue(Backend.FocusOmni) - case _ => () - }) - } yield res - - def handler: Update = - case m => - ZIO.debug(s"UI-$m from ${Thread.currentThread()}").as(None) - - override def start(): Unit = { - val ui = UI.make() val io = for - hub <- Hub.unbounded[Msg] - msgLayer = ZLayer.succeed(hub) - qLayer = msgLayer >>> EventsQ.toBus - _ <- MainView().init.forkDaemon.provide(msgLayer) - _ <- registerCallback(this, handler).provide(msgLayer).forkDaemon - _ <- - ZIO.debug(s"Init - ${Thread.currentThread()}") *> hub - .publish(Backend.Init) - main <- mkWindow.provide(ZLayer.succeed(ui), qLayer, msgLayer).onExecutionContext(currentThreadEC) - s <- mkScene(main).provide(qLayer) - ret <- - ZIO.attempt { - println(s"KM8 - ${Thread.currentThread()} - ${Platform.isFxApplicationThread}") - new JFXApp3.PrimaryStage { - title = s"KM8 - ${Thread.currentThread()} - ${Platform.isFxApplicationThread}" - 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 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 4410d4c..1305341 100644 --- a/fx/src/main/scala/io/km8/fx/models/UI.scala +++ b/fx/src/main/scala/io/km8/fx/models/UI.scala @@ -1,6 +1,7 @@ package io.km8.fx package models +import io.km8.fx.views.ViewState import javafx.application.Platform import zio.* import zio.prelude.NonEmptyList @@ -153,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)), @@ -171,11 +171,12 @@ case class UI( config: UIConfig) object UI: + def make(seed: TestSeed = ""): UI = - UI( - data = List.range(1, 4).map(gen), - config = UIConfig(leftWidth = 300) - ) + UI( + data = List.range(1, 4).map(gen), + config = UIConfig(leftWidth = 300) + ) def makeLayer(seed: TestSeed = ""): URLayer[Any, UI] = ZLayer.succeed( @@ -192,9 +193,6 @@ case class Message( case class MessageHeader(key: String, value: Array[Byte]) -type UIEnv = UI - - object Test { val stuff: String = "" } @@ -202,51 +200,71 @@ object Test { trait Msg enum Backend extends Msg: - case Init + case LoadClusters + case LoadConfig case FocusOmni - case SearchClicked(search: String) + case Search(search: String) + case LoadTopics enum Signal extends Msg: + case Nop case Search - case ChangedClusters(value: List[Cluster]) + case ChangedClusters -type MsgBus = Hub[Msg] -type EventsQ = SQueue[Msg] +type MsgBus[S] = Hub[(S, Msg)] -object EventsQ { - val schedule = Schedule.spaced(100.millis) - def toBus = - ZLayer.fromZIO( - for - hub <- ZIO.service[Hub[Msg]] - eventsQ <- ZIO.attempt(SQueue.empty[Msg]) - _ <- ZIO.attempt(eventsQ.dequeueAll(_ => true)) - .flatMap(hub.publishAll) - .repeat(schedule) - .forkDaemon - yield eventsQ - ) -} +object MsgBus: + def layer[S: Tag] = ZLayer.fromZIO(Hub.unbounded[(S, Msg)]) + + def signal[S: Tag](m: Msg, s: S) = + ZIO.service[MsgBus[S]].flatMap { hub => + hub.publish(s -> m) + } + +type EventsQ[S] = SQueue[(S, Msg)] -extension (c: => Unit) - def fx = ZIO.succeed(Platform.runLater(() => c)) +def fireFX[S](m: Msg, s: S = ViewState.empty): EventsQ[S] ?=> Unit = + summon[EventsQ[S]].enqueue(s -> m) -type Update = PartialFunction[Msg , ZIO[Any, Nothing, Option[Msg]]] +import io.km8.fx.views.* -def publishMessage(msg: Msg): ZIO[MsgBus, Nothing, Unit] = - ZIO.service[MsgBus].flatMap(_.publish(msg)).unit +object App: + val schedule = Schedule.spaced(100.millis) -def registerCallback(sender: Object, cb: Update): ZIO[MsgBus , Nothing, Unit] = - for { - hub <- ZIO.service[MsgBus] - _ <- ZIO.debug(s"${sender.toString} - ${Thread.currentThread()}") + def initialize[S: Tag, V <: View[S]](v: View[S]): UIO[(ULayer[Hub[(S, Msg)]], ULayer[SQueue[(S, Msg)]])] = + for + hub <- Hub.unbounded[(S, Msg)] + hubLayer = ZLayer.succeed(hub) + _ <- v.init.forkDaemon.provide(hubLayer) + eventsQ <- ZIO.succeed(SQueue.empty[(S, Msg)]) + _ <- 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] = ((S, Msg)) => UIO[Option[(S, Msg)]] + +def publishMessage[S: Tag](msg: (S, Msg)): 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 Some(v) => ZIO.debug(s"Sending $v") *> hub.publish(v) - case None => ZIO.unit - } - } yield () - } - } yield () + for + res <- cb(m) + _ <- res match + case Some(v) => ZIO.debug(s"Sending $v") *> hub.publish(v) + case None => ZIO.unit + + 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 9df9d6e..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]: - 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 d791a41..5eb9b34 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[UI] { +class ClusterListControl extends BaseControl[ViewState, UI & MsgBus[ViewState]] { val mkList = for { ui <- ZIO.service[UI] @@ -18,5 +19,9 @@ class ClusterListControl extends BaseControl[UI] { private[components] val view = mkList - override def render: RIO[UI, Node] = view + override def render = + registerCallbackAsync(this, update ) *> view + + val update: Update[ViewState] = + case s => ZIO.debug(s"${this.getClass} - $s").as(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 ce2d24b..e033ff5 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 @@ -13,7 +13,7 @@ import scalafx.beans.property.ReadOnlyDoubleProperty import io.km8.fx.views.* import javafx.application.Platform -class HeaderControl extends BaseControl[UIEnv & MsgBus & EventsQ]: +class HeaderControl extends BaseControl[ViewState, MsgBus[ViewState] & EventsQ[ViewState]]: lazy val omniBar = new TextField { @@ -21,26 +21,26 @@ class HeaderControl extends BaseControl[UIEnv & MsgBus & EventsQ]: prefWidth = 600 } - lazy val buttonZIO = + val buttonZIO = for { - q <- ZIO.service[EventsQ] + given EventsQ[ViewState] <- ZIO.service[EventsQ[ViewState]] button <- ZIO.attempt(new Button { id = "search-omni" text = "Search" - onMouseClicked = - _ => q.enqueue(Backend.SearchClicked(omniBar.text.value)) + onMouseClicked = _ => + fireFX(Backend.Search(omniBar.text.value)) }) } yield button - val update: Update = - case Backend.FocusOmni => - ZIO.debug("Focus omni") *> - omniBar.requestFocus().fx.as(None) + val update: Update[ViewState] = { + case _ -> Backend.FocusOmni => omniBar.requestFocus().fx.as(None) + case _ => ZIO.none + } - private[ui] lazy val view = + override def render = for { button <- buttonZIO - toolBar <- ZIO.attempt( + toolBar = new ToolBar { prefHeight = 76 maxHeight = 76 @@ -59,8 +59,6 @@ class HeaderControl extends BaseControl[UIEnv & MsgBus & EventsQ]: omniBar, button ) - }) - _ <- registerCallback(this, update).forkDaemon + } + _ <- registerCallbackAsync(this, update) } yield toolBar - - override def render = view 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 773cbb2..1a0a337 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[UI]: +class MainContentControl extends BaseControl[ViewState, MsgBus[ViewState]]: private lazy val view = ZIO.attempt(new ScrollPane()) - override def render: ZIO[UI, Throwable, Node] = view + override def render = + registerCallbackAsync(this, update) *> view + + val update: Update[ViewState] = + case (_, m) => ZIO.debug(s"${this.getClass} - $m").as(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..80738b7 --- /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()}").as(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 212d404..cb52543 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[UI]: - - private lazy val view = - for - ui <- ZIO.service[UI] - tv <- treeView - ret <- ZIO.attempt( - 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,27 +28,38 @@ class NavigatorControl extends BaseControl[UI]: ) } - private val treeView = - for - ui <- ZIO.service[UI] - tv <- ZIO.attempt { - 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 (state, 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())).as(None) + case _ => ZIO.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 - - override def render: ZIO[UI, Throwable, Node] = - for ret <- view - yield ret + .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 = + registerCallbackAsync(this, update).as(scrollPane) diff --git a/fx/src/main/scala/io/km8/fx/views/Clusters.scala b/fx/src/main/scala/io/km8/fx/views/Clusters.scala deleted file mode 100644 index cb8eec7..0000000 --- a/fx/src/main/scala/io/km8/fx/views/Clusters.scala +++ /dev/null @@ -1,45 +0,0 @@ -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 - - -trait View[S: Tag]: - val children: List[View[S]] = Nil - - def init: ZIO[MsgBus, Nothing, Unit] = - for - f <- registerCallback(this, update).forkDaemon - _ <- ZIO.foreach(children)(_.init) - yield () - - def update: Update - -case class ViewState( - clusterDetails: List[Cluster], - currentCluster: Option[Cluster]) - -case class MainView() extends View[ViewState]: - override val children = ClustersView() :: SearchView(None) :: Nil - override def update = { case _ => ZIO.none} - -case class ClustersView() extends View[ViewState]: - override val children = TitleView() :: Nil - - override def update = { - case Backend.Init => - val c1 = gen[Cluster]() - val c2 = gen[Cluster]() - ZIO.debug("creating clusters").as(Some(Signal.ChangedClusters(List(c1, c2)))) - case Backend.SearchClicked(search) => ZIO.debug(s"Searched $search").as(None) - case m => ZIO.debug(s"ClusterView update: $m") *> ZIO.none - } - -case class SearchView(search: Option[String]) extends View[ViewState]: - override def update = { case m => ZIO.debug(s"SearchView update: $m").as(None)} - -case class TitleView() extends View[ViewState]: - override def update = {case m => ZIO.debug(s"TitleView update: $m").as(None)} 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..1157224 --- /dev/null +++ b/fx/src/main/scala/io/km8/fx/views/Views.scala @@ -0,0 +1,65 @@ +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 _ => ZIO.none + +object ClustersView extends View[ViewState]: + override val children = TitleView :: Nil + + override def update = + case (state, Backend.LoadClusters) => + Data.loadClusters.flatMap { newClusters => + ZIO + .debug("creating clusters") + .as(Some(state.copy(clusterDetails = newClusters) -> Signal.ChangedClusters)) + } + case _ -> Backend.Search(search) => + ZIO.debug(s"Searched $search").as(None) + case (_, m) => + ZIO.debug(s"ClusterView update: $m") *> ZIO.none + +object SearchView extends View[ViewState]: + + override def update = + case (_, m) => ZIO.debug(s"SearchView update: $m").as(None) + +object TitleView extends View[ViewState]: + + override def update = + case _ -> m => ZIO.debug(s"TitleView update: $m").as(None) From e7f7b7e289a71f64a955378a3abf07894e86304e Mon Sep 17 00:00:00 2001 From: Octav Zaharia Date: Tue, 16 Aug 2022 07:47:47 +0300 Subject: [PATCH 5/6] wip --- fx/src/main/scala/io/km8/fx/models/UI.scala | 19 ++++++++++++--- .../fx/ui/components/ClusterListControl.scala | 2 +- .../km8/fx/ui/components/HeaderControl.scala | 4 ++-- .../fx/ui/components/MainContentControl.scala | 2 +- .../io/km8/fx/ui/components/MainStage.scala | 2 +- .../fx/ui/components/NavigatorControl.scala | 4 ++-- fx/src/main/scala/io/km8/fx/views/Views.scala | 23 +++++++++++++------ 7 files changed, 39 insertions(+), 17 deletions(-) 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 1305341..370bfd1 100644 --- a/fx/src/main/scala/io/km8/fx/models/UI.scala +++ b/fx/src/main/scala/io/km8/fx/models/UI.scala @@ -205,6 +205,7 @@ enum Backend extends Msg: case FocusOmni case Search(search: String) case LoadTopics + case KeyPressed(key: Char) enum Signal extends Msg: case Nop @@ -246,7 +247,19 @@ object App: extension (c: => Unit) def fx = ZIO.succeed(Platform.runLater(() => c)) -type Update[S] = ((S, Msg)) => UIO[Option[(S, Msg)]] +type Update[S] = ((S, Msg)) => UIO[(Option[S], Option[Msg])] + +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[(Option[S], Option[Msg])] = + ZIO.succeed(Some(state), Some(msg)) + + def none[S]: UIO[(Option[S], Option[Msg])] = ZIO.succeed(None -> None) def publishMessage[S: Tag](msg: (S, Msg)): URIO[MsgBus[S], Unit] = ZIO.service[MsgBus[S]].flatMap(_.publish(msg)).unit @@ -262,8 +275,8 @@ def registerCallback[S: Tag](sender: Object, cb: Update[S]): URIO[MsgBus[S], Uni for res <- cb(m) _ <- res match - case Some(v) => ZIO.debug(s"Sending $v") *> hub.publish(v) - case None => ZIO.unit + case s -> m if s.isDefined || m.isDefined => ZIO.debug(s"Sending $res") *> hub.publish(res) + case _ => Update.none yield () } 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 5eb9b34..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 @@ -23,5 +23,5 @@ class ClusterListControl extends BaseControl[ViewState, UI & MsgBus[ViewState]] registerCallbackAsync(this, update ) *> view val update: Update[ViewState] = - case s => ZIO.debug(s"${this.getClass} - $s").as(None) + 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 e033ff5..94f26ee 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 @@ -33,8 +33,8 @@ class HeaderControl extends BaseControl[ViewState, MsgBus[ViewState] & EventsQ[V } yield button val update: Update[ViewState] = { - case _ -> Backend.FocusOmni => omniBar.requestFocus().fx.as(None) - case _ => ZIO.none + case _ -> Backend.FocusOmni => omniBar.requestFocus().fx *> Update.none + case _ => Update.none } override def render = 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 1a0a337..fb3dac0 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 @@ -17,4 +17,4 @@ class MainContentControl extends BaseControl[ViewState, MsgBus[ViewState]]: registerCallbackAsync(this, update) *> view val update: Update[ViewState] = - case (_, m) => ZIO.debug(s"${this.getClass} - $m").as(None) + case (_, 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 index 80738b7..54c4f76 100644 --- a/fx/src/main/scala/io/km8/fx/ui/components/MainStage.scala +++ b/fx/src/main/scala/io/km8/fx/ui/components/MainStage.scala @@ -38,7 +38,7 @@ class MainStage: yield p def handler: Update[ViewState] = - case m => ZIO.debug(s"UI-$m from ${Thread.currentThread()}").as(None) + case m => ZIO.debug(s"UI-$m from ${Thread.currentThread()}") *> Update.none def render = 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 cb52543..1d2656a 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 @@ -33,8 +33,8 @@ class NavigatorControl extends BaseControl[ViewState, MsgBus[ViewState]]: ZIO.attempt { val nodes = state.clusterDetails.map(s => clusterNode(s).delegate) treeView.root.value.getChildren.addAll(nodes: _*) - }.catchAll(e => ZIO.succeed(e.printStackTrace())).as(None) - case _ => ZIO.none + }.catchAll(e => ZIO.succeed(e.printStackTrace())) *> Update.none + case _ => Update.none private lazy val treeView = new TreeView[String] { diff --git a/fx/src/main/scala/io/km8/fx/views/Views.scala b/fx/src/main/scala/io/km8/fx/views/Views.scala index 1157224..eb7b9e5 100644 --- a/fx/src/main/scala/io/km8/fx/views/Views.scala +++ b/fx/src/main/scala/io/km8/fx/views/Views.scala @@ -37,7 +37,7 @@ object MainView extends View[ViewState]: override val children = ClustersView :: SearchView :: Nil override def update = - case _ => ZIO.none + case _ => Update.none object ClustersView extends View[ViewState]: override val children = TitleView :: Nil @@ -46,20 +46,29 @@ object ClustersView extends View[ViewState]: case (state, Backend.LoadClusters) => Data.loadClusters.flatMap { newClusters => ZIO - .debug("creating clusters") - .as(Some(state.copy(clusterDetails = newClusters) -> Signal.ChangedClusters)) + .debug("creating clusters") *> + Update(state.copy(clusterDetails = newClusters), Signal.ChangedClusters) } case _ -> Backend.Search(search) => - ZIO.debug(s"Searched $search").as(None) + ZIO.debug(s"Searched $search") *> Update.none case (_, m) => - ZIO.debug(s"ClusterView update: $m") *> ZIO.none + 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 (_, m) => ZIO.debug(s"SearchView update: $m").as(None) + case (_, m) => ZIO.debug(s"SearchView update: $m") *> Update.none object TitleView extends View[ViewState]: override def update = - case _ -> m => ZIO.debug(s"TitleView update: $m").as(None) + case _ -> m => ZIO.debug(s"TitleView update: $m") *> Update.none From f032640f21e42894645161a632305d257e4fb427 Mon Sep 17 00:00:00 2001 From: Octav Date: Wed, 17 Aug 2022 07:01:38 +0300 Subject: [PATCH 6/6] wip --- fx/src/main/scala/io/km8/fx/models/UI.scala | 40 ++++++++++++------- .../km8/fx/ui/components/HeaderControl.scala | 3 +- .../fx/ui/components/MainContentControl.scala | 2 +- .../fx/ui/components/NavigatorControl.scala | 2 +- fx/src/main/scala/io/km8/fx/views/Views.scala | 10 ++--- 5 files changed, 34 insertions(+), 23 deletions(-) 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 370bfd1..2340ebc 100644 --- a/fx/src/main/scala/io/km8/fx/models/UI.scala +++ b/fx/src/main/scala/io/km8/fx/models/UI.scala @@ -212,32 +212,41 @@ enum Signal extends Msg: case Search case ChangedClusters -type MsgBus[S] = Hub[(S, Msg)] +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[(S, Msg)]) + 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(s -> m) + hub.publish(EventData(Some(s), Some(m))) } -type EventsQ[S] = SQueue[(S, Msg)] +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: S = ViewState.empty): EventsQ[S] ?=> Unit = - summon[EventsQ[S]].enqueue(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[Hub[(S, Msg)]], ULayer[SQueue[(S, Msg)]])] = + def initialize[S: Tag, V <: View[S]](v: View[S]): UIO[(ULayer[MsgBus[S]], ULayer[EventsQ[S]])] = for - hub <- Hub.unbounded[(S, Msg)] + hub <- Hub.unbounded[EventData[S]] hubLayer = ZLayer.succeed(hub) _ <- v.init.forkDaemon.provide(hubLayer) - eventsQ <- ZIO.succeed(SQueue.empty[(S, Msg)]) + eventsQ <- ZIO.succeed(SQueue.empty[EventData[S]]) _ <- ZIO .succeed(eventsQ.dequeueAll(_ => true)) .flatMap(hub.publishAll) @@ -247,7 +256,7 @@ object App: extension (c: => Unit) def fx = ZIO.succeed(Platform.runLater(() => c)) -type Update[S] = ((S, Msg)) => UIO[(Option[S], Option[Msg])] +type Update[S] = EventData[S] => UIO[EventData[S]] object Update: def state[S](state: S): UIO[(Option[S], Option[Msg])] = @@ -256,12 +265,12 @@ object Update: 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[(Option[S], Option[Msg])] = - ZIO.succeed(Some(state), Some(msg)) + def apply[S](state: S, msg: Msg): UIO[EventData[S]] = + ZIO.succeed(EventData(Some(state), Some(msg))) - def none[S]: UIO[(Option[S], Option[Msg])] = ZIO.succeed(None -> None) + def none[S]: UIO[EventData[S]] = ZIO.succeed(EventData(None , None)) -def publishMessage[S: Tag](msg: (S, Msg)): URIO[MsgBus[S], Unit] = +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]) = @@ -275,7 +284,8 @@ def registerCallback[S: Tag](sender: Object, cb: Update[S]): URIO[MsgBus[S], Uni for res <- cb(m) _ <- res match - case s -> m if s.isDefined || m.isDefined => ZIO.debug(s"Sending $res") *> hub.publish(res) + case EventData(s, m) if s.isDefined || m.isDefined => + ZIO.debug(s"Sending $res") *> hub.publish(res) case _ => Update.none yield () 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 94f26ee..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 @@ -33,7 +33,8 @@ class HeaderControl extends BaseControl[ViewState, MsgBus[ViewState] & EventsQ[V } yield button val update: Update[ViewState] = { - case _ -> Backend.FocusOmni => omniBar.requestFocus().fx *> Update.none + case EventData(_ , Some(Backend.FocusOmni)) => + omniBar.requestFocus().fx *> Update.none case _ => Update.none } 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 fb3dac0..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 @@ -17,4 +17,4 @@ class MainContentControl extends BaseControl[ViewState, MsgBus[ViewState]]: registerCallbackAsync(this, update) *> view val update: Update[ViewState] = - case (_, m) => ZIO.debug(s"${this.getClass} - $m") *> Update.none + case EventData(_, m) => ZIO.debug(s"${this.getClass} - $m") *> Update.none 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 1d2656a..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 @@ -29,7 +29,7 @@ class NavigatorControl extends BaseControl[ViewState, MsgBus[ViewState]]: } val update: Update[ViewState] = - case (state, Signal.ChangedClusters) => + case EventData(Some(state), Some(Signal.ChangedClusters)) => ZIO.attempt { val nodes = state.clusterDetails.map(s => clusterNode(s).delegate) treeView.root.value.getChildren.addAll(nodes: _*) diff --git a/fx/src/main/scala/io/km8/fx/views/Views.scala b/fx/src/main/scala/io/km8/fx/views/Views.scala index eb7b9e5..a56e5e9 100644 --- a/fx/src/main/scala/io/km8/fx/views/Views.scala +++ b/fx/src/main/scala/io/km8/fx/views/Views.scala @@ -43,15 +43,15 @@ object ClustersView extends View[ViewState]: override val children = TitleView :: Nil override def update = - case (state, Backend.LoadClusters) => + case EventData(Some(state), Some(Backend.LoadClusters)) => Data.loadClusters.flatMap { newClusters => ZIO .debug("creating clusters") *> Update(state.copy(clusterDetails = newClusters), Signal.ChangedClusters) } - case _ -> Backend.Search(search) => + case EventData(_, Some(Backend.Search(search))) => ZIO.debug(s"Searched $search") *> Update.none - case (_, m) => + case EventData(_, m) => ZIO.debug(s"ClusterView update: $m") *> Update.none /* @@ -66,9 +66,9 @@ object TextInput extends View[String]: object SearchView extends View[ViewState]: override def update = - case (_, m) => ZIO.debug(s"SearchView update: $m") *> Update.none + case EventData(_, m) => ZIO.debug(s"SearchView update: $m") *> Update.none object TitleView extends View[ViewState]: override def update = - case _ -> m => ZIO.debug(s"TitleView update: $m") *> Update.none + case EventData(_ , m) => ZIO.debug(s"TitleView update: $m") *> Update.none