IC-40 Shows input stream progress

Author: fernandomora

src/main/scala/com/intenthq/action_processor/integrations/feeds/Feed.scala

@@ -6,6 +6,8 @@ import cats.effect.IO
 
 trait Feed[I, O] {
 
+  def toString(i: I): String = i.toString
+
   lazy val feedName: String = getClass.getSimpleName.stripSuffix("$")
   def date(feedContext: FeedContext[IO], clock: Clock = Clock.systemDefaultZone()): IO[LocalDate] =
     feedContext.filter.date.fold(IO.delay(java.time.LocalDate.now(clock)))(IO.pure)
@@ -14,9 +16,10 @@ trait Feed[I, O] {
   def inputStream(feedContext: FeedContext[IO]): fs2.Stream[IO, I]
   def transform(feedContext: FeedContext[IO]): fs2.Pipe[IO, I, (O, Long)]
   def serialize(o: O, counter: Long): Array[Byte]
+  val serialization: fs2.Pipe[IO, (O, Long), Array[Byte]] = _.map((serialize _).tupled)
 
-  final def stream(feedContext: FeedContext[IO]): fs2.Stream[IO, Array[Byte]] =
+  def stream(feedContext: FeedContext[IO]): fs2.Stream[IO, Array[Byte]] =
     inputStream(feedContext)
       .through(transform(feedContext))
-      .map((serialize _).tupled)
+      .through(serialization)
 }