jczuchnowski / zio-pulsar Goto Github PK

Apache Pulsar client for Scala with ZIO and ZIO Streams integration.

License: BSD 2-Clause "Simplified" License

Scala 96.69% Shell 3.31%

zio-pulsar's Introduction

ZIO Pulsar

CI	Release	Snapshot

Purely functional Scala wrapper over the official Pulsar client.

Type-safe (utilizes Scala type system to reduce runtime exceptions present in the official Java client)
Streaming-enabled (naturally integrates with ZIO Streams)
ZIO integrated (uses common ZIO primitives like ZIO effect and ZManaged to reduce the boilerplate and increase expressiveness)

Compatibility

ZIO Pulsar is a Scala 3 library, so it's compatible with Scala 3 applications as well as Scala 2.13.6+ (see forward compatibility for more information.

Getting started

Add the following dependency to your build.sbt file:

Scala 3

libraryDependencies += "com.github.jczuchnowski" %% "zio-pulsar" % zioPulsarVersion

Scala 2.13.6+ (sbt 1.5.x)

libraryDependencies += 
  ("com.github.jczuchnowski" %% "zio-pulsar" % zioPulsarVersion).cross(CrossVersion.for2_13Use3)

ZIO Pulsar also needs ZIO and ZIO Streams to be provided:

libraryDependencies ++= Seq(
  "dev.zio" %% "zio"         % zioVersion,
  "dev.zio" %% "zio-streams" % zioVersion
)

Simple example of consumer and producer:

import org.apache.pulsar.client.api.{ PulsarClientException, Schema }
import zio._
import zio.pulsar._

object Main extends App:

  val pulsarClient = PulsarClient.live("localhost", 6650)

  val topic = "my-topic"

  val app: ZManaged[PulsarClient, PulsarClientException, Unit] =
    for
      builder  <- ConsumerBuilder.make(Schema.STRING).toManaged_
      consumer <- builder
                    .topic(topic)
                    .subscription(
                      Subscription(
                        "my-subscription", 
                        SubscriptionType.Shared))
                    .build
      producer <- Producer.make(topic, Schema.STRING)
      _        <- producer.send("Hello!").toManaged_
      m        <- consumer.receive.toManaged_
    yield ()
    
  def run(args: List[String]): URIO[ZEnv, ExitCode] =
    app.provideCustomLayer(pulsarClient).useNow.exitCode

Running examples locally

To try the examples from the examples subproject you'll need a Pulsar instance running locally. You can set one up using docker:

docker run -it \
  -p 6650:6650 \
  -p 8080:8080 \
  --mount source=pulsardata,target=/pulsar/data \
  --mount source=pulsarconf,target=/pulsar/conf \
  --network pulsar \
  apachepulsar/pulsar:2.7.0 \
  bin/pulsar standalone

zio-pulsar's People

Contributors

Stargazers

Watchers

Forkers

sanju041298 scala-steward isgasho yakob-vasyl jxnu-liguobin

zio-pulsar's Issues

Research type-safe implementation of `autoUpdatePartitions` and `autoUpdatePartitionsInterval`

Research the support for `property` setup in ConsumerBuilder

Set up Testcontainers for intergation tests

Provide a typesafe way for adding configurations

When creating a PulsarClient, Producer, Consumer and Readers it will be good to have a way to provide all the necessary configurations in a typesafe way.
For example at the moment when creating a PulsarClient, we can provide the serviceURL:

JPulsarClient.builder().serviceUrl(s"pulsar://$host:$port")

but we need to support adding more configurations, like security etc.
We need to think whats the best way to implement this

Research how `priorityLevel` works with different subscription types...

...and see if there's an opportunity to make it type-safe

Add Scala-style Schema support

We need Schema support that's using common Scala mechanics.

We could start with adding JSON Schema. Java version works on POJOs so it's basically useless in Scala. Instead we should support case classes.

The first approach could be along these lines:

Modify the ConsumerBuilder. Instead

def ConsumerBuilder.make[M](schema: Schema[M])

make use of using clause

def ConsumerBuilder.make[M](using schema: Schema[M])

Same for the Producer
Implement given instance of a new JSONSchema using ZIO JSON along these lines:

given jsonSchema[T](using encoder: JsonEncoder[T], decoder: JsonDecoder[T]): Schema[T] with
  def clone(): Schema[T] = ???
  def encode(t: T): Array[Byte] = ???
  def decode(bytes: Array[Byte]): T = ???
  def getSchemaInfo: SchemaInfo = ???

SchemaInfo should probably be implemented similarly to the one the Java Client uses.

The question remains if this is going to be compatible with the POJO serialization and should it be compatible. So basically are we going to be able to serialize/deserialize between equivalent:
Scala/case class <-> Java/POJO

Add support for crypto-* methods for ConsumerBuilder

Update the deprecated ZIO 1.0 components to 2.0

Initial the library was using zio 1.0. We need to update the components in place with the new ones.
For example in the Pulsar client:

ZIO.effect() to Zio.attempt()

and

ZManaged.make() to ZManaged.acquireReleaseWith()

Add tests for Java <> Scala Schema compatibility

Depends on #74

It should be possible to send JSON messages between ZIO Pulsar (Scala case classes) and Java Client (POJOs).

Upgrade to ZIO 2.0

ZIO 2.0 is at Milestone 4, with an RC expected in the next few weeks.
https://github.com/zio/zio/releases/tag/v2.0.0-M4

The API is nearly stable at this point, so any early migration work against this version should pay off towards the official 2.0 release.

The progress is being tracked here:
zio/zio#5470

The Stream Encoding work in progress is the only area where the API might still change before the RC.

We are actively working on a ScalaFix rule that will cover the bulk of the simple API changes:
https://github.com/zio/zio/blob/series/2.x/scalafix/rules/src/main/scala/fix/Zio2Upgrade.scala
We highly recommend starting with that, and then working through any remaining compilation errors :)

To assist with the rest of the migration, we have created this guide:
https://zio.dev/howto/migrate/zio-2.x-migration-guide/

If you would like assistance with the migration from myself or other ZIO contributors, please let us know!