RxScala scheduler for ScalaFX

ScalaFXScheduler

package io.github.vigoo.rx.schedulers

import java.util.{TimerTask, Timer}
import java.util.concurrent.TimeUnit

import rx.Scheduler.Worker
import rx.Subscription
import rx.functions.Action0
import rx.lang.scala.Scheduler
import rx.subscriptions.{Subscriptions, BooleanSubscription, CompositeSubscription}

import scalafx.application.Platform

class ScalaFXScheduler extends Scheduler {

  val asJavaScheduler = new rx.Scheduler {
    override def createWorker(): Worker = new ScalaFXWorker()
  }
}

class ScalaFXWorker extends rx.Scheduler.Worker {

  private val subscriptions = new CompositeSubscription()
  private val timer = new Timer()

  override def schedule(action: Action0): Subscription = {
    val subscription = new BooleanSubscription()
    subscriptions.add(subscription)

    Platform.runLater {
      if (!subscriptions.isUnsubscribed && !subscription.isUnsubscribed) {
        action.call()
        subscriptions.remove(subscription)
      }
    }

    Subscriptions.create(new Action0 {
      def call() = {
        subscription.unsubscribe()
        subscriptions.remove(subscription)
      }
    })
  }

  override def schedule(action: Action0, delayTime: Long, unit: TimeUnit): Subscription = {
    val subscription = new BooleanSubscription()
    subscriptions.add(subscription)

    timer.schedule(new TimerTask {
      override def run(): Unit = {
        Platform.runLater {
          if (!subscriptions.isUnsubscribed && !subscription.isUnsubscribed) {
            action.call()
            subscriptions.remove(subscription)
          }
        }
      }
    }, unit.toMillis(delayTime))

    Subscriptions.create(new Action0 {
      def call() = {
        subscription.unsubscribe()
        subscriptions.remove(subscription)
      }
    })
  }

  override def isUnsubscribed: Boolean = subscriptions.isUnsubscribed

  override def unsubscribe(): Unit = subscriptions.unsubscribe()
}