title: How to dynamically update an RX Observable?
body: <p>(Working in RxKotlin and RxJava, but using metacode for simplicity)</p>
  
  <p>Many <code>Reactive Extensions</code> guides begin by creating an <code>Observable</code> from already available data.  From <a href="https://gist.github.com/staltz/868e7e9bc2a7b8c1f754" rel="nofollow noreferrer">The introduction to Reactive Programming you've been missing</a>, it's created from a single string</p>
  
  <pre><code>var soureStream= Rx.Observable.just('https://api.github.com/users');
  </code></pre>
  
  <p>Similarly, from the frontpage of <a href="https://github.com/ReactiveX/RxKotlin" rel="nofollow noreferrer">RxKotlin</a>, from a populated list</p>
  
  <pre><code>val list = listOf(1,2,3,4,5)
  list.toObservable()     
  </code></pre>
  
  <p>Now consider a simple filter that yields an <code>outStream</code>,  </p>
  
  <pre><code>var outStream = sourceStream.filter({x &gt; 3})
  </code></pre>
  
  <p>In both guides the source events are declared <em>apriori</em>.  Which means the timeline of events has some form</p>
  
  <pre><code>source: ----1,2,3,4,5-------
  out:    --------------4,5---
  </code></pre>
  
  <p>How can I modify <code>sourceStream</code> to become more of a pipeline?  In other words, no input data is available during <code>sourceStream</code> creation?  When a source event becomes available, it is immediately processed by out:</p>
  
  <pre><code>source: ---1--2--3-4---5-------
  out:    ------------4---5-------
  </code></pre>
  
  <p>I expected to find an <code>Observable.add()</code> for dynamic updates</p>
  
  <pre><code>var sourceStream = Observable.empty()
  var outStream = sourceStream.filter({x&gt;3})
  
  //print each element as its added 
  sourceStream .subscribe({println(it)})
  outStream.subscribe({println(it)})
  
  for i in range(5):
      sourceStream.add(i)
  </code></pre>
  
  <p>Is this possible?</p>