title: RxJs: How to only maintain the latest value until inner observable complete
body: <p>I'm new to RxJs and having trouble to achieve this in "RxJs way":</p>
  
  <p>An infinite stream <code>a$</code> emits a value <code>a</code> once a while.</p>
  
  <p><code>async()</code> takes the <code>a</code> and performs an async operation.</p>
  
  <p>If <code>a$</code> emits values while <code>async</code> is pending, only keep the latest one <code>al</code>.</p>
  
  <p>After the previous <code>async</code> completes, if there is an <code>al</code>, run <code>async(al)</code>.</p>
  
  <p>And so on.</p>
  
  <pre><code>a$:----a1----------a2----a3-----------------------a4-----------
         async(a1):------------end                  async(a4):---
                               async(a3):-----end
  </code></pre>
  
  <p>Here is what I came up with, a bit nasty:</p>
  
  <pre><code>var asyncIdle$ = new Rx.BehaviorSubject()
  var asyncRunning$ = new Rx.Subject()
  var async$ = asyncIdle$
  
  function async (val) {
    async$ = asyncRunning$
    // do something with val
    console.log(val + ' handling')
    setTimeout(() =&gt; {
      console.log(val + ' complete')
      async$.next()
      async$ = asyncIdle$
    }, 2000)
  }
  
  // simulate a$
  var a$ = Rx.Observable.fromEvent(document, 'click')
  .mapTo(1)
  .scan((acc, curr) =&gt; acc + curr)
  .do(val =&gt; console.log('got ' + val))
  
  
  a$.debounce(() =&gt; async$)
  .subscribe(val =&gt; {
    async(val)
  })
  </code></pre>