ZIO with Alpakka Kafka

zio-with-alpakka-kafka.scala

import akka.actor.ActorSystem
import akka.kafka.ConsumerMessage.CommittableMessage
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ ConsumerSettings, Subscriptions }
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{ Keep, Sink => AkkaSink }
import org.apache.kafka.common.serialization.StringDeserializer
import scalaz.zio._
import scalaz.zio.stream.Sink
import scalaz.zio.interop.reactiveStreams._

object AkkaTest extends App {

  override def run(
    args: List[String]
  ): ZIO[AkkaTest.Environment, Nothing, Int] = {
    val app = for {
      system <- Managed.make(Task(ActorSystem()))(system => Task.fromFuture(_ => system.terminate()).void.orDie)
      mat    <- Managed.fromEffect(Task(ActorMaterializer()(system)))
      consumerSettings <- Managed.fromEffect(
                           Task(
                             ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
                               .withBootstrapServers("localhost:9092")
                               .withGroupId("test")
                           )
                         )
      publisher <- Task {
                    Consumer
                      .committableSource(consumerSettings, Subscriptions.topics("test-topic"))
                      .toMat(AkkaSink.asPublisher(false))(Keep.both)
                      .run()(mat)
                  }.toManaged {
                    case (control, _) => Task.fromFuture(_ => control.shutdown).void.orDie
                  }.map(_._2)
      zioStream = publisher.toStream(15)
      messages  <- Managed.fromEffect(zioStream.take(50).run(Sink.collect[CommittableMessage[String, String]]))
      _         <- Managed.fromEffect(console.putStrLn(messages.mkString("\n")))
    } yield 0

    app.use_(ZIO.succeed(0)).orDie
  }
}