Refactor Producer Results + Better FS2 Error Handling

etspaceman · Jun 5, 2023 · 5ac970e · 5ac970e
1 parent 01f2353
commit 5ac970e
Show file tree

Hide file tree

Showing 15 changed files with 402 additions and 410 deletions.
diff --git a/...main/scala/kinesis4cats/client/producer/fs2/localstack/LocalstackFS2KinesisProducer.scala b/...main/scala/kinesis4cats/client/producer/fs2/localstack/LocalstackFS2KinesisProducer.scala
@@ -52,7 +52,7 @@ object LocalstackFS2KinesisProducer {
           KinesisClient[F],
           StreamNameOrArn
       ) => F[Either[ShardMapCache.Error, ShardMap]],
-      callback: Producer.Res[PutRecordsResponse] => F[Unit]
+      callback: Producer.Result[PutRecordsResponse] => F[Unit]
   )(implicit F: Async[F]) {
 
     def withLocalstackConfig(localstackConfig: LocalstackConfig): Builder[F] =
@@ -139,7 +139,7 @@ object LocalstackFS2KinesisProducer {
         Nil,
         (client: KinesisClient[F], snoa: StreamNameOrArn) =>
           KinesisProducer.getShardMap(client, snoa),
-        (_: Producer.Res[PutRecordsResponse]) => F.unit
+        (_: Producer.Result[PutRecordsResponse]) => F.unit
       )
 
     @annotation.unused

diff --git a/kinesis-client/src/main/scala/kinesis4cats/client/producer/fs2/FS2KinesisProducer.scala b/kinesis-client/src/main/scala/kinesis4cats/client/producer/fs2/FS2KinesisProducer.scala
@@ -52,7 +52,9 @@ final class FS2KinesisProducer[F[_]] private[kinesis4cats] (
     override protected val channel: Channel[F, Record],
     override protected val underlying: KinesisProducer[F]
 )(
-    override protected val callback: Producer.Res[PutRecordsResponse] => F[Unit]
+    override protected val callback: Producer.Result[PutRecordsResponse] => F[
+      Unit
+    ]
 )(implicit
     F: Async[F]
 ) extends FS2Producer[F, PutRecordsRequest, PutRecordsResponse]
@@ -63,7 +65,7 @@ object FS2KinesisProducer {
       clientResource: Resource[F, KinesisClient[F]],
       encoders: KinesisProducer.LogEncoders,
       logger: StructuredLogger[F],
-      callback: Producer.Res[PutRecordsResponse] => F[Unit]
+      callback: Producer.Result[PutRecordsResponse] => F[Unit]
   )(implicit F: Async[F]) {
     def withConfig(config: FS2Producer.Config[F]): Builder[F] = copy(
       config = config
@@ -108,7 +110,7 @@ object FS2KinesisProducer {
       KinesisClient.Builder.default.build,
       KinesisProducer.LogEncoders.show,
       Slf4jLogger.getLogger,
-      (_: Producer.Res[PutRecordsResponse]) => F.unit
+      (_: Producer.Result[PutRecordsResponse]) => F.unit
     )
 
     @annotation.unused