       dx: moveEv.clientX - startpos.x
       dy: moveEv.clientY - startpos.y
 
-app.demos.mouseMove = ->
-  mouseDrags(document).subscribe (drag) ->
-    app.events.mouseDemo.onNext(drag)
-  mouseUp.subscribe () ->
-    app.events.mouseDemo.onNext(
-      startpos: {x: 0, y: 0}
-      target: ''
-      dx: 0
-      dy: 0)
-
-app.demos.timer = () ->
-  timerOn = true
-  timerSwitch = new Subject()
-  Observable
-    .interval(500) # ms
+mouseDragResetEv =
+  startpos: {x: 0, y: 0}
+  target: ''
+  dx: 0
+  dy: 0
+
+demos.mouseDrags = (eventSources) ->
+  dragsSource = mouseDrags(document)
+  dragsSource.subscribe (drag) -> eventSources.mouseDrags.onNext(drag)
+  dragsSource.subscribe(logAllObserver('drags'))
+  mouseUp.subscribe () -> eventSources.mouseDrags.onNext(mouseDragResetEv)
+
+demos.timer = (eventSources) ->
+  ticks = Observable
+    .interval(1000) # ms
     .timeInterval()
-    .pausable(timerSwitch)
+    .pausable(eventSources.timerSwitch)
     .pluck("value")
-    .subscribe(log)
-  Observable.fromEvent(document, 'click')
-    .subscribe ->
-      timerOn = !timerOn
-      if timerOn
-        console.log("timer on")
-      else
-        console.log("timer off")
-      timerSwitch.onNext(timerOn)
-
-stackTrace = () -> new Error().stack
-app.demos.toAsync = () ->
+  ticks.subscribe((tick) -> eventSources.timer.onNext(tick))
+  ticks.subscribe(logAllObserver('timer'))
+
+################################################################################
+# additional misc Observable demos
+
+demos.toAsync = () ->
   f = (arg) ->
     "async: #{arg} #{stackTrace()}"
   fasync = Observable.toAsync(f)
   Observable.interval(1000).flatMap(fasync).subscribe(log)
 
-app.demos.replay = () ->
+demos.replay = () ->
   sub = new Rx.ReplaySubject()
-  # bufferSize, windowSize, scheduler
+  # ReplaySubject args: bufferSize, windowSize, scheduler
   sub.onNext(2)
   sub.onNext(3)
-  sub.subscribe(log)
+  sub.subscribe(logAllObserver('replay'))
   sub.onNext(4)
-
-app.demos.behaviours = () ->
+  sub.onCompleted()
+  sub.onNext(5)
+
+demos.behaviours = () ->
   sub = new Rx.BehaviorSubject(42)
   # same as `sub = new Rx.ReplaySubject(1); sub.onNext(42)`
   sub.onNext(2)
   sub.onNext(3)
-  sub.subscribe(log)
+  sub.subscribe(logAllObserver('behaviours'))
   sub.onNext(4)
-
-app.demos.backpressure = () ->
+  sub.onCompleted()
+
+demos.backpressure = () ->
   source = Observable.range(0, 10).timeInterval().controlled()
-  source.subscribe(logWith("controlled:"))
+  source.subscribe(logAllObserver("controlled"))
   source.request(4)
   source.request(4)
-
+  sub.onCompleted()
+  source.request(4)
+
+################################################################################
+# Demo UI using React.js
+#
+
+# a mixin to handle Rx <-> React boilerplate
+ReactRxMixin =
+  componentWillMount: ->
+    @subscriptions = []
+  componentWillUnmount: ->
+    # this is not essential in this demo, but keep cleanup of
+    # subscriptions in mind
+    log("unmounting and cleaning up subscriptions: ", @)
+    @subscriptions.forEach((sub) -> sub.dispose())
+  manageSubcription: (sub) ->
+    @subscriptions.push(sub)
+  observeDom: (domEvent, observer) ->
+    # see http://blog.ankl.am/rx-101-event-delegation/ for more sophisticated approaches
+    @manageSubcription(Observable.fromEvent(@getDOMNode(), domEvent).subscribe(observer))
+
+RequestsWidget = React.createClass
+  mixins: [ReactRxMixin]
+  componentWillMount: ->
+    @manageSubcription(@props.events.subscribe(@setState.bind(@)))
+  getInitialState: ->
+    '1m':  '-'
+    '5m':  '-'
+    '15m': '-'
+    count: '-'
+    mean:  '-'
+  render: ->
+    R.div {},
+      (R.h2 {}, "Vert.x Request Metrics over WebSocket"),
+      '1m: ' + @state['1m'],
+      R.br {},
+      '5m: ' + @state['5m'],
+      R.br {},
+      '15m: ' + @state['15m'],
+      R.br {},
+      'count: ' + @state.count,
+      R.br {},
+      'mean: ' + @state.mean, ''
+
+MouseDragWidget = React.createClass
+  mixins: [ReactRxMixin]
+  getInitialState: -> mouseDragResetEv
+  componentWillMount: ->
+    @manageSubcription(@props.events.subscribe(@setState.bind(@)))
+  render: ->
+    R.div {},
+      (R.h2 {}, "Mouse Drag Demo"),
+      "start pos: #{@state.startpos.x} #{@state.startpos.y}",
+      R.br {},
+      "target: #{@state.target}",
+      R.br {},
+      "dx: #{@state.dx}",
+      R.br {},
+      "dy: #{@state.dy}", ''
+
+TimerWidget = React.createClass
+  mixins: [ReactRxMixin]
+  getInitialState: ->
+    timer: 0
+    active: false
+
+  render: ->
+    display = if @state.active then @state.timer else 'off'
+    R.div {},
+       (R.h2 {}, "Pausable Timer Demo"),
+       "timer: #{display}", ''
+
+  componentDidMount: ->
+    @manageSubcription(@props.events.subscribe((t) => @setState(timer: t)))
+    @observeDom('click', (_ev) =>
+      @setState(active: !@state.active)
+      @props.timerSwitch.onNext(@state.active))
+
+  ## or instead of `observeDom`: {onClick: @handleClick} +
+  #handleClick: (ev) ->
+    #@setState(active: !@state.active)
+    #@props.timerSwitch.onNext(@state.active)
+
+initUI = (eventSources) ->
+  rootComponent = R.div {},
+    RequestsWidget(events: eventSources.requests),
+    MouseDragWidget(events: eventSources.mouseDrags),
+    TimerWidget(events: eventSources.timer, timerSwitch: eventSources.timerSwitch)
+  React.renderComponent(rootComponent, document.getElementById('app'))
+
+
+################################################################################
+# start your engines
 onload = ->
-  init_ui()
-  app.demos.mouseMove()
-  init_websockets()
-  #log("--")
-  #app.demos.behaviours()
-  #app.demos.replay()
-  #app.demos.backpressures()
-  #app.demos.timer()
-  #app.demos.toAsync()
-#
+  app.uiRoot = initUI(app.events)
+  # start the related demo event sources
+  demos.mouseDrags(app.events)
+  demos.timer(app.events)
+  demos.metricsWebsockets(app.config, app.events)
+  app.events.responseTimes.subscribe(logAllObserver('responseTimes'))
+  log('loaded')
+
 window.addEventListener("load", onload, false)

pkg = require('./package.json')
_ = require('lodash')
React = require('react')
R = React.DOM
Rx = require('rx')
#Rxdom = require('rx-dom') # may not need

log = console.log.bind(console)

{Observable, Observer, Subject} = Rx

window.app = app =
  ns:

     requests: "/streams/meters/requests"
     responseTimes: "/streams/metrics/histograms/responseTimes"
  events:
    # Rx.Subjects to decouple UI from event sources
    timer: new Subject()
    timerSwitch: new Subject()
    mouseDrags: new Subject()
    requests: new Subject()
    responseTimes: new Subject()
  demos: {}
  uiRoot: null

demos = app.demos

################################################################################
# consume mod_metrics data from David's vert.x app over websockets:
wsurl = (path) ->
  host = document.location.hostname
  wsport = "8080" #document.location.port
  "ws://#{host}:#{wsport}#{path}"

demos.metricsWebsockets = (config, eventSources) ->
  url = (endpoint_name) ->
    url0 = config.wsAPI[endpoint_name]
    if url0[0] == "/"
      wsurl(url0)
    else
      url0
  #TODO add reconnector, plus manage subscriptions
  observeWebSocket(url("requests"))
    .sample(500)
    .map((ev) -> JSON.parse(ev.data))
    .map((d) ->
      d = _.mapValues(d, (v) -> v.toFixed(2))
      d.count = Math.round(d.count)
      d)
    .subscribe((e) -> eventSources.requests.onNext(e))
  observeWebSocket(url("responseTimes"))
    .sample(500)
    .subscribe((e) -> eventSources.responseTimes.onNext(e))

################################################################################
# additional Observable event sources demos included in presentation

mouseUp = Observable.fromEvent(document, 'mouseup')
mouseMoves = Observable.fromEvent(document, 'mousemove')
