如何创建斯卡拉在FS2功能流分离的流?

问题描述 投票:3回答:1

是否有可能创造FS2离散事件流?如果是如何做到这一点。我刚开始与库玩,我知道我有很多的研究。但我没有看到任何相关的例子。例如我想创建一个流为“鼠标移动”或scalajs或摇摆“点击”。我要寻找一个在RxJS类似的东西我可以用一个Rx.Observable.create创建离散事件是这样的:

//note: pseudo code
var mouse = Rx.Observable.create( subscriber => {
     document.body.addEventListener("mousemove", event =>{
      subscriber.onNext(event)
 })
} ) 

在FS2等效可能没那么微不足道,但如果任何人都可以建议我怎么样。我想这将是使用处理器和推/拉的数据类型,但我很远,以了解如何。

干杯。

scala functional-programming stream system.reactive fs2
1个回答
1
投票

这是我想出了演示如何使用FS2 JavaFX的一个例子:

import cats.implicits._
import cats.effect._
import cats.effect.implicits._
import javafx.application.{Application, Platform}
import javafx.scene.{Node, Scene}
import javafx.scene.layout._
import javafx.stage.Stage
import fs2._
import fs2.concurrent._
import javafx.beans.value.WritableValue
import javafx.scene.control.{Label, TextField}
import javafx.scene.input.KeyEvent

import scala.concurrent.ExecutionContext

import scala.util.Try

class Fs2Ui extends Application {
  override def start(primaryStage: Stage): Unit = {
    implicit val cs: ContextShift[IO] = IO.contextShift(ExecutionContext.global)
    implicit val timer: Timer[IO] = IO.timer(ExecutionContext.global)

    new Logic[IO]().run(primaryStage).start.unsafeRunSync()
  }

  class Logic[F[_]: ConcurrentEffect: ContextShift: Timer] {
    import Fs2Ui._
    import java.time.{Duration, Instant}
    import java.util.concurrent.TimeUnit.MILLISECONDS

    def run(primaryStage: Stage): F[Unit] = for {
      v <- initializeUi(primaryStage)
      View(input, feedback) = v

      _ <- Stream(input).covary[F]
        .through(typedChars)
        .through(processInput)
        .through(displayFeedback(feedback.textProperty))
        .compile.drain
    } yield ()

    private def initializeUi(primaryStage: Stage): F[View] = updateUi {
      val input = new TextField()
      input.setPrefWidth(300)
      val feedback = new Label("...")

      val vbox = new VBox(input, feedback)
      val root = new StackPane(vbox)
      val scene = new Scene(root)

      primaryStage.setScene(scene)
      primaryStage.show()

      View(input, feedback)
    }

    private def processInput: Pipe[F, TypedChar, Feedback] = for {
      typed <- _
      _ <- Stream.eval(ContextShift[F].shift)
      res <- Stream.eval { time(processSingle(typed)) }
      (d, Feedback(str)) = res
    } yield Feedback(s"$str in [$d]")

    private def displayFeedback(value: WritableValue[String]): Pipe[F, Feedback, Unit] =
      _.map { case Feedback(str) => str } through updateValue(value)

    private def time[A](f: F[A]): F[(Duration, A)] = {
      val now = Timer[F].clock.monotonic(MILLISECONDS).map(Instant.ofEpochMilli)
      for {
        start <- now
        a <- f
        stop <- now
        d = Duration.between(start, stop)
      } yield (d, a)
    }

    private val processSingle: TypedChar => F[Feedback] = {
      import scala.util.Random
      import scala.concurrent.duration._

      val prng = new Random()
      def randomDelay: F[Unit] = Timer[F].sleep { (250 + prng.nextInt(750)).millis }

      c => randomDelay *> Sync[F].delay(Feedback(s"processed $c"))
    }
  }
}

object Fs2Ui {
  case class View(input: TextField, feedback: Label)

  case class TypedChar(value: String)
  case class Feedback(value: String)

  private def typedChars[F[_]: ConcurrentEffect]: Pipe[F, Node, TypedChar] = for {
    node <- _
    q <- Stream.eval(Queue.unbounded[F, KeyEvent])
    _ <- Stream.eval(Sync[F].delay {
      node.setOnKeyTyped { evt => (q enqueue1 evt).toIO.unsafeRunSync() }
    })
    keyEvent <- q.dequeue
  } yield TypedChar(keyEvent.getCharacter)

  private def updateValue[F[_]: Async, A](value: WritableValue[A]): Pipe[F, A, Unit] = for {
    a <- _
    _ <- Stream.eval(updateUi(value setValue a))
  } yield ()

  private def updateUi[F[_]: Async, A](action: => A): F[A] =
    Async[F].async[A] { cb =>
      Platform.runLater { () =>
        cb(Try(action).toEither)
      }
    }
}

演示FS2和JavaFX之间的绑定的特定部分是两个Pipes:typedCharsupdateValue。我个人认为,最有挑战的是一个适应听众KeyEvent看起来像事件的FS2 Stream

node.setOnKeyTyped { evt => (q enqueue1 evt).toIO.unsafeRunSync() }
© www.soinside.com 2019 - 2024. All rights reserved.