Scala 3
libraryDependencies += "io.github.jxnu-liguobin" %% "zio-pulsar" % "latest version"
Scala 2.13.6+ (sbt 1.5.x)
libraryDependencies +=
("io.github.jxnu-liguobin" %% "zio-pulsar" % "latest version").cross(CrossVersion.for2_13Use3)
These dependencies are required in the project classpath (ZIO projects only need to pay attention to whether they have imported zio-streams):
libraryDependencies ++= Seq(
"dev.zio" %% "zio" % zioVersion,
"dev.zio" %% "zio-streams" % zioVersion,
"dev.zio" %% "zio-json" % zioJsonVersion
)
object SingleMessageExample extends ZIOAppDefault:
// Note: Do not duplicate the construction of this object, just keep one instance!
lazy val pulsarClient = PulsarClient.live("localhost", 6650)
// val pulsarClient = PulsarClient.live(""pulsar://localhost:6650,localhost:6651,localhost:6652"")
val topic = "single-topic"
// The client should be a long-term/on-demand object, while the consumer/producer is perishable after use
val app: ZIO[PulsarClient & Scope, PulsarClientException, Unit] =
for
builder <- ConsumerBuilder.make(JSchema.STRING)
consumer <- builder
.topic(topic)
.subscription(Subscription("my-subscription", SubscriptionType.Shared))
.build
producer <- Producer.make(topic, JSchema.STRING)
_ <- producer.send("Hello!")
m <- consumer.receive
_ = println(m.getValue)
yield ()
override def run = app.provideLayer(pulsarClient ++ Scope.default).exitCodeIf you need producer and consumer, or it is not convenient to propagate R, then we can use constructor injection:
object UserService {
val live = ZLayer.fromFunction(UserService.apply)
}
final case class UserService(zioPulsar: ZioPulsar) {
def sendPulsar(): ZIO[Scope, PulsarClientException, Message[String]] =
for {
consumerBuilder <- zioPulsar
.consumerBuilder(JSchema.STRING)
consumer <- consumerBuilder
.topic("zio-topic")
.subscription(
Subscription(
name = "zio-subscription",
`type` = SubscriptionType.Shared
)
)
.build // or use `unsafeBuild` to not close the consumer
msg <- consumer.receive(10, TimeUnit.SECONDS)
} yield msg
}
object ZioPulsarExample extends ZIOAppDefault:
val app: ZIO[UserService with Scope, IOException, Unit] =
ZIO.serviceWithZIO[UserService](_.sendPulsar()).unit
override def run = app
.provide(
UserService.live,
ZioPulsar.live,
Scope.default,
ZLayer.succeed(PulsarClientConfig("pulsar://localhost:6650"))
)
.exitCode