Как я могу повторно использовать очередь в другом вычислении? поток fs2 Scala - PullRequest
Новая
       108

Как я могу повторно использовать очередь в другом вычислении? поток fs2 Scala

0 голосов
/ 19 июня 2020

Я получаю сообщение об ошибке Unit вместо Stream[IO, String]. Я пытаюсь повторно использовать результат очереди в следующей очереди 

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

import scala.concurrent.duration._
import scala.util.Random

class StreamTypeIntToDouble(q1: Queue[IO, Int], q2: Queue[IO, String])(
  implicit timer: Timer[IO]
) {

  def storeInQueueFirst: Stream[IO, Unit] = {
    Stream(1, 2, 3)
      .covary[IO]
      .evalTap(n => IO.delay(println(s"Pushing $n to Queue First")))
      .metered(Random.between(1, 20).seconds)
      .through(q1.enqueue)

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

  }
  def storeInQueueSecond(s: Stream[IO, Int]): Stream[IO, Unit] = {
    s.map { n =>
        n.toString
      }
      .metered(Random.between(1, 20).seconds)
      .through(q2.enqueue)
  }

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

object Five extends IOApp {
  override def run(args: List[String]): IO[ExitCode] = {
    val program = for {
      q1 <- Queue.bounded[IO, Int](10)
      q2 <- Queue.bounded[IO, String](10)

      b = new StreamTypeIntToDouble(q1, q2)
      _ <- b.storeInQueueFirst.compile.drain.start
      a <- b.getFromQueueFirst.compile.drain
      _ <- b.storeInQueueSecond(a).compile.drain
      _ <- b.getFromQueueSecond.compile.drain
    } yield ()
    program.as(ExitCode.Success)
  }
}

1 Ответ

1 голос
/ 19 июня 2020

Попробуйте изменить getFromQueueFirst так, чтобы он давал Stream[IO, Int], а не Stream[IO, Unit]

def getFromQueueFirst: Stream[IO, Int] = {
  q1.dequeue
    evalTap(n => IO.delay(println(s"Pulling from queue Second $n")))
}

Затем 

val program = for {
  q1 <- Queue.bounded[IO, Int](10)
  q2 <- Queue.bounded[IO, String](10)

  b = new StreamTypeIntToDouble(q1, q2)
  _ <- b.storeInQueueFirst.compile.drain.start
  a <- b.getFromQueueFirst.compile.lastOrError
  _ <- b.storeInQueueSecond(Stream(a)).compile.drain
  _ <- b.getFromQueueSecond.compile.drain
} yield ()

компилируется.

...