test.scala

object Test {
  import zio._
  import zio.duration._
  import zio.kafka.consumer._


  val settings: ConsumerSettings =
    ConsumerSettings(List("localhost:9092"))
      .withGroupId("group")
      .withClientId("client")
      .withCloseTimeout(30.seconds)


import zio._
  import zio.console._
  import zio.kafka.consumer._
  import zio.kafka.serde._

  val subscription: Subscription = Subscription.topics("topic")
  val readKafka: RIO[Console with Blocking with Clock, Unit] =
    Consumer.consumeWith(settings, subscription, Serde.string, Serde.string) {
      case (key, value) =>
        putStrLn(s"Received message ${key}: ${value}")
      // Perform an effect with the received message
    }


 private val readKafka2: RIO[ZEnv with Consumer, Unit] =
    zio.kafka.consumer.Consumer.withConsumerService(
      _.consumeWith(subscription, Serde.string, Serde.string)({
        case (key, value) =>
          putStrLn(s"Received message ${key}: ${value}")
      })
    )

private val service: ZManaged[Clock with Blocking, Throwable, Consumer.Service] =
    Consumer.make(settings)

  private val layer: ZLayer[Clock with Blocking, Throwable, Consumer] = ZLayer(
    service
  )

  val readKafkaPrg: URIO[ZEnv, ExitCode] =
    readKafka2.provideCustomLayer(layer).exitCode

}