Producer and Consumer services as ZLayers #128

kciesielski · 2020-02-26T12:15:31Z

This PR exposes producer and consumer interfaces as ZLayers. Top-level types are:

  type Producer[R, K, V] = Has[Producer.Service[R, K, V]]
  type Consumer[R, K, V] = Has[Consumer.Service[R, K, V]]

where:

R is the environment required for serialization/deserialization
K is the type of keys
V is the type of values

Public API for creating service layers:

val pLayer: NoDeps[Throwable, Producer[R, K, V]] = Producer.make(settings, keySerde, valSerde)
val cLayer: NoDeps[Throwable, Consumer[R, K, V]] = Consumer.make(settings, keySerde, valSerde)

and from this point layers can be used as dependencies of programs.
It may be worth adding some additional aliases/methods for common use cases:

cases without the R parameter for cases when serialization and deserialization don't have dependencies
cases where keys and values are String or Array[Byte]

src/main/scala/zio/kafka/client/Producer.scala

src/main/scala/zio/kafka/client/package.scala

adamgfraser · 2020-02-26T13:25:05Z

src/main/scala/zio/kafka/client/package.scala

-
-import zio.ZIO
+import zio.{ Has, ZIO }
+import zio._


Don't need the wildcard import if you are importing the specific types you are using.

src/main/scala/zio/kafka/client/Producer.scala

adamgfraser · 2020-02-26T13:26:12Z

src/main/scala/zio/kafka/client/Producer.scala

@@ -2,6 +2,7 @@ package zio.kafka.client

 import java.util.concurrent.atomic.AtomicLong

+import izreflect.fundamentals.reflection.Tags.Tag


I think this should be able to go away since you already have Tagged in scope through the wildcard import.

kciesielski · 2020-02-27T13:39:27Z

src/test/scala/zio/kafka/client/KafkaTestUtils.scala

+
+  val testProducer: ZLayer[Kafka, Throwable, StringProducer] =
+    (for {
+      settings <- producerSettings.toManaged_


I have a feeling that this can be written in a simpler way, without converting the settings to a Managed and then back to Layer.

I think this is indicating that we may want to express the dependencies of the producer as services as well:

object producer { def live[R, K, B]: ZLayer[Kafka with Serializer[R, K] with Serializer[R, V], Throwable, Consumer] = ??? } object KafkaTestUtils { def testProducer: ZLayer[Kafka, Throwable, StringProducer] = (ZLayer.identity[Kafka] ++ ZLayer.succeed(Serde.string)) >>> producer.live

My intuition says that we shouldn't force users that want to use ZLayer to provide everything (incl. settings and serdes) as R, but maybe that's misguided.

At any rate, I'm fine with this being written as-is.

I think if we had a definition of live like the above we could then offer a method that would take the settings and series as normal parameters and lift them into services so we could provide the same user experience but have the layer be more composable internally.

That's a good solution 👍🏻

kciesielski · 2020-02-27T13:42:51Z

src/test/scala/zio/kafka/client/KafkaTestUtils.scala

+  def withProducerStrings[R: Tag, A: Tag](
+    r: Producer.Service[Any, String, String] => RIO[R, A]
+  ): ZIO[R with StringProducer, Throwable, A] =
+    ZLayer.fromService(r).build.use(_.get)


Similarly here, I think I should be able to express this without going through a ZLayer.

ZIO.accessM(env => r(env.get))?

adamgfraser · 2020-02-27T15:05:07Z

src/main/scala/zio/kafka/client/Producer.scala

+     * Flushes the producer's internal buffer. This will guarantee that all records
+     * currently buffered will be transmitted to the broker.
+     */
+    def flush: BlockingTask[Unit] = effectBlocking(p.flush())


Do you want this to be part of the API? If so I think you want to make it a method on Service otherwise you are going to be in an awkward position where it is only available on Live instead of Service but when you access through the environment you are just going to get Service so you are not going to be able to call flush.

Right, flush should definitely be part of the API :-)

adamgfraser · 2020-02-27T15:07:27Z

src/main/scala/zio/kafka/client/Producer.scala

+      ZSink.drain.contramapM(produceChunk)
+  }
+
+  class Live[R, K, V](


final case class?

adamgfraser · 2020-02-27T15:17:40Z

src/main/scala/zio/kafka/client/Producer.scala

+    valueSerializer: Serializer[R, V]
+  ): Live[R, K, V] = new Live(p, keySerializer, valueSerializer)
+
+  def producer[R, K, V](


Maybe call this make?

Yes, let's keep it as make. That's the conventional name for effectful creation of things in ZIO.

adamgfraser · 2020-02-27T15:17:47Z

src/main/scala/zio/kafka/client/Producer.scala

-   * @return Producer as a Managed resource that will be closed automatically after use.
-   */
-  def make[R, K, V](
+  private def createLive[R, K, V](


Maybe inline this? You are only calling it one place in make and it is taking significantly more space just to right the method signature than the actual implementation.

adamgfraser · 2020-02-27T15:19:45Z

src/main/scala/zio/kafka/client/Producer.scala

+    ZLayer.fromManaged(
+      p.toManaged(p => UIO(p.close(settings.closeTimeout.asJava)))
+        .map(producer => createLive(producer, keySerializer, valueSerializer))
+    )
  }


Normally at this point you would write your environmental accessors. For example:

def produce[R, K, V](record: ProducerRecord[R, K, V]): RIO[R with Producer with Blocking, Task[RecordMetadata]] = ZIO.accessM(_.get.produce(record))

This will dramatically improve the ergonomics of working with the producer in the environment.

adamgfraser · 2020-02-27T15:20:03Z

src/main/scala/zio/kafka/client/Producer.scala

+    ZLayer.fromManaged(
+      p.toManaged(p => UIO(p.close(settings.closeTimeout.asJava)))
+        .map(producer => createLive(producer, keySerializer, valueSerializer))
+    )
  }


Normally at this point you would write your environmental accessors. For example:

def produce[R, K, V](record: ProducerRecord[R, K, V]): RIO[R with Producer with Blocking, Task[RecordMetadata]] = ZIO.accessM(_.get.produce(record))

This will dramatically improve the ergonomics of working with the producer in the environment.

adamgfraser · 2020-02-27T15:23:07Z

src/main/scala/zio/kafka/client/Producer.scala

+    ZLayer.fromManaged(
+      p.toManaged(p => UIO(p.close(settings.closeTimeout.asJava)))
+        .map(producer => createLive(producer, keySerializer, valueSerializer))
+    )
  }


Normally at this point you would write your environmental accessors. For example:

def produce[R, K, V](record: ProducerRecord[R, K, V]): RIO[R with Producer with Blocking, Task[RecordMetadata]] = ZIO.accessM(_.get.produce(record))

This will dramatically improve the ergonomics of working with the producer in the environment.

adamgfraser · 2020-02-27T15:24:42Z

src/main/scala/zio/kafka/client/Producer.scala

@@ -139,24 +142,9 @@ object Producer {
      )
    }



It may not be possible given the degree of polymorphism here but are there any specialized "default" constructors that would provide sensible implementations of one or more of the parameters to make that would make it a little easier to create the producer layer for common cases?

I thought of providing a version that takes String serializers without dependencies, like makeSimpleStrings(settings: ProducerSettings), + maybe another one for Array[Byte].

adamgfraser · 2020-02-27T15:26:20Z

src/test/scala/zio/kafka/client/KafkaTestUtils.scala

@@ -2,8 +2,9 @@ package zio.kafka.client

 import java.util.UUID

+import izreflect.fundamentals.reflection.Tags.Tag


I think the Tag concept should be able to go away here as well in favor of Tagged.

adamgfraser · 2020-02-27T15:29:30Z

src/main/scala/zio/kafka/client/Producer.scala

@@ -139,24 +142,9 @@ object Producer {
      )
    }



It may not be possible given the degree of polymorphism here but are there any specialized "default" constructors that would provide sensible implementations of one or more of the parameters to make that would make it a little easier to create the producer layer for common cases?

iravid

@kciesielski Thank you for picking this up! Looking great so far. We just need accessors and to re-enable the tests 💪🏻

iravid · 2020-02-27T19:38:27Z

And thank you @adamgfraser for the comprehensive review!

kciesielski · 2020-03-03T08:16:20Z

src/main/scala/zio/kafka/client/Producer.scala

-    }
+  )(
+    implicit ts: Tagged[Serializer[R, K]],
+    td: Tagged[Serializer[R, V]],


@adamgfraser I can't seem to get rid of these implicit Tagged requirements here.

iravid

Looks great. Amazing work @kciesielski!

kciesielski added 2 commits February 26, 2020 13:10

Switch to zio with paremeterized service support

d91821a

Use a ZLayer for producer

e0b38fc

adamgfraser reviewed Feb 26, 2020

View reviewed changes

src/main/scala/zio/kafka/client/Producer.scala Outdated Show resolved Hide resolved

adamgfraser reviewed Feb 26, 2020

View reviewed changes

src/main/scala/zio/kafka/client/package.scala Outdated Show resolved Hide resolved

adamgfraser reviewed Feb 26, 2020

View reviewed changes

src/main/scala/zio/kafka/client/Producer.scala Outdated Show resolved Hide resolved

adamgfraser reviewed Feb 26, 2020

View reviewed changes

kciesielski added 6 commits February 26, 2020 17:20

Fix type name and organize imports

1aff844

Correctly deal with Tagged constraint

046f5d9

Organize comments

673d0bf

Adjust producer specs to ZLayers

5393e1c

Temporarily comment out consumer specs

d7846b0

Simplify withProducerStrings

7c1d566

kciesielski commented Feb 27, 2020

View reviewed changes

adamgfraser reviewed Feb 27, 2020

View reviewed changes

iravid reviewed Feb 27, 2020

View reviewed changes

kciesielski added 2 commits March 3, 2020 09:14

Update Producer layer API after reviews

b09e56f

Use updated Producer API in test utils

2070439

kciesielski commented Mar 3, 2020

View reviewed changes

kciesielski added 2 commits March 3, 2020 18:35

Remove unnecessary val

61b0981

Consumer implementation with ZLayers

e274679

kciesielski added 4 commits March 3, 2020 19:58

Update scaladoc with consumer example

14cca72

Restore the "non-empty chunk" producer test

56dd2fe

Adjust consumer tests to ZLayers

77ab974

Switch to ZIO 1.0-RC18

3cb7c61

kciesielski changed the title ~~[WIP] Producer and Consumer services as ZLayers~~ Producer and Consumer services as ZLayers Mar 4, 2020

kciesielski added 4 commits March 4, 2020 12:27

Expose unsubscribe and restore Scala 2.12 compatibility

8517d89

Add producer accessors

1849943

Add consumer accessors

4a1ab99

Add explicit types to help inference for Scala 2.12

db8a88b

iravid approved these changes Mar 5, 2020

View reviewed changes

iravid merged commit 952b0e8 into zio:master Mar 5, 2020

svroonland mentioned this pull request Mar 10, 2020

Modules for the Consumer and Producer #106

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Producer and Consumer services as ZLayers #128

Producer and Consumer services as ZLayers #128

kciesielski commented Feb 26, 2020 •

edited

adamgfraser Feb 26, 2020

adamgfraser Feb 26, 2020

kciesielski Feb 27, 2020 •

edited

adamgfraser Feb 27, 2020

iravid Feb 27, 2020

adamgfraser Feb 27, 2020

iravid Feb 27, 2020

kciesielski Feb 27, 2020

adamgfraser Feb 27, 2020

adamgfraser Feb 27, 2020

iravid Feb 27, 2020

adamgfraser Feb 27, 2020

adamgfraser Feb 27, 2020

iravid Feb 27, 2020

adamgfraser Feb 27, 2020

adamgfraser Feb 27, 2020

adamgfraser Feb 27, 2020

adamgfraser Feb 27, 2020

adamgfraser Feb 27, 2020

kciesielski Mar 2, 2020

adamgfraser Feb 27, 2020

adamgfraser Feb 27, 2020

iravid left a comment

iravid commented Feb 27, 2020

kciesielski Mar 3, 2020 •

edited

iravid left a comment

		@@ -2,6 +2,7 @@ package zio.kafka.client

		import java.util.concurrent.atomic.AtomicLong

		import izreflect.fundamentals.reflection.Tags.Tag

		@@ -2,8 +2,9 @@ package zio.kafka.client

		import java.util.UUID

		import izreflect.fundamentals.reflection.Tags.Tag

Producer and Consumer services as ZLayers #128

Producer and Consumer services as ZLayers #128

Conversation

kciesielski commented Feb 26, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kciesielski Feb 27, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iravid left a comment

Choose a reason for hiding this comment

iravid commented Feb 27, 2020

kciesielski Mar 3, 2020 • edited

Choose a reason for hiding this comment

iravid left a comment

Choose a reason for hiding this comment

kciesielski commented Feb 26, 2020 •

edited

kciesielski Feb 27, 2020 •

edited

kciesielski Mar 3, 2020 •

edited