use Chunk.toNel with NonEmptyTraverse

Banno · Dec 28, 2023 · 3253892 · 3253892
1 parent 590ffc2
commit 3253892
Showing 1 changed file with 9 additions and 0 deletions.
diff --git a/core/src/main/scala/com/banno/kafka/producer/ProducerOps.scala b/core/src/main/scala/com/banno/kafka/producer/ProducerOps.scala
@@ -17,6 +17,7 @@
 package com.banno.kafka.producer
 
 import cats.*
+import cats.data.NonEmptyList
 import cats.syntax.all.*
 import fs2.*
 import org.apache.kafka.common.*
@@ -115,6 +116,14 @@ case class ProducerOps[F[_], K, V](producer: ProducerApi[F, K, V]) {
     s =>
       pipeSendBatch[Chunk](Traverse[Chunk], F)(s.chunks).flatMap(Stream.chunk)
 
+  def pipeSendBatchChunksNonEmpty(implicit
+      F: FlatMap[F]
+  ): Pipe[F, ProducerRecord[K, V], RecordMetadata] =
+    s =>
+      pipeSendBatchNonEmpty[NonEmptyList](NonEmptyTraverse[NonEmptyList], F)(
+        s.chunks.map(_.toNel).unNone
+      ).flatMap(nel => Stream.emits(nel.toList))
+
   /** Calls chunkN on the input stream, to create chunks of size `n`, and sends
     * those chunks as batches to the producer.
     */