untested; updated to use scala.rx