Skip to content

Instantly share code, notes, and snippets.

@iravid
Created March 18, 2019 21:10
Show Gist options
  • Save iravid/62ffb9ce687c91498101fd0ca3bc1860 to your computer and use it in GitHub Desktop.
Save iravid/62ffb9ce687c91498101fd0ca3bc1860 to your computer and use it in GitHub Desktop.
ZIO with Alpakka Kafka
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
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment