Make Observable an instance of Monad

This is not yet tested, and it requires a lot of testing. A previous
helper that attempted to support `bind` never quite worked correctly,
but I believe that the new approach of implementing `join` and `map`
instead of `bind` makes the code more resilient.

Even though `map` and `ap` can be derived from `point` and `bind`, I'm
keeping both of them, as `bind` requires quite a bit of subscription
manipulation. Those methods have much simpler implementations.

src/main/scala/org/gerweck/scalafx/util/observable.scala

@@ -8,7 +8,7 @@ import scalafx.beans.property._
 import scalafx.beans.value._
 
 trait ObservableImplicits {
-  implicit val observableApplicative = new Applicative[Observable] with Functor[Observable] {
+  implicit val observableInstances = new Applicative[Observable] with Functor[Observable] with Monad[Observable] {
     /* Map can be derived from `ap`, but this adds less overhead. */
     override def map[A, B](a: Observable[A])(f: A => B): ObservableValue[B, B] = {
       @inline def recalculate(): B = f(a.value)
@@ -29,7 +29,8 @@ trait ObservableImplicits {
       ObjectProperty[A](a)
     }
 
-    def ap[A, B](fa: => Observable[A])(f: => Observable[A => B]): ObservableValue[B, B] = {
+    /* Ap can be derived from `point` and `bind`, but this has less overhead. */
+    override def ap[A, B](fa: => Observable[A])(f: => Observable[A => B]): ObservableValue[B, B] = {
       @inline def recalculate(): B = (f.value)(fa.value)
 
       val originalValue = recalculate()
@@ -51,6 +52,48 @@ trait ObservableImplicits {
 
       prop
     }
+
+    /* Aka `flatMap` */
+    override def bind[A, B](fa: Observable[A])(f: A => Observable[B]): ObservableValue[B, B] = {
+      join(map(fa)(f))
+    }
+
+    /* Aka `flatten` */
+    override def join[A](ooa: Observable[Observable[A]]): ObservableValue[A, A] = {
+      @inline def oa() = ooa.value
+      @inline def calc(): A = oa().value
+
+      val originalValue = calc()
+
+      val prop = ObjectProperty[A](originalValue)
+
+      var prevValue = originalValue
+
+      def innerHandle() = {
+        val newVal = calc()
+        if (prevValue != newVal) {
+          prop.value = newVal
+          prevValue = newVal
+        }
+      }
+      var innerSub = oa() onChange innerHandle
+
+      var prevOuter = oa()
+      def outerHandle() = {
+        val newOuter = oa()
+        /* We need reference equality here: we're subscribing to a specific object. */
+        if (prevOuter ne newOuter) {
+          innerSub.cancel()
+          innerSub = newOuter onChange innerHandle
+          prevOuter = newOuter
+          innerHandle()
+        }
+      }
+
+      ooa onChange outerHandle
+
+      prop
+    }
   }
 
   implicit def enrichObservable[A, B](o: ObservableValue[A, B]) = new RichObservable(o)
@@ -94,9 +137,10 @@ class RichTuple[A <: Product](val self: A) extends AnyVal {
 
 class RichObservable[A, C](val self: ObservableValue[A, C]) extends AnyVal {
   private type ObjObs[X] = ObservableValue[X, X]
-  @inline private def oapp = observableApplicative
+  @inline private def oapp = observableInstances
 
   def map[B](f: A => B) = oapp.map(self)(f)
+  def flatMap[B](f: A => Observable[B]) = oapp.bind(self)(f)
   def <*>[B](f: Observable[A => B]): Observable[B] = oapp.ap(self)(f)
   def tuple[B](f: Observable[B]): Observable[(A,B)] = oapp.tuple2(self, f)
   final def *>[B](fb: ObjObs[B]): Observable[B] = oapp.apply2(self,fb)((_,b) => b)