我想解决的是下面这个问题：
给定一个无限运行的Akka流，我希望能够监视流的某些点。我能想到的最好的方法是将消息发送到一个Actor，它也是一个Source。这使得我可以非常灵活地将单个源连接到WebSocket或将多个源合并到我想要连接的任何其他客户端。但是在这个特定的例子中，我尝试连接ScalaFX和Akka Source，但是它没有像预期的那样工作。
当我运行下面的代码时，两个计数器都开始正常，但过了一会儿，其中一个停止了，再也没有恢复。我知道在使用ScalaFX时有一些关于线程的特殊注意事项，但是我没有足够的知识来理解这里发生了什么或调试它。下面是一个运行的最小示例，大约5秒后应该可以看到问题。
我的问题是：
我怎样才能改变这段代码，让它像预期的那样工作？

import akka.NotUsed

import scalafx.Includes._
import akka.actor.{ActorRef, ActorSystem}
import akka.stream.{ActorMaterializer, OverflowStrategy, ThrottleMode}
import akka.stream.scaladsl.{Flow, Sink, Source}

import scalafx.application.JFXApp
import scalafx.beans.property.{IntegerProperty, StringProperty}
import scalafx.scene.Scene
import scalafx.scene.layout.BorderPane
import scalafx.scene.text.Text
import scala.concurrent.duration._

/**
  * Created by henke on 2017-06-10.
  */
object MonitorApp extends JFXApp {

  implicit val system = ActorSystem("monitor")
  implicit val mat = ActorMaterializer()

  val value1 = StringProperty("0")
  val value2 = StringProperty("0")

  stage = new JFXApp.PrimaryStage {
    title = "Akka Stream Monitor"
    scene = new Scene(600, 400) {
      root = new BorderPane() {
        left = new Text {
          text <== value1
        }
        right = new Text {
          text <== value2
        }
      }
    }
  }

  override def stopApp() = system.terminate()

  val monitor1 = createMonitor[Int]
  val monitor2 = createMonitor[Int]

  val marketChangeActor1 = monitor1
    .to(Sink.foreach{ v =>
      value1() = v.toString
    }).run()

  val marketChangeActor2 = monitor2
    .to(Sink.foreach{ v =>
      value2() = v.toString
    }).run()

  val monitorActor = Source[Int](1 to 100)
    .throttle(1, 1.second, 1, ThrottleMode.shaping)
    .via(logToMonitorAndContinue(marketChangeActor1))
    .map(_ * 10)
    .via(logToMonitorAndContinue(marketChangeActor2))
    .to(Sink.ignore).run()

  def createMonitor[T]: Source[T, ActorRef] = Source.actorRef[T](Int.MaxValue, OverflowStrategy.fail)

  def logToMonitorAndContinue[T](monitor: ActorRef): Flow[T, T, NotUsed] = {
    Flow[T].map{ e =>
      monitor ! e
      e
    }
  }
}