Fs2 streams processing time in Queue

Question

I'm exploring fs2 library and for testing in I have a Stream[IO, Int] that are pushed into a Queue[IO,(Double, String, String, String)]. I need to identify the entry time to the queue as well as the exit time.

The problem here is that the processing time may vary among elements. For this, I thought of using .metered((Random.between(1,10)).seconds) The issue is that I can not figure out a way in order to store entry time at the time of enqueuing and the exitTime after a certain amount of time given with the .metered((Random.between(1,10)).seconds)

Here is what I have tried :

import cats.effect.{ExitCode, IO, IOApp, Timer}
import fs2._
import fs2.concurrent.Queue

import scala.concurrent.duration._
import scala.util.Random
class StreamTypeIntToDouble(q: Queue[IO, (Double, String, String, String)])(
  implicit timer: Timer[IO]
) {
  import core.Processing._
  def storeInQueue: Stream[IO, Unit] = {

    Stream(1, 2, 3)
      .covary[IO]
      .evalTap(n => IO.delay(println(s"Pushing $n to Queue")))
      .map { n =>
        val entryTime = currentTimeNow
        val exitTime = currentTimeNow
        (n.toDouble, "Service", entryTime, exitTime)
      }
      .metered(Random.between(1, 20).seconds)
      .through(q.enqueue)
    //.metered((Random.between(1, 10).seconds))

  }
  def getFromQueue: Stream[IO, Unit] = {
    q.dequeue
      .evalMap(n => IO.delay(println(s"Pulling from queue $n")))
  }
}

object Five extends IOApp {
  override def run(args: List[String]): IO[ExitCode] = {
    val program = for {
      q <- Queue.bounded[IO, (Double, String, String, String)](10)
      b = new StreamTypeIntToDouble(q)
      _ <- b.storeInQueue.compile.drain.start
      _ <- b.getFromQueue.compile.drain
    } yield ()
    program.as(ExitCode.Success)
  }
}

The method used for identifying the current time is implemented as follows

  def currentTimeNow: String = {
        val format = new SimpleDateFormat("dd-MM-yy hh:mm:ss")
        format.format(Calendar.getInstance().getTime())
      }

Question : How can I keep track of the entry and exit time to construct the new output stream?

Fs2 streams processing time in Queue

0 Answers